From e8ff760cd1bd9f715411bd52079eef598dbd627f Mon Sep 17 00:00:00 2001 From: mo Date: Sat, 27 Jun 2026 13:30:08 +0200 Subject: [PATCH] feat(openmetadata): backfill Airflow task detail + dataset lineage OM 1.13 cannot deserialize Airflow 3.0.6 serialized DAGs, so pipelines had no tasks. af_tasks.py parses serialized_dag and PATCHes task lists; af_lineage.py adds generator->table and mask_to_curated/hadoop_to_trino lineage edges. Note: OM MySQL datadir lives on the 448MB /opt partition which had filled up (103MB binlog). Disabled binary logging + shrank redo log capacity in the OM docker-compose to restore headroom. --- deploy/openmetadata/ingest/af_lineage.py | 85 ++++++++++++++++++++++ deploy/openmetadata/ingest/af_tasks.py | 92 ++++++++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 deploy/openmetadata/ingest/af_lineage.py create mode 100644 deploy/openmetadata/ingest/af_tasks.py diff --git a/deploy/openmetadata/ingest/af_lineage.py b/deploy/openmetadata/ingest/af_lineage.py new file mode 100644 index 0000000..0a5ec45 --- /dev/null +++ b/deploy/openmetadata/ingest/af_lineage.py @@ -0,0 +1,85 @@ +"""Add dataset lineage for the Airflow DAGs into OpenMetadata. + +- Generator DAGs produce a source table -> pipeline:table edge +- ETL DAGs transform source -> target -> table:table edge carrying the pipeline +""" +import json +import os +import urllib.error +import urllib.parse +import urllib.request + +OM = "http://openmetadata-server:8585/api" +TOKEN = os.environ["OM_TOKEN"] +HDRS = {"Authorization": "Bearer " + TOKEN} + + +def req(method, path, body=None): + data = json.dumps(body).encode() if body is not None else None + h = dict(HDRS) + if data is not None: + h["Content-Type"] = "application/json" + last = "?" + for _ in range(3): + r = urllib.request.Request(OM + path, data=data, headers=h, method=method) + try: + with urllib.request.urlopen(r, timeout=60) as resp: + return resp.status, json.loads(resp.read().decode() or "{}") + except urllib.error.HTTPError as e: + return e.code, e.read().decode() + except Exception as e: # noqa: BLE001 - timeouts / transient + last = repr(e) + return 0, last + + +def get_id(kind, fqn): + base = "/v1/pipelines/name/" if kind == "pipeline" else "/v1/tables/name/" + st, d = req("GET", base + urllib.parse.quote(fqn, safe="")) + if st == 200: + return d["id"] + print(" ! cannot resolve", kind, fqn, "->", st) + return None + + +def edge(from_kind, from_fqn, to_fqn, pipeline_fqn=None): + fid = get_id(from_kind, from_fqn) + tid = get_id("table", to_fqn) + if not fid or not tid: + return + body = { + "edge": { + "fromEntity": {"id": fid, "type": from_kind}, + "toEntity": {"id": tid, "type": "table"}, + } + } + if pipeline_fqn: + pid = get_id("pipeline", pipeline_fqn) + if pid: + body["edge"]["lineageDetails"] = { + "pipeline": {"id": pid, "type": "pipeline"} + } + st, res = req("PUT", "/v1/lineage", body) + label = (pipeline_fqn or from_fqn).split(".")[-1] + print(" %-22s %s -> %s" % (label, from_fqn.split(".")[-1], to_fqn.split(".")[-1]), + "OK" if st in (200, 201) else ("ERR %s %s" % (st, res)), flush=True) + + +PG = "atc_postgres.postgres.public.sales_orders" +MY = "atc_mysql.default.hr.employee_events" +MO = "atc_mongodb.supplychain.supplychain.events" + +print("generators:") +edge("pipeline", "atc_airflow.gen_postgres", PG) +edge("pipeline", "atc_airflow.gen_mysql", MY) +edge("pipeline", "atc_airflow.gen_mongodb", MO) +edge("pipeline", "atc_airflow.generate_data_all_databases", PG) +edge("pipeline", "atc_airflow.generate_data_all_databases", MY) +edge("pipeline", "atc_airflow.generate_data_all_databases", MO) + +print("ETL (table -> table via pipeline):") +edge("table", PG, "atc_trino.iceberg.curated_masked.sales_orders_masked", + "atc_airflow.mask_to_curated") +edge("table", MY, "atc_trino.iceberg.curated_masked.employee_events_masked", + "atc_airflow.mask_to_curated") +edge("pipeline", "atc_airflow.hadoop_to_trino", + "atc_trino.iceberg.hadoop.historical_sales_hdfs") diff --git a/deploy/openmetadata/ingest/af_tasks.py b/deploy/openmetadata/ingest/af_tasks.py new file mode 100644 index 0000000..9644a80 --- /dev/null +++ b/deploy/openmetadata/ingest/af_tasks.py @@ -0,0 +1,92 @@ +"""Backfill task-level detail into OpenMetadata Airflow pipelines. + +OM 1.13's Airflow connector cannot deserialize Airflow 3.0.6 serialized DAGs, so +pipelines land with zero tasks. We parse the serialized_dag table ourselves and +PATCH each pipeline with its tasks + intra-DAG dependencies (downstreamTasks). +""" +import json +import os +import sqlite3 +import urllib.error +import urllib.request +import zlib + +DB = "/" + "/".join(["tmp", "airflow.db"]) +OM = "http://openmetadata-server:8585/api" +TOKEN = os.environ["OM_TOKEN"] +SERVICE = "atc_airflow" + +HDRS = {"Authorization": "Bearer " + TOKEN} + + +def req(method, path, body=None, ctype="application/json"): + data = json.dumps(body).encode() if body is not None else None + h = dict(HDRS) + if data is not None: + h["Content-Type"] = ctype + r = urllib.request.Request(OM + path, data=data, headers=h, method=method) + try: + with urllib.request.urlopen(r, timeout=30) as resp: + return resp.status, json.loads(resp.read().decode() or "{}") + except urllib.error.HTTPError as e: + return e.code, e.read().decode() + + +def latest_serialized(): + c = sqlite3.connect(DB) + rows = c.execute( + "SELECT s.dag_id, s.data, s.data_compressed FROM serialized_dag s " + "JOIN (SELECT dag_id, MAX(id) mid FROM serialized_dag GROUP BY dag_id) m " + "ON s.id = m.mid" + ).fetchall() + out = {} + for dag_id, data, comp in rows: + if data is None and comp is not None: + data = zlib.decompress(comp).decode() + d = json.loads(data) if isinstance(data, str) else data + out[dag_id] = d.get("dag", d) + return out + + +def build_tasks(dag): + tasks = [] + for t in dag.get("tasks", []): + tt = t.get("__var", t) if isinstance(t, dict) else t + tid = tt.get("task_id") + if not tid: + continue + tasks.append( + { + "name": tid, + "displayName": tid, + "taskType": tt.get("task_type") or tt.get("_task_type") or "Task", + "downstreamTasks": list(tt.get("downstream_task_ids") or []), + } + ) + return tasks + + +def main(): + dags = latest_serialized() + for dag_id, dag in sorted(dags.items()): + tasks = build_tasks(dag) + fqn = SERVICE + "." + dag_id + st, pipe = req("GET", "/v1/pipelines/name/" + fqn) + if st != 200: + print("SKIP", dag_id, "get-failed", st, pipe) + continue + pid = pipe["id"] + op = "replace" if "tasks" in pipe else "add" + patch = [{"op": op, "path": "/tasks", "value": tasks}] + st2, res = req( + "PATCH", "/v1/pipelines/" + pid, patch, "application/json-patch+json" + ) + edges = sum(len(t["downstreamTasks"]) for t in tasks) + print( + "%-30s tasks=%-2d edges=%-2d -> %s" + % (dag_id, len(tasks), edges, "OK" if st2 == 200 else ("ERR %s %s" % (st2, res))) + ) + + +if __name__ == "__main__": + main()