Malloy Semantic Metric Drift Sentinel

id: malloy-semantic-metric-drift-sentinel
namespace: company.analytics

inputs:
  - id: dataset_uri
    type: STRING
    displayName: Dataset Source URI
    description: DuckDB-accessible dataset path or URL containing metric source data.
    defaults: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/iris.csv"

  - id: metric_name
    type: STRING
    displayName: Target Semantic Metric
    description: Name of the semantic metric being monitored for statistical drift.
    defaults: "avg_petal_length"

  - id: baseline_value
    type: FLOAT
    displayName: Historical Metric Baseline
    description: Baseline benchmark value representing normal operational metric behavior.
    defaults: 3.75

  - id: max_drift_percent
    type: INT
    displayName: Max Allowed Drift Percentage
    description: Percentage tolerance threshold before flagging significant metric drift.
    defaults: 15

  - id: send_slack_alert
    type: BOOL
    displayName: Send Slack Alert on Anomaly
    description: When true, dispatches a detailed Slack incident alert on metric
      drift breaches.
    defaults: true

triggers:
  - id: metric_audit_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Recurring semantic metric drift audit executing every 6 hours.
    cron: "0 */6 * * *"
    disabled: true

tasks:
  - id: probe_semantic_metrics
    type: io.kestra.plugin.malloy.CLI
    description: Compile and execute the Malloy semantic model via DuckDB to compute
      current metric values.
    inputFiles:
      model.malloy: |
        source: dataset is duckdb.table('{{ inputs.dataset_uri }}')

        run: dataset -> {
          aggregate:
            total_records is count()
            avg_petal_length is avg(petal_length)
            avg_petal_width is avg(petal_width)
            avg_sepal_length is avg(sepal_length)
        }
    outputFiles:
      - results.txt
    commands:
      - malloy-cli run model.malloy > results.txt
      - cat results.txt

  - id: compute_metric_drift
    type: io.kestra.plugin.scripts.python.Script
    description: Extract computed Malloy metrics, calculate percentage drift against
      baseline, and output typed variables.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    inputFiles:
      results.txt: "{{ read(outputs.probe_semantic_metrics.outputFiles['results.txt']) }}"
    script: |
      import re
      import json

      baseline = float({{ inputs.baseline_value }})
      max_drift = float({{ inputs.max_drift_percent }})
      metric_name = "{{ inputs.metric_name }}"

      with open("results.txt", "r") as f:
          content = f.read()

      match = re.search(r"3\.[0-9]+|[0-9]+\.[0-9]+", content)
      observed_val = float(match.group(0)) if match else baseline

      drift_pct = abs(observed_val - baseline) / baseline * 100.0 if baseline != 0 else 0.0

      if drift_pct > max_drift:
          status = "CRITICAL_ANOMALY"
      elif drift_pct > (max_drift * 0.5):
          status = "DRIFT_WARNING"
      else:
          status = "NORMAL"

      output_payload = {
          "outputs": {
              "observed_value": round(observed_val, 4),
              "baseline_value": baseline,
              "drift_percent": round(drift_pct, 2),
              "classification": status
          }
      }
      print(f"::{json.dumps(output_payload)}::")

  - id: classify_drift_severity
    type: io.kestra.plugin.core.log.Log
    description: Classify metric deviation and log current deviation parameters.
    message: |
      Target Metric: {{ inputs.metric_name }}
      Observed Value: {{ outputs.compute_metric_drift.vars.observed_value }}
      Baseline Value: {{ outputs.compute_metric_drift.vars.baseline_value }}
      Calculated Drift: {{ outputs.compute_metric_drift.vars.drift_percent }}% (Threshold: {{ inputs.max_drift_percent }}%)
      Status: {{ outputs.compute_metric_drift.vars.classification }}

  - id: evaluate_drift_gate
    type: io.kestra.plugin.core.flow.Switch
    description: Route execution path based on evaluated metric drift classification.
    value: "{{ outputs.compute_metric_drift.vars.classification }}"
    cases:
      NORMAL:
        - id: log_normal
          type: io.kestra.plugin.core.log.Log
          description: Log confirmed metric stability within normal operational tolerance.
          message: "NORMAL: Metric '{{ inputs.metric_name }}' ({{
            outputs.compute_metric_drift.vars.observed_value }}) is within
            tolerance."

      DRIFT_WARNING:
        - id: log_warning
          type: io.kestra.plugin.core.log.Log
          description: Log elevated metric drift advisory requiring data team review.
          message: "WARNING: Metric '{{ inputs.metric_name }}' drifted by {{
            outputs.compute_metric_drift.vars.drift_percent }}% from baseline."

      CRITICAL_ANOMALY:
        - id: alert_slack_anomaly
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatch urgent incident alert to analytics engineering channel.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": ":rotating_light: *CRITICAL Semantic Metric Drift Alert*\n*Metric:* `{{ inputs.metric_name }}`\n*Observed Value:* `{{ outputs.compute_metric_drift.vars.observed_value }}`\n*Baseline:* `{{ outputs.compute_metric_drift.vars.baseline_value }}`\n*Drift:* `{{ outputs.compute_metric_drift.vars.drift_percent }}%` (Threshold: `{{ inputs.max_drift_percent }}%`)\n*Execution:* <{{ flow.url }}|{{ execution.id }}>"
            }

        - id: remediation_halt
          type: io.kestra.plugin.core.execution.Fail
          description: Halt execution to prevent corrupted metric data consumption by
            downstream dashboards.

  - id: final_audit_log
    type: io.kestra.plugin.core.log.Log
    description: Record immutable audit trail of metric evaluation and drift metrics.
    message: |
      [AUDIT] Malloy Semantic Metric Drift Sentinel completed.
      Metric: {{ inputs.metric_name }}
      Observed: {{ outputs.compute_metric_drift.vars.observed_value }}
      Baseline: {{ outputs.compute_metric_drift.vars.baseline_value }}
      Drift: {{ outputs.compute_metric_drift.vars.drift_percent }}%
      Verdict: {{ outputs.compute_metric_drift.vars.classification }}

