feat(openmetadata): populate Data Insights dashboard

Assign owners to all 177 tables and descriptions + tier tags to the 18 business tables, then trigger SearchIndexing + DataInsights apps so the platform Data Insights page shows real totals, description coverage and tier distribution instead of empty plots.
2026-06-27 15:17:02 +00:00
parent 28821e8b04
commit b8a5b10bc4
3 changed files with 157 additions and 0 deletions
@@ -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)
@@ -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]))
@@ -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))