ATC/atc-agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(masking): mask_to_curated DAG + OpenMetadata lineage + versioned OM artifacts

Browse Source 
- deploy/airflow/mask_to_curated_dag.py: Trino-SQL masking (hash/redact/generalize)
  of PII from postgres/mysql sources into iceberg.curated_masked.* (verified:
  14k+11k masked rows, all PII masked).
- OM source->masked lineage edges created; curated tables cataloged.
- Version OM compose + ingestion configs + om_api helper under deploy/openmetadata.
This commit is contained in:
mo
2026-06-27 03:13:16 +02:00
parent 944635ffe1
commit c3cacc141d
11 changed files with 987 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/mask_pii_setup.py
+33 -20
View File
@@ -17,6 +17,13 @@ DB_HOST = os.getenv("SRC_DB_HOST", "10.0.21.51")
DB_USER = os.getenv("SRC_DB_USER", "mo")
DB_PASS = os.getenv("SRC_DB_PASSWORD", "Dell2026!")
N = int(os.getenv("PII_BACKFILL_ROWS", "4000"))
# asc: backfill lowest ids (so OpenMetadata's first-rows sample sees PII);
# desc: newest rows; both: do both ends.
ORDER = os.getenv("PII_BACKFILL_ORDER", "both").lower()
def _orders():
return (["ASC", "DESC"] if ORDER == "both" else [ORDER.upper()])
FIRST = ["Sophie", "Liam", "Emma", "Noah", "Julia", "Lucas", "Mila", "Daan", "Anna", "Sem",
"Eva", "Finn", "Tess", "Bram", "Lotte", "Max", "Sara", "Thijs", "Nina", "Ruben"]
@@ -70,16 +77,19 @@ def do_postgres():
("customer_ip", "text"), ("billing_iban", "text"), ("shipping_address", "text")]:
cur.execute(f"ALTER TABLE public.sales_orders ADD COLUMN IF NOT EXISTS {col} {typ}")
print(" columns ensured")
cur.execute("SELECT order_id FROM public.sales_orders ORDER BY order_id DESC LIMIT %s", (N,))
ids = [r[0] for r in cur.fetchall()]
rows = []
for oid in ids:
nm = full_name()
rows.append((nm, email(nm), phone(), ip(), iban(), address(), oid))
cur.executemany(
"UPDATE public.sales_orders SET customer_name=%s, customer_email=%s, customer_phone=%s, "
"customer_ip=%s, billing_iban=%s, shipping_address=%s WHERE order_id=%s", rows)
print(f" backfilled {len(rows)} rows with PII")
total = 0
for direction in _orders():
cur.execute(f"SELECT order_id FROM public.sales_orders ORDER BY order_id {direction} LIMIT %s", (N,))
ids = [r[0] for r in cur.fetchall()]
rows = []
for oid in ids:
nm = full_name()
rows.append((nm, email(nm), phone(), ip(), iban(), address(), oid))
cur.executemany(
"UPDATE public.sales_orders SET customer_name=%s, customer_email=%s, customer_phone=%s, "
"customer_ip=%s, billing_iban=%s, shipping_address=%s WHERE order_id=%s", rows)
total += len(rows)
print(f" backfilled {total} rows with PII ({'+'.join(_orders())})")
cur.close()
conn.close()
@@ -98,16 +108,19 @@ def do_mysql():
if "Duplicate column" not in str(e):
raise
print(" columns ensured")
cur.execute("SELECT event_id FROM employee_events ORDER BY event_id DESC LIMIT %s", (N,))
ids = [r[0] for r in cur.fetchall()]
rows = []
for eid in ids:
nm = full_name()
rows.append((nm, email(nm), phone(), national_id(), address(), dob(), eid))
cur.executemany(
"UPDATE employee_events SET employee_name=%s, employee_email=%s, employee_phone=%s, "
"national_id=%s, home_address=%s, date_of_birth=%s WHERE event_id=%s", rows)
print(f" backfilled {len(rows)} rows with PII")
total = 0
for direction in _orders():
cur.execute(f"SELECT event_id FROM employee_events ORDER BY event_id {direction} LIMIT %s", (N,))
ids = [r[0] for r in cur.fetchall()]
rows = []
for eid in ids:
nm = full_name()
rows.append((nm, email(nm), phone(), national_id(), address(), dob(), eid))
cur.executemany(
"UPDATE employee_events SET employee_name=%s, employee_email=%s, employee_phone=%s, "
"national_id=%s, home_address=%s, date_of_birth=%s WHERE event_id=%s", rows)
total += len(rows)
print(f" backfilled {total} rows with PII ({'+'.join(_orders())})")
cur.close()
conn.close()