Malloy Semantic Row-Count Parity Gate for Kestra

id: malloy-semantic-row-count-parity-gate
namespace: company.team
description: |
  Run the same reusable Malloy semantic model against a source dataset and
  its replicated destination copy on a schedule, compare the resulting row
  counts, classify the result as PARITY_OK, PARITY_DRIFT_DETECTED, or
  PROBE_FAILED, log clean passes, and alert Slack the moment a replication
  or ETL pipeline leaves the two sides out of sync.

inputs:
  - id: source_table_uri
    type: STRING
    defaults: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/iris.csv"
    displayName: "Source Dataset URI"
    description: "DuckDB-queryable URI (CSV/Parquet path or URL) for the
      source-of-truth dataset. Defaults to plugin-malloy's own official example
      dataset so a first run succeeds without configuration."

  - id: destination_table_uri
    type: STRING
    defaults: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/iris.csv"
    displayName: "Destination Dataset URI"
    description: "DuckDB-queryable URI for the replicated or warehouse-loaded copy
      of the same dataset. Defaults to the same demo dataset as
      source_table_uri; point this at your real downstream copy to audit actual
      replication parity."

  - id: max_row_count_drift
    type: INT
    defaults: 0
    displayName: "Max Row Count Drift"
    description: "Maximum allowed absolute difference between source and destination
      row counts before the pipeline is classified as drifted. The default of 0
      enforces exact parity."

  - id: malloy_container_image
    type: STRING
    defaults: "ghcr.io/kestra-io/malloy"
    displayName: "Malloy Runner Container Image"
    description: "Docker image the malloy-cli binary runs in; this matches
      plugin-malloy's own default image."

  - id: send_slack_alert
    type: STRING
    defaults: "true"
    displayName: "Send Slack Alert"
    description: "When set to true, dispatch a Slack notification when row-count
      parity drifts beyond tolerance; when false, only log the breach."

triggers:
  - id: parity_audit_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Audit source-to-destination row-count parity every 6 hours."
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      source_table_uri: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/iris.csv"
      destination_table_uri: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/iris.csv"
      max_row_count_drift: 0
      malloy_container_image: "ghcr.io/kestra-io/malloy"
      send_slack_alert: "true"