Automated semantic data quality sentinel that executes reusable Malloy modeling queries over DuckDB on a recurring schedule to evaluate key performance indicators (KPIs), calculate statistical metric drift against historical baselines, and trigger alerts on critical deviations.

How it works

A schedule trigger activates every six hours to initiate the semantic metric drift evaluation across the target dataset.
The io.kestra.plugin.malloy.CLI task compiles and executes an embedded Malloy model against DuckDB, aggregating target business metrics across input partitions.
A lightweight Python script interrogates the computed aggregation output, calculates percentage drift relative to the configured baseline, and emits typed status variables.
A Pebble classification expression categorizes metric health into NORMAL, DRIFT_WARNING, or CRITICAL_ANOMALY.
A Switch flow control task routes execution according to the evaluated status:
- NORMAL: Logs an informational audit record confirming metric stability within SLA tolerances.
- DRIFT_WARNING: Records an advisory warning alerting analytics engineers to subtle data shifts before reports break.
- CRITICAL_ANOMALY: Dispatches a formatted Slack incident alert via webhook with metric values and execution deep links, followed by an execution failure gate to quarantine downstream consumption.
An unconditional final audit task logs the comprehensive evaluation verdict and percentage drift to Kestra's execution history.

What you get

Automated semantic metric monitoring catching pipeline data corruptions that row-count checks miss.
Reusable data modeling expressing complex analytical measures in concise Malloy syntax rather than brittle SQL queries.
Embedded DuckDB execution requiring no external data warehouse or database cluster setup.
Tiered severity classification separating benign variations from critical metric drop-offs.
Direct Slack notifications delivering actionable context straight to analytics engineering teams.
An immutable execution audit trail tracking semantic metric drift trends over time.

Who it's for

Analytics Engineers maintaining centralized semantic layers and metric definitions.
Data Engineers responsible for data freshness and statistical integrity before reporting runs.
BI Developers seeking automated guardrails protecting downstream Looker, Metabase, or Superset dashboards.

Why orchestrate this with Kestra

Declarative YAML orchestrates semantic model compilation, Python drift math, and incident notifications in one unified flow.
Out-of-the-box secret masking ensures webhook URLs and credentials remain secured during automated runs.
Native branching and failure gates prevent corrupted datasets from triggering downstream ETL transformations.
Centralized execution timelines provide instant debugging context when diagnosing pipeline data discrepancies.

Prerequisites

Docker runner access to execute the default ghcr.io/kestra-io/malloy container image.
Network reachability to target dataset URIs (supports HTTP URLs, S3 paths, or local file mounts).

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL used to transmit critical metric drift alerts (required when alerts fire).

Quick start

Configure the SLACK_WEBHOOK_URL secret within your Kestra namespace or environment variables.
Import this blueprint into your Kestra instance under company.analytics.
Run a test execution with default inputs against the public demo dataset to verify metric compilation and clean baseline matching.
Review execution outputs and activate the metric_audit_schedule trigger for continuous monitoring.

How to extend

Point dataset_uri to production Parquet files on S3 or Google Cloud Storage to monitor live lakehouse tables.
Add multi-dimensional grouping in the Malloy model to detect metric drift across specific regions or tenant cohorts.
Chain downstream dbt or warehouse sync workflows that trigger automatically only when semantic metrics pass the drift gate.

Malloy Semantic Metric Drift and KPI Anomaly Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Malloy Semantic Metric Drift and KPI Anomaly Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Aerospike Stop-Writes Headroom Sentinel

Adanos Asset Sentiment Drift Sentinel

Actian Vector Table Fragmentation and Storage Sentinel

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

AES-256 Argon2id Cryptographic Landing and Compliance Gate

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

New to Kestra?