diff --git a/deploy/openmetadata/enrich/di_enrich.py b/deploy/openmetadata/enrich/di_enrich.py new file mode 100644 index 0000000..4e157da --- /dev/null +++ b/deploy/openmetadata/enrich/di_enrich.py @@ -0,0 +1,104 @@ +"""Populate OpenMetadata Data Insights: assign owners (all tables), +descriptions + tier tags (business tables), then the DataInsights app can +produce a meaningful Data Assets dashboard (coverage > 0).""" +import json, os, time, urllib.request, urllib.parse, urllib.error + +OM = "http://openmetadata-server:8585/api" +H = {"Authorization": "Bearer " + os.environ["OM_TOKEN"]} +MO = "f0fc4958-5a46-4088-b5d0-9be137dae95d" +BART = "e2849c82-4d42-4e85-b8c7-802181e688be" + + +def req(method, path, body=None, ct="application/json"): + data = json.dumps(body).encode() if body is not None else None + h = dict(H) + if data is not None: + h["Content-Type"] = ct + 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()[:200] + + +def patch(eid, ops): + return req("PATCH", "/v1/tables/" + eid, ops, "application/json-patch+json") + + +# ---- map all tables: fqn -> id -------------------------------------------- +st, d = req("GET", "/v1/search/query?q=*&index=table_search_index&size=400") +fqn2id = {} +for h in d["hits"]["hits"]: + fqn2id[h["_source"]["fullyQualifiedName"]] = h["_id"] +print("tables found:", len(fqn2id)) + +# ---- 1) owner on ALL tables (ownership coverage) -------------------------- +owned = 0 +for fqn, eid in fqn2id.items(): + st, r = patch(eid, [{"op": "add", "path": "/owners", + "value": [{"id": MO, "type": "user"}]}]) + if st == 200: + owned += 1 + elif st == 400: # owners already present -> replace + st2, _ = patch(eid, [{"op": "replace", "path": "/owners", + "value": [{"id": MO, "type": "user"}]}]) + owned += 1 if st2 == 200 else 0 +print("owners set on:", owned) + +# ---- 2) descriptions + tier on business tables ---------------------------- +BIZ = { + "atc_postgres.postgres.public.sales_orders": ( + "Operational sales orders - source of truth in PostgreSQL. Continuously " + "change-captured by Debezium into Kafka and curated into the lakehouse.", "Tier.Tier1"), + "atc_mysql.default.hr.employee_events": ( + "HR employee lifecycle events (hire/leave/transfer) in MySQL. CDC source, " + "masked for PII in the curated layer.", "Tier.Tier1"), + "atc_mongodb.supplychain.supplychain.events": ( + "Supply-chain events stream in MongoDB. CDC source feeding Kafka and S3.", "Tier.Tier1"), + "atc_trino.iceberg.curated_masked.sales_orders_masked": ( + "Curated, PII-masked sales orders (Iceberg on S3). Governed analytics layer.", "Tier.Tier1"), + "atc_trino.iceberg.curated_masked.employee_events_masked": ( + "Curated, PII-masked HR employee events (Iceberg on S3).", "Tier.Tier1"), + "atc_trino.iceberg.hadoop.historical_sales": ( + "Historical sales offloaded from Hadoop/HDFS into Iceberg for long-term analytics.", "Tier.Tier2"), + "atc_trino.iceberg.hadoop.historical_sales_hdfs": ( + "Historical sales materialized from HDFS via the hadoop_to_trino pipeline.", "Tier.Tier2"), + "atc_trino.cassandra_telemetry.telemetry.device_metrics": ( + "Device telemetry time-series in Cassandra, queryable via Trino.", "Tier.Tier2"), + 'atc_trino.kafka.default."postgres-sales.public.sales_orders"': ( + "Debezium CDC stream of sales_orders (Kafka), queryable via Trino.", "Tier.Tier2"), + 'atc_trino.kafka.default."mongodb-supplychain.supplychain.events"': ( + "Debezium CDC stream of supply-chain events (Kafka), queryable via Trino.", "Tier.Tier2"), + "atc_trino.postgres_sales.public.sales_orders": ( + "Trino read-through view of the PostgreSQL sales_orders source.", "Tier.Tier2"), + "atc_trino.mysql_hr.hr.employee_events": ( + "Trino read-through view of the MySQL employee_events source.", "Tier.Tier2"), + "atc_trino.mongodb_supplychain.supplychain.events": ( + "Trino read-through view of the MongoDB supply-chain events source.", "Tier.Tier2"), + "atc_trino.postgres_sales.monitor.cdc_operations": ( + "Operational view: CDC operation counts per table.", "Tier.Tier3"), + "atc_trino.postgres_sales.monitor.cdc_recent_events": ( + "Operational view: most recent CDC events.", "Tier.Tier3"), + "atc_trino.postgres_sales.monitor.debezium_connectors": ( + "Operational view: Debezium connector status.", "Tier.Tier3"), + "atc_trino.postgres_sales.monitor.kafka_topics": ( + "Operational view: Kafka topic inventory and lag.", "Tier.Tier3"), + "atc_trino.postgres_sales.monitor.spark_applications": ( + "Operational view: Spark application run history.", "Tier.Tier3"), +} +done = 0 +for fqn, (desc, tier) in BIZ.items(): + eid = fqn2id.get(fqn) + if not eid: + print(" ! not found:", fqn); continue + st, t = req("GET", "/v1/tables/" + eid + "?fields=tags,owners") + ops = [{"op": "add", "path": "/description", "value": desc}] + tags = [tg for tg in (t.get("tags") or []) if not tg.get("tagFQN", "").startswith("Tier.")] + tags.append({"tagFQN": tier, "source": "Classification", + "labelType": "Manual", "state": "Confirmed"}) + ops.append({"op": "add" if "tags" not in t else "replace", "path": "/tags", "value": tags}) + st2, r2 = patch(eid, ops) + print(" biz %-55s desc+%s -> %s" % (fqn.split(".")[-1][:50], tier, st2)) + done += 1 if st2 == 200 else 0 +print("business enriched:", done) diff --git a/deploy/openmetadata/enrich/di_trigger.py b/deploy/openmetadata/enrich/di_trigger.py new file mode 100644 index 0000000..420ab89 --- /dev/null +++ b/deploy/openmetadata/enrich/di_trigger.py @@ -0,0 +1,22 @@ +import json, os, time, urllib.request, urllib.error + +OM = "http://openmetadata-server:8585/api" +H = {"Authorization": "Bearer " + os.environ["OM_TOKEN"]} + + +def req(method, path, body=None): + data = json.dumps(body).encode() if body is not None else None + h = dict(H) + if data is not None: + h["Content-Type"] = "application/json" + 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()[:250] + + +for app in ["SearchIndexingApplication", "DataInsightsApplication"]: + st, r = req("POST", "/v1/apps/trigger/" + app, {}) + print("trigger %-30s -> %s %s" % (app, st, str(r)[:120])) diff --git a/deploy/openmetadata/enrich/di_verify.py b/deploy/openmetadata/enrich/di_verify.py new file mode 100644 index 0000000..eb39355 --- /dev/null +++ b/deploy/openmetadata/enrich/di_verify.py @@ -0,0 +1,31 @@ +import json, os, time, urllib.request, urllib.parse, urllib.error + +OM = "http://openmetadata-server:8585/api" +H = {"Authorization": "Bearer " + os.environ["OM_TOKEN"]} +now = int(time.time() * 1000) +start = now - 7 * 86400000 + + +def get(path): + r = urllib.request.Request(OM + path, headers=H) + try: + with urllib.request.urlopen(r, timeout=30) as resp: + return resp.status, json.loads(resp.read().decode()) + except urllib.error.HTTPError as e: + return e.code, e.read().decode()[:160] + + +charts = [ + "total_data_assets", + "percentage_of_data_asset_with_owner", + "percentage_of_data_asset_with_description", + "total_data_assets_by_tier", +] +for c in charts: + p = "/v1/analytics/dataInsights/system/charts/name/%s/data?start=%d&end=%d" % (c, start, now) + st, r = get(p) + summary = r + if isinstance(r, dict): + res = r.get("results") or r.get("data") or [] + summary = "results=%d e.g. %s" % (len(res), json.dumps(res[:2])[:220]) + print("%-45s %s %s" % (c, st, summary))