tasks:
  - id: query_source_row_count
    type: io.kestra.plugin.malloy.CLI
    description: "Run a minimal Malloy semantic model against the source dataset,
      counting rows, and emit the count through Kestra's structured-output
      marker."
    containerImage: "{{ inputs.malloy_container_image }}"
    inputFiles:
      model.malloy: |
        source: dataset is duckdb.table('{{ inputs.source_table_uri }}')

        run: dataset -> {
          aggregate: row_count is count()
        }
    outputFiles:
      - results.json
    commands:
      - malloy-cli run model.malloy > results.json
      - cat results.json
      - ROW_COUNT=$(jq '.[0].row_count' results.json)
      - printf '::{"outputs":{"row_count":%s}}::\n' "$ROW_COUNT"
    allowFailure: true

  - id: query_destination_row_count
    type: io.kestra.plugin.malloy.CLI
    description: "Run the identical Malloy semantic model against the destination
      dataset, counting rows, and emit the count through Kestra's
      structured-output marker."
    containerImage: "{{ inputs.malloy_container_image }}"
    inputFiles:
      model.malloy: |
        source: dataset is duckdb.table('{{ inputs.destination_table_uri }}')

        run: dataset -> {
          aggregate: row_count is count()
        }
    outputFiles:
      - results.json
    commands:
      - malloy-cli run model.malloy > results.json
      - cat results.json
      - ROW_COUNT=$(jq '.[0].row_count' results.json)
      - printf '::{"outputs":{"row_count":%s}}::\n' "$ROW_COUNT"
    allowFailure: true

  - id: classify_parity_status
    type: io.kestra.plugin.core.debug.Return
    description: "Classify as PROBE_FAILED when either side's row count was never
      emitted (Malloy compile error, missing jq, or container failure),
      PARITY_OK when the two counts are within max_row_count_drift of each
      other, otherwise PARITY_DRIFT_DETECTED."
    format: "{{ (outputs.query_source_row_count.vars.row_count is not defined or
      outputs.query_destination_row_count.vars.row_count is not defined) ?
      'PROBE_FAILED' : (((outputs.query_source_row_count.vars.row_count -
      outputs.query_destination_row_count.vars.row_count <=
      inputs.max_row_count_drift) and
      (outputs.query_destination_row_count.vars.row_count -
      outputs.query_source_row_count.vars.row_count <=
      inputs.max_row_count_drift)) ? 'PARITY_OK' : 'PARITY_DRIFT_DETECTED') }}"

  - id: route_by_parity_status
    type: io.kestra.plugin.core.flow.Switch
    description: "Branch on the classification: page Slack when row counts drift
      beyond tolerance or the probe fails, otherwise log a healthy audit line."
    value: "{{ outputs.classify_parity_status.value }}"
    cases:
      "PARITY_DRIFT_DETECTED":
        - id: evaluate_alert_toggle
          type: io.kestra.plugin.core.flow.If
          description: "Only call the Slack webhook when send_slack_alert is true;
            otherwise log the breach without paging."
          condition: "{{ inputs.send_slack_alert == 'true' }}"
          then:
            - id: dispatch_parity_drift_alert
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: "Page the data platform channel with both row counts and the
                configured tolerance."
              url: "{{ secret('SLACK_WEBHOOK') }}"
              payload: |
                {
                  "text": "Malloy row-count parity drift: source has {{ outputs.query_source_row_count.vars.row_count }} rows, destination has {{ outputs.query_destination_row_count.vars.row_count }} rows, outside the allowed drift of {{ inputs.max_row_count_drift }}. Source: {{ inputs.source_table_uri }}. Destination: {{ inputs.destination_table_uri }}. Execution {{ execution.id }}."
                }
          else:
            - id: log_alert_suppressed
              type: io.kestra.plugin.core.log.Log
              description: "Record the drift even when Slack paging is disabled, so the
                execution history still shows it."
              message: "Row-count parity drift detected (source {{
                outputs.query_source_row_count.vars.row_count }}, destination {{
                outputs.query_destination_row_count.vars.row_count }}) but Slack
                alert suppressed by send_slack_alert=false."
      "PARITY_OK":
        - id: log_parity_healthy
          type: io.kestra.plugin.core.log.Log
          description: "Record a healthy audit result when both row counts are within
            tolerance."
          message: "Row-count parity healthy: source {{
            outputs.query_source_row_count.vars.row_count }} rows, destination
            {{ outputs.query_destination_row_count.vars.row_count }} rows (max
            drift {{ inputs.max_row_count_drift }})."
      "PROBE_FAILED":
        - id: alert_probe_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Page the channel when either Malloy query could not run or emit a
            row count, since parity is now unknown."
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Malloy row-count parity sentinel could not complete for source {{ inputs.source_table_uri }} / destination {{ inputs.destination_table_uri }}. Check that both URIs are reachable by duckdb.table(), that malloy-cli compiled the model, and that jq is available in {{ inputs.malloy_container_image }}. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: "Fallback branch; reached only if classify_parity_status produces a
          value outside the three known cases."
        message: "Malloy row-count parity sentinel produced an unclassified status: {{
          outputs.classify_parity_status.value }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: "Final post-audit status line recorded on every run regardless of branch."
    message: "Row-count parity audit completed for source {{ inputs.source_table_uri
      }} vs destination {{ inputs.destination_table_uri }}. Status: {{
      outputs.classify_parity_status.value }}. Execution ID: {{ execution.id
      }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Flow-level error handler: page the channel when the sentinel flow
      itself fails outright."
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "CRITICAL: Malloy row-count parity gate failed during audit of source {{ inputs.source_table_uri }} vs destination {{ inputs.destination_table_uri }}. Execution ID: {{ execution.id }}."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_parity_status.value }}"

