id: hf-dataset-drift
namespace: company.team
description: |
Hugging Face datasets change shape between revisions and training or eval
jobs quietly retrain on data nobody reviewed. A keyed webhook accepts the
revision stats for the baseline and the candidate snapshot, a shared
workspace materializes them, and a Python step compares row counts,
column sets, and label shares against configurable tolerances. Drift past
tolerance posts a Slack alert naming every drifted feature and the largest
delta, an unreadable payload escalates through its own branch, and the
verdict, the drifted feature list, and the largest delta are published as
outputs.
triggers:
- id: post_revision_stats
type: io.kestra.plugin.core.trigger.Webhook
description: The dataset monitor POSTs baseline and candidate revision stats
here after each dataset revision is inspected.
key: "{{ secret('HF_DRIFT_WEBHOOK_KEY') }}"
- id: nightly_drift_sweep
type: io.kestra.plugin.core.trigger.Schedule
description: Fallback net, a nightly drift sweep over the pinned revision pair,
shipped disabled until an operator switches it on.
cron: "0 4 * * *"
disabled: true
inputs:
- id: revision_stats
type: JSON
defaults:
dataset: acme/customer-intents
baseline:
revision: 2026-09-01
rows: 48000
columns: [ text, label, source ]
labels:
billing: 0.42
support: 0.36
churn: 0.22
candidate:
revision: 2026-10-14
rows: 61000
columns: [ text, label, source ]
labels:
billing: 0.25
support: 0.38
churn: 0.21
spam: 0.16
description: Revision stats the comparison uses when no webhook body is posted,
shaped as a dataset name with baseline and candidate revision objects.
- id: max_label_delta_pct
type: FLOAT
defaults: 10.0
description: Largest label share shift in percentage points a revision pair may
show before that label counts as drifted.
- id: max_row_delta_pct
type: FLOAT
defaults: 20.0
description: Largest relative row count change in percent a revision pair may
show before the row count counts as drifted.
tasks:
# --- 1. revision comparison ---
- id: drift_workspace
type: io.kestra.plugin.core.flow.WorkingDirectory
description: One shared directory so the shell step that writes the revision
stats and the Python step that compares them read the same file.
tasks:
- id: write_revision_stats
type: io.kestra.plugin.scripts.shell.Commands
description: Materialize the posted revision stats, or the input default on
manual runs, into the shared workspace with a quoted heredoc and size
the file for the log.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
entryPoint: []
containerImage: python:3.12-slim
commands:
- |
set -u
cat > revision_stats.json <<'STATS_EOF'
{{ (trigger.body ?? inputs.revision_stats) | toJson }}
STATS_EOF
echo "wrote $(wc -c < revision_stats.json) bytes of revision stats to the shared workspace"
- id: compare_revisions
type: io.kestra.plugin.scripts.python.Script
description: Compare row counts, column sets, and label shares across the two
revisions, flag every feature past its tolerance, and emit the drift
verdict through the outputs protocol.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
env:
MAX_LABEL_DELTA_PCT: "{{ inputs.max_label_delta_pct }}"
MAX_ROW_DELTA_PCT: "{{ inputs.max_row_delta_pct }}"
script: |
import json
import os
STATS_FILE = "revision_stats.json"
def env_float(name, default):
try:
return float(os.environ.get(name, default))
except (TypeError, ValueError):
return float(default)
MAX_LABEL_DELTA_PCT = env_float("MAX_LABEL_DELTA_PCT", 10.0)
MAX_ROW_DELTA_PCT = env_float("MAX_ROW_DELTA_PCT", 20.0)
def load_stats(path):
try:
if not os.path.exists(path) or os.path.getsize(path) == 0:
return None, "revision stats file is missing or empty"
with open(path, "r", encoding="utf-8") as handle:
document = json.load(handle)
document = json.loads(document) if isinstance(document, str) else document
if not isinstance(document, dict):
return None, "revision stats is not a JSON object"
return document, ""
except ValueError:
return None, "revision stats is not valid JSON"
def as_int(value):
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
def share_map(value):
shares = {}
if isinstance(value, dict):
for label, raw in value.items():
try:
shares[str(label)] = float(raw)
except (TypeError, ValueError):
continue
return shares
stats, problem = load_stats(STATS_FILE)
dataset = "unknown-dataset"
baseline_revision = ""
candidate_revision = ""
rows_delta_pct = 0.0
column_changes = 0
added_columns = []
removed_columns = []
label_deltas = {}
drift_features = []
max_delta = 0.0
error = ""
verdict = "stable"
if stats is None:
verdict = "error"
error = problem
else:
dataset = str(stats.get("dataset") or "unknown-dataset")
baseline = stats.get("baseline")
candidate = stats.get("candidate")
if not isinstance(baseline, dict) or not isinstance(candidate, dict):
verdict = "error"
error = "revision stats must carry a baseline and a candidate revision"
else:
baseline_revision = str(baseline.get("revision") or "unknown-revision")
candidate_revision = str(candidate.get("revision") or "unknown-revision")
base_rows = as_int(baseline.get("rows"))
cand_rows = as_int(candidate.get("rows"))
if base_rows > 0:
rows_delta_pct = round(abs(cand_rows - base_rows) * 100.0 / base_rows, 2)
else:
rows_delta_pct = 100.0 if cand_rows else 0.0
base_columns = [str(c) for c in (baseline.get("columns") or []) if str(c)]
cand_columns = [str(c) for c in (candidate.get("columns") or []) if str(c)]
added_columns = sorted(set(cand_columns) - set(base_columns))
removed_columns = sorted(set(base_columns) - set(cand_columns))
order_changed = (not added_columns and not removed_columns
and cand_columns != base_columns)
column_changes = len(added_columns) + len(removed_columns) + (1 if order_changed else 0)
for column in added_columns:
drift_features.append("column " + column + " added")
for column in removed_columns:
drift_features.append("column " + column + " removed")
if order_changed:
drift_features.append("column order changed")
if rows_delta_pct > MAX_ROW_DELTA_PCT:
direction = "up" if cand_rows >= base_rows else "down"
drift_features.append("row count moved " + format(rows_delta_pct, ".2f")
+ " pct " + direction)
base_shares = share_map(baseline.get("labels"))
cand_shares = share_map(candidate.get("labels"))
for label in sorted(set(base_shares) | set(cand_shares)):
before = base_shares.get(label, 0.0)
after = cand_shares.get(label, 0.0)
delta = round(abs(after - before) * 100.0, 2)
label_deltas[label] = delta
if label not in base_shares:
if delta > MAX_LABEL_DELTA_PCT:
drift_features.append("label " + label + " appeared with "
+ format(delta, ".2f") + " points")
elif label not in cand_shares:
if delta > MAX_LABEL_DELTA_PCT:
drift_features.append("label " + label + " disappeared from "
+ format(delta, ".2f") + " points")
elif delta > MAX_LABEL_DELTA_PCT:
drift_features.append("label " + label + " moved "
+ format(delta, ".2f") + " points")
deltas = [rows_delta_pct] + list(label_deltas.values())
max_delta = round(max(deltas) if deltas else 0.0, 2)
verdict = "drift_detected" if drift_features else "stable"
payload = {
"verdict": verdict,
"dataset": dataset,
"baseline_revision": baseline_revision,
"candidate_revision": candidate_revision,
"rows_delta_pct": rows_delta_pct,
"column_changes": column_changes,
"added_columns": added_columns,
"removed_columns": removed_columns,
"label_deltas": label_deltas,
"drift_features": drift_features,
"max_delta": max_delta,
"drift": bool(drift_features),
"error": error,
}
print("dataset=%s features=%d max_delta=%.2f verdict=%s" % (
dataset, len(drift_features), max_delta, verdict))
print("::" + json.dumps({"outputs": payload}) + "::")
# --- 2. drift verdict ---
- id: check_drift
type: io.kestra.plugin.core.flow.If
description: Escalate to Slack when any feature moved past its tolerance, and
verify the error branch instead of trusting a parse failure.
condition: "{{ outputs.compare_revisions.vars.verdict == 'drift_detected' }}"
then:
- id: notify_drift
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Tell the data channel which features drifted between the revisions
and the largest delta the comparison observed.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"text": "DATASET DRIFT: {{ outputs.compare_revisions.vars.dataset }} between {{ outputs.compare_revisions.vars.baseline_revision }} and {{ outputs.compare_revisions.vars.candidate_revision }} - {{ outputs.compare_revisions.vars.drift_features | length }} feature(s) past tolerance ({{ outputs.compare_revisions.vars.drift_features | join('; ') }}), max delta {{ outputs.compare_revisions.vars.max_delta }}. Execution {{ execution.id }}."
}
else:
- id: check_drift_error
type: io.kestra.plugin.core.flow.If
description: Surface revision stats that could not be compared as their own
branch so a broken feed never reads as a stable dataset.
condition: "{{ outputs.compare_revisions.vars.verdict == 'error' }}"
then:
- id: alert_unreadable_stats
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Tell the data channel the revision stats could not be compared and
carry the load problem named by the grader.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"text": "DATASET drift grade unreadable for {{ outputs.compare_revisions.vars.dataset }} - {{ outputs.compare_revisions.vars.error }}. Execution {{ execution.id }}."
}
else:
- id: log_stable
type: io.kestra.plugin.core.log.Log
description: Record the stable comparison so the execution history doubles as
evidence the revisions matched.
message: "stable - {{ outputs.compare_revisions.vars.dataset }} {{
outputs.compare_revisions.vars.baseline_revision }} to {{
outputs.compare_revisions.vars.candidate_revision }} stayed within
tolerance, max delta {{ outputs.compare_revisions.vars.max_delta
}}."
errors:
- id: notify_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Escalate any failure to the ops channel with the execution url.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{"text": "FAILED: {{ flow.id }} {{ execution.id }}"}
outputs:
- id: verdict
type: STRING
description: stable when every feature stayed within tolerance, drift_detected
when one moved past it, error when the stats could not be compared.
value: "{{ outputs.compare_revisions.vars.verdict }}"
- id: drift_features
type: ARRAY
description: Human readable features that drifted, for example the billing label
share shift or an added column.
value: "{{ outputs.compare_revisions.vars.drift_features | toJson }}"
- id: max_delta
type: FLOAT
description: Largest observed delta across the row count and every label share,
for example 27.08.
value: "{{ outputs.compare_revisions.vars.max_delta }}"
- id: rows_delta_pct
type: FLOAT
description: Relative row count change between the two revisions in percent, for
example 27.08.
value: "{{ outputs.compare_revisions.vars.rows_delta_pct }}"
- id: column_changes
type: INT
description: Number of column additions, removals, or reorderings observed
between the two revisions.
value: "{{ outputs.compare_revisions.vars.column_changes }}"
