第09章 数据血缘¶

# 导入 DataHub REST 发射器，用于向 DataHub 服务端推送元数据
from datahub.emitter.rest_emitter import DatahubRestEmitter
# 导入元数据变更提案包装器，封装元数据变更请求
from datahub.emitter.mcp import MetadataChangeProposalWrapper
# 导入各类元数据 Schema 定义类
from datahub.metadata.schema_classes import (
    DatasetPropertiesClass,       # 数据集属性（名称、描述等）
    SchemaMetadataClass,          # Schema 元数据（字段定义）
    SchemaFieldClass,             # 单个字段定义
    StringTypeClass,              # 字符串类型
    NumberTypeClass,              # 数值类型
    UpstreamLineageClass,         # 上游血缘关系
    UpstreamClass,                # 上游数据集
    DatasetLineageTypeClass,      # 血缘类型（转换、复制等）
)

# 创建 DataHub Emitter
emitter = DatahubRestEmitter(
    gms_server="http://localhost:8080"
)

# 定义数据集 URN
dataset_urn = "urn:li:dataset:(urn:li:dataPlatform:hive,sales_data,PROD)"

# 发送数据集属性
dataset_properties = MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=DatasetPropertiesClass(
        name="Sales Data",
        description="Daily sales data",
        customProperties={"team": "data_engineering"}
    )
)
emitter.emit(dataset_properties)

# 发送 Schema 元数据，定义数据集的字段结构信息
schema_metadata = MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=SchemaMetadataClass(
        schemaName="sales_schema",         # Schema 名称
        platform="urn:li:dataPlatform:hive",  # 数据平台标识
        hash="",
        platformSchema=SchemaMetadataClass.PlatformSchemaClass(),
        fields=[
            # 定义 sale_id 字段：BIGINT 数值类型
            SchemaFieldClass(
                fieldPath="sale_id",
                type=NumberTypeClass(),
                nativeDataType="BIGINT",
                description="Sale ID"
            ),
            # 定义 product_id 字段：BIGINT 数值类型
            SchemaFieldClass(
                fieldPath="product_id",
                type=NumberTypeClass(),
                nativeDataType="BIGINT",
                description="Product ID"
            ),
        ]
    )
)
emitter.emit(schema_metadata)  # 将 Schema 信息推送到 DataHub

# 发送血缘关系（上游依赖），描述数据集之间的转换关系
# 上游数据集：MySQL 中的 raw_orders 表
upstream_urn = "urn:li:dataset:(urn:li:dataPlatform:mysql,raw_orders,PROD)"
lineage = MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=UpstreamLineageClass(
        upstreams=[
            UpstreamClass(
                dataset=upstream_urn,
                # TRANSFORMED 表示数据经过转换处理（ETL）
                type=DatasetLineageTypeClass.TRANSFORMED
            )
        ]
    )
)
emitter.emit(lineage)  # 将血缘关系推送到 DataHub

print(f"DataHub metadata emitted for: {dataset_urn}")

# Amundsen 使用示例（通过 REST API 调用）
import requests

AMUNDSEN_SEARCH_URL = "http://localhost:5001/search"
AMUNDSEN_METADATA_URL = "http://localhost:5002/table"

# 搜索数据资产
response = requests.get(
    f"{AMUNDSEN_SEARCH_URL}",
    params={"query": "sales", "page_index": 0}
)
search_results = response.json()

# 遍历搜索结果，打印每个数据表的基本信息
for table in search_results.get("results", []):
    print(f"Table: {table.get('name')}")          # 表名
    print(f"Schema: {table.get('schema')}")        # 所属 Schema
    print(f"Database: {table.get('database')}")    # 所属数据库
    print(f"Description: {table.get('description')}")  # 描述信息
    print("---")

# 获取表的详细元数据（包含列信息、统计数据等）
table_key = "hive://gold.test_schema/sales"
metadata_resp = requests.get(
    f"{AMUNDSEN_METADATA_URL}",
    params={"key": table_key}
)
table_detail = metadata_resp.json()
# 提取并打印表的所有列名列表
print(f"Columns: {[col['name'] for col in table_detail.get('columns', [])]}")

from datetime import datetime

class MetadataManager:
    """元数据管理类"""

    def __init__(self):
        self.metadata = {}

    def extract_metadata(self, df):
        """从 DataFrame 中自动提取元数据信息"""
        # 构建表级元数据：表名和行数
        metadata = {
            'table_name': 'sales',
            'row_count': len(df),
            'columns': []
        }

        # 遍历每一列，收集列级元数据
        for column in df.columns:
            column_info = {
                'name': column,                    # 列名
                'type': str(df[column].dtype),      # 数据类型
                'null_count': df[column].isnull().sum(),    # 空值数量
                'unique_count': df[column].nunique(),       # 唯一值数量
                # 对于数值列，记录最小值和最大值
                'min_value': df[column].min() if df[column].dtype != 'object' else None,
                'max_value': df[column].max() if df[column].dtype != 'object' else None
            }
            metadata['columns'].append(column_info)

        return metadata

    def track_lineage(self, source_table, target_table, transformation):
        """追踪数据血缘关系，记录数据从源表到目标表的转换路径"""
        lineage = {
            'source': source_table,              # 源表（上游数据）
            'target': target_table,              # 目标表（下游数据）
            'transformation': transformation,    # 转换逻辑描述
            'timestamp': datetime.now().isoformat()  # 记录时间戳
        }

        return lineage

    def get_metadata(self, table_name):
        """获取元数据"""
        return self.metadata.get(table_name)

    def update_metadata(self, table_name, metadata):
        """更新元数据"""
        self.metadata[table_name] = metadata