An ETL or replication pipeline that silently drops rows, duplicates them, or stalls partway through usually shows up first as a row-count mismatch between the source system and its downstream copy, and usually nobody notices until a report looks wrong. This blueprint uses plugin-malloy's CLI task to run one reusable Malloy semantic model, the same aggregate: row_count is count() definition, against both sides of a pipeline, so the parity check is expressed once instead of as two hand-written, slowly-diverging SQL queries.

How it works

query_source_row_count (io.kestra.plugin.malloy.CLI) writes a small Malloy model that opens source_table_uri via duckdb.table(...) and runs a single count() aggregate, redirects the result to results.json, and emits the parsed count through Kestra's ::{"outputs":...}:: structured-output marker so it lands in vars. allowFailure: true lets the flow continue instead of aborting on a compile error or unreachable URI.
query_destination_row_count runs the identical model shape against destination_table_uri.
classify_parity_status (io.kestra.plugin.core.debug.Return) checks whether both vars.row_count values are defined. If either is missing, the classification is PROBE_FAILED. If both are present and within max_row_count_drift of each other in both directions, the classification is PARITY_OK; otherwise PARITY_DRIFT_DETECTED.
route_by_parity_status (io.kestra.plugin.core.flow.Switch) branches on that string. PARITY_DRIFT_DETECTED runs evaluate_alert_toggle (io.kestra.plugin.core.flow.If), which pages Slack via dispatch_parity_drift_alert only when send_slack_alert is true, logging a suppressed-alert line otherwise. PARITY_OK logs a clean audit line. PROBE_FAILED pages Slack directly since parity is now unknown. The defaults branch logs a fallback line that is unreachable under normal operation.
log_audit_summary records a status line on every run regardless of branch, so execution history reads as a continuous parity trend.
The flow-level errors block pages Slack separately if the sentinel flow itself fails outright.
The parity_audit_schedule trigger polls every 6 hours with explicit input overrides, shipped disabled: true until pointed at a real source and destination.

What you get

A scheduled row-count parity check expressed as one reusable Malloy model instead of two independently maintained SQL queries.
A three-way status output (PARITY_OK / PARITY_DRIFT_DETECTED / PROBE_FAILED) other flows can gate on before trusting a downstream report or dataset.
A Slack page naming both row counts, both URIs, and the configured tolerance on breach or probe failure.
A separate failure channel for the sentinel flow itself, plus a standing per-run audit log line.

Who it's for

Data engineers running ETL or replication pipelines who need an early signal that the destination silently diverged from the source.
Analytics engineering teams standardizing on Malloy as a semantic layer who want the same model reused for monitoring, not just for dashboards.
Platform teams building Day 2 data-governance checks that compare two systems without hand-duplicating query logic per backend.

Why orchestrate this with Kestra

A one-off SELECT COUNT(*) run by hand on each side proves parity for one instant and leaves no record, and it has to be rewritten per database dialect. This flow keeps every audit in Kestra's execution history, turns the result into a typed status output other flows can depend on, and gives you an If/Switch branch to grow into real remediation (re-triggering the replication job, opening a ticket) without rewriting the comparison logic.

Prerequisites

Both source_table_uri and destination_table_uri reachable by DuckDB's duckdb.table() function from inside the Malloy runner container (local file paths mounted into the container, or an HTTP(S) URL as in the default).
jq available inside malloy_container_image to parse results.json; this blueprint has not verified whether the default ghcr.io/kestra-io/malloy image ships jq, see Pitfalls.
A Slack incoming webhook for the drift, probe-failure, and flow-failure alerts.

Quick start

Add the SLACK_WEBHOOK secret.
Run the flow once with the shipped defaults, both URIs point at the same public demo dataset, and confirm status reads PARITY_OK.
Replace destination_table_uri with your real replicated dataset to audit actual parity.
Enable parity_audit_schedule.

How to extend

