数据目录与血缘追踪：元数据驱动的数据资产管理

{
  "asset": "warehouse.ods.orders",
  "field": "order_id",
  "urn": "urn:li:dataset:(urn:li:dataPlatform:postgres,warehouse.ods.orders,PROD)#order_id",
  "technical": {
    "data_type": "varchar(64)",
    "nullable": false,
    "partitioned": false
  },
  "business": {
    "description": "订单唯一编号，由订单中心生成",
    "owner": "order-squad",
    "business_glossary": "term://order/order_id"
  },
  "governance": {
    "classification": "PII-LOW",
    "tier": "critical",
    "tags": ["order", "core"]
  },
  "operational": {
    "freshness_minutes": 5,
    "quality_grade": "A"
  }
}

# datahub_recipe.yaml
source:
  type: postgres
  config:
    host_port: dw-host:5432
    database: warehouse
    username: datahub
    password: ${DATAHUB_DB_PASSWORD}
    schema_pattern:
      allow: ["ods", "dim", "dws"]
    profiling:
      enabled: true

# dbt 模型的上游血缘通过 manifest.json 自动构建
source2:
  type: dbt
  config:
    manifest_path: ./target/manifest.json
    catalog_path: ./target/catalog.json
    node_names_pattern:
      allow:
        - "model.*"

datahub ingest -c datahub_recipe.yaml
# 或者以增量模式调度
datahub ingest -c datahub_recipe.yaml --dry-run

{
  "source": {
    "type": "mysql",
    "serviceName": "warehouse_mysql",
    "serviceConnection": {
      "config": {
        "type": "Mysql",
        "hostPort": "dw-host:3306",
        "username": "openmetadata",
        "password": "${OPENMETADATA_DB_PASSWORD}",
        "databaseSchema": "ods"
      }
    },
    "sourceConfig": {
      "config": {
        "type": "DatabaseMetadata",
        "includeTags": true,
        "sampleDataCount": 100
      }
    }
  },
  "sink": {
    "type": "metadata-rest",
    "config": {}
  }
}

metadata ingest -c openmetadata_mysql.json
metadata profile -c openmetadata_mysql.json   # 字段画像

# amundsen_ingest.py
from amundsen.databuilder.extractor.mysql_extractor import MySQLTableExtractor
from amundsen.databuilder.transformer.table_metadata_transformer import TableMetadataTransformer
from amundsen.databuilder.publisher.neo4j_csv_publisher import Neo4jCsvPublisher

extractor = MySQLTableExtractor(
    connector=MySQLTableExtractor.CONNECTION_STRING_KEY,
    database="warehouse",
    table_names=["ods_orders"],
)
transformer = TableMetadataTransformer()
publisher = Neo4jCsvPublisher(
    node_files=["table_metadata.csv"],
    relationship_files=["table_table.csv"],
    neo4j_endpoint="http://neo4j:7474",
)

# classifier.py
import re

PII_RULES = [
    (r"(?i)(id_card|ssn|passport)", "PII-HIGH"),
    (r"(?i)(phone|mobile|email)", "PII-MEDIUM"),
    (r"(?i)(name|address|ip)", "PII-LOW"),
]

def classify_column(col: str, sample_values=None) -> str:
    for pattern, level in PII_RULES:
        if re.search(pattern, col):
            return level
    if sample_values:
        # 简单指纹：邮箱格式命中即 PII-MEDIUM
        if any("@" in v for v in sample_values[:100]):
            return "PII-MEDIUM"
    return "NONE"

print(classify_column("id_card_no"))
print(classify_column("user_email", ["a@x.com", "b@y.com"]))

# lineage_parse.py
from sqllineage.runner import LineageRunner

sql = """
CREATE TABLE dws.daily_order_revenue AS
SELECT o.order_id,
       o.user_id,
       o.amount,
       u.country
FROM ods.orders o
JOIN dim.users u ON o.user_id = u.user_id
WHERE o.status = 'PAID';
"""

runner = LineageRunner(sql)
for col in runner.get_column_lineage():
    print(f"{col.src_columns}  ->  {col.target_column}")

# sqlglot_deps.py
import sqlglot

ast = sqlglot.parse_one(
    """
    WITH paid AS (
      SELECT order_id, amount FROM ods.orders WHERE status = 'PAID'
    )
    SELECT p.order_id, d.channel
    FROM paid p
    JOIN dim.dim_channel d ON p.order_id = d.order_id
    """,
    read="snowflake",
)
print(ast.find_all(sqlglot.exp.Table))

{
  "eventType": "COMPLETE",
  "job": {
    "namespace": "airflow",
    "name": "etl_orders_to_dws"
  },
  "inputs": [
    {"namespace": "postgres.warehouse", "name": "ods.orders"},
    {"namespace": "postgres.warehouse", "name": "dim.users"}
  ],
  "outputs": [
    {"namespace": "snowflake.warehouse", "name": "dws.daily_order_revenue"}
  ],
  "run": {
    "runId": "f3c4d5e6-...",
    "facets": {"sql": {"query": "INSERT OVERWRITE TABLE dws... "}}
  }
}

# mark_data_sources.py
from datahub.emitter.mce_builder import make_dataset_urn
from datahub.emitter.rest_emitter import DatahubRestEmitter

emitter = DatahubRestEmitter("http://datahub-gms:8080")

# 标记表属于 order 数据域
emitter.emit_mcp(
    make_dataset_urn("postgres", "warehouse.ods.orders", "PROD"),
    aspect={
        "type": "globalTags",
        "tags": [{"tag": "domain.order"}],
    },
)

# tag_policy.yaml
tags:
  - name: PII-HIGH
    auto_apply:
      regex: ["(?i)(id_card|ssn|passport)"]
    enforced_actions:
      - column_mask: sha256
      - require_approval: true
  - name: cost_center
    auto_apply:
      source: dataset.owner.cost_center
    enforced_actions:
      - allocate_budget: true

# governance_health.py
def audit_governance(datasets: list) -> list:
    violations = []
    for ds in datasets:
        if not ds.get("owner"):
            violations.append({"asset": ds["name"], "type": "MISSING_OWNER"})
        if ds.get("tier") == "critical" and not ds.get("classification"):
            violations.append({"asset": ds["name"], "type": "MISSING_CLASSIFICATION"})
    return violations

# impact_analysis.py
from collections import deque

class LineageImpact:
    def __init__(self, edges: dict):
        self.downstream = edges  # node -> [targets]

    def impact(self, node: str, max_depth: int = 4) -> list:
        result, queue = [], deque([(node, 0)])
        seen = {node}
        while queue:
            cur, d = queue.popleft()
            for t in self.downstream.get(cur, []):
                if t not in seen:
                    seen.add(t)
                    result.append({"asset": t, "depth": d + 1})
                    if d + 1 < max_depth:
                        queue.append((t, d + 1))
        return result

edges = {
    "ods.orders": ["ods.orders_clean", "dws.order_fact"],
    "ods.orders_clean": ["dws.order_fact", "dws.daily_revenue"],
    "dws.order_fact": ["cube.revenue_cube"],
    "cube.revenue_cube": ["dashboards.revenue"],
}
print(LineageImpact(edges).impact("ods.orders"))

#!/bin/bash
# lineage_gate.sh
IMPACT_COUNT=$(python impact_analysis.py --source ods.orders --max-depth 4 --json | jq 'length')
if [ "$IMPACT_COUNT" -gt 30 ]; then
  echo "Impact ${IMPACT_COUNT} assets - requires data owner approval"
  exit 1
fi