Extend the Malloy model with additional aggregates (sums, distinct counts) beyond row_count to catch value-level drift, not just row-count drift; parse the additional fields from results.json the same way.
Fan this out across every table in a replication pipeline with a Loop task, reusing the same model shape with a different table URI per iteration.
Push status into a KV store to alert only on state transitions instead of every run.
If your installed malloy-cli version's JSON output does not match the .[0].row_count shape this blueprint assumes, adjust the jq filter to match your actual output structure before relying on this in production.

Secrets

SLACK_WEBHOOK: Slack incoming webhook used for the drift, probe-failure, and flow-failure alerts.
OSS note: if your Kestra instance reads secrets from environment variables instead of a secrets backend, set SECRET_SLACK_WEBHOOK as base64, for example echo -n "<your-webhook-url>" | base64.

Inputs

source_table_uri (STRING, default the plugin's own iris.csv demo dataset): source-of-truth dataset URI.
destination_table_uri (STRING, default same demo dataset): replicated/destination dataset URI.
max_row_count_drift (INT, default 0): maximum allowed absolute row-count difference before classifying as drifted.
malloy_container_image (STRING, default ghcr.io/kestra-io/malloy): container image running malloy-cli.
send_slack_alert (STRING, default "true"): when "true", page Slack on drift; when "false", only log it.

Outputs

{{ outputs.query_source_row_count.vars.row_count }}: parsed row count from the source dataset.
{{ outputs.query_destination_row_count.vars.row_count }}: parsed row count from the destination dataset.
{{ outputs.query_source_row_count.outputFiles['results.json'] }}: internal storage URI of the raw source query result.
{{ outputs.classify_parity_status.value }}: classification string, one of PARITY_OK, PARITY_DRIFT_DETECTED, PROBE_FAILED.
{{ outputs.status }}: flow-level alias of the same classification.

Pitfalls

This blueprint assumes malloy-cli run model.malloy writes a JSON array of row objects (for example [{"row_count": 150}]), inferred from plugin-malloy's own official example naming its redirected output results.json and from Malloy's query syntax, where row_count is count() names the output field directly. This has not been executed and confirmed against a running malloy-cli; verify the actual shape against your installed version before trusting it in production, and adjust the jq '.[0].row_count' filter if it differs.
jq must be present in malloy_container_image; if the default ghcr.io/kestra-io/malloy image lacks it, add a beforeCommands step to install it, or replace the jq line with a different parser.
Row-count parity alone does not catch row-level corruption, duplicate rows that cancel out, or column-level drift; it only catches count mismatches. Extend the model's aggregates for deeper coverage.
Both tasks write to the same filename, results.json, inside their own isolated working directories; they do not share state or overwrite each other, but do not assume a single shared results.json across both tasks if you refactor this into one task.
duckdb.table() needs outbound network access from the container when pointed at an HTTP(S) URL, and a correctly mounted path when pointed at a local file; a reachability failure surfaces as PROBE_FAILED, not PARITY_DRIFT_DETECTED.

Local Container Setup

This flow needs no separate database container: both Malloy tasks pull and run ghcr.io/kestra-io/malloy directly via plugin-malloy's default Docker task runner, and DuckDB executes in-process against the given URIs. To confirm the image works locally before running the flow, execute:

docker run --rm ghcr.io/kestra-io/malloy malloy-cli --help

If Kestra's Worker is running inside a Docker network and you point either URI at a container-hosted file server, connect that container to Kestra's network:

docker network connect YOUR_KESTRA_NETWORK

(Run docker network ls to verify your active Kestra network name.)

Malloy Semantic Row-Count Parity Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Quick start

How to extend

Secrets

Inputs

Outputs

Pitfalls

Local Container Setup

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Malloy Semantic Row-Count Parity Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Quick start

How to extend

Secrets

Inputs

Outputs

Pitfalls

Local Container Setup

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Read out an A/B test with a sample ratio check and a guardrail metric

AI Data Quality Anomaly Advisor and SQL Remediation Generator

Adanos Asset Sentiment Drift Sentinel

AES-256 Argon2id Cryptographic Landing and Compliance Gate

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Apache ActiveMQ Artemis Dead Letter Address (DLA) Message Accumulation Sentinel

New to Kestra?