数据血缘与溯源：用Delta Lake Transaction Log画一张数据家族树

数据血缘，说白了就是搞清楚「这份数据从哪来、经过谁的手、变成了什么样」。我刚开始做数据治理那会儿，最头疼的就是业务方跑来问：「这个报表里的销售额，到底是从哪个原始表算出来的？」——没人答得上来。

Delta Lake 的 Transaction Log 其实天生就是干这个的。每次写入、更新、删除，它都记了一笔账。我们只需要把这些「账本」串起来，就能还原出完整的数据流转路径。

Transaction Log 里到底藏了什么？

每个 Delta 表都有一个 _delta_log/ 目录，里面是一堆 JSON 文件。文件名就是版本号：00000000000000000000.json、00000000000000000001.json……

我打开一个典型的日志文件给你看：

{
  "protocol": {"minReaderVersion": 1, "minWriterVersion": 2},
  "metaData": {
    "id": "123e4567-e89b-12d3-a456-426614174000",
    "format": {"provider": "parquet", "options": {}},
    "schemaString": "...",
    "partitionColumns": ["dt"],
    "createdTime": 1700000000000
  },
  "add": {
    "path": "dt=2024-01-01/part-00000-abc.snappy.parquet",
    "partitionValues": {"dt": "2024-01-01"},
    "size": 2048576,
    "modificationTime": 1700000100000,
    "dataChange": true,
    "tags": {"INSERT_TIME": "1700000100000", "USER": "etl_user"}
  },
  "commitInfo": {
    "timestamp": 1700000100000,
    "userId": "user_001",
    "userName": "zhangsan",
    "operation": "WRITE",
    "operationParameters": {"mode": "Append", "partitionBy": "[\"dt\"]"},
    "readVersion": 0,
    "isolationLevel": "Serializable"
  }
}

关键字段就这几个：add 记录新增的文件，remove 记录删除的文件，commitInfo 告诉你谁在什么时间做了什么操作。嗯，这就是血缘追踪的原始素材。

怎么从日志里提取血缘信息？

我个人习惯分三步走：

1. 解析所有版本：按版本号顺序读取 JSON，提取每个版本的 add 和 remove 列表。
2. 关联操作与文件：通过 commitInfo.operation 判断是写入、更新还是删除，再结合 readVersion 找到上游版本。
3. 构建依赖图：把「表 A 版本 N → 表 B 版本 M」的依赖关系画成有向图。

举个例子，假设我们有三个表：

表名	操作	版本	上游表	上游版本
raw_orders	WRITE	v1	—	—
clean_orders	WRITE	v1	raw_orders	v1
agg_daily_sales	WRITE	v1	clean_orders	v1

你看，血缘关系一目了然。实际项目中，我还会记录每个字段级别的映射关系，比如 raw_orders.total_price → clean_orders.amount。这需要额外解析 SQL 或 ETL 脚本，但 Delta Log 至少给了我们文件级的线索。

构建数据溯源图谱

光有血缘还不够，我们得把它可视化。下面这张图是我用 Delta Log 信息画出来的数据流转路径：

这张图里，每个箭头背后都对应着一次 Delta Lake 事务。你点开任意一个版本的 commitInfo，就能看到是谁、在什么时候、用什么参数执行的这次操作。

实战：用 PySpark 读取 Transaction Log 构建血缘

我写过一个工具脚本，专门用来从 Delta Log 里提取血缘信息。核心逻辑是这样的：

from pyspark.sql import SparkSession
import json
from pathlib import Path

def extract_lineage(delta_table_path: str, max_versions: int = 100):
    """
    从 Delta Lake Transaction Log 提取血缘信息
    """
    spark = SparkSession.builder.getOrCreate()
    log_path = f"{delta_table_path}/_delta_log/"
    
    lineage = []
    
    # 读取所有版本日志
    for version in range(max_versions):
        log_file = f"{log_path}{version:020d}.json"
        try:
            with open(log_file, 'r') as f:
                for line in f:
                    entry = json.loads(line)
                    
                    # 提取 commitInfo
                    if "commitInfo" in entry:
                        commit = entry["commitInfo"]
                        operation = commit.get("operation", "UNKNOWN")
                        read_version = commit.get("readVersion", None)
                        user = commit.get("userName", "unknown")
                        timestamp = commit.get("timestamp", 0)
                        
                        # 提取 add 文件列表
                        added_files = []
                        if "add" in entry:
                            added_files.append(entry["add"]["path"])
                        
                        lineage.append({
                            "table": delta_table_path,
                            "version": version,
                            "operation": operation,
                            "user": user,
                            "timestamp": timestamp,
                            "read_version": read_version,
                            "added_files": added_files
                        })
        except FileNotFoundError:
            break
    
    return lineage

# 使用示例
lineage_info = extract_lineage("/data/delta/clean_orders")
for entry in lineage_info:
    print(f"版本 {entry['version']}: {entry['operation']} by {entry['user']}, "
          f"上游版本 {entry['read_version']}, 新增文件 {entry['added_files']}")

这段代码会输出类似这样的结果：

版本 0: WRITE by etl_user, 上游版本 None, 新增文件 ['dt=2024-01-01/part-00000-abc.snappy.parquet']
版本 1: UPDATE by analyst_wang, 上游版本 0, 新增文件 ['dt=2024-01-01/part-00001-def.snappy.parquet']
版本 2: DELETE by data_governance, 上游版本 1, 新增文件 []

跨表血缘追踪

单表血缘好办，难的是跨表。比如 agg_daily_sales 是从 clean_orders 聚合来的，而 clean_orders 又依赖 raw_orders。怎么把这条链串起来？

我的做法是维护一个「血缘注册表」：

目标表	目标版本	源表	源版本	ETL 任务 ID	执行时间
clean_orders	v1	raw_orders	v1	job_001	2024-01-01 01:00:00
agg_daily_sales	v1	clean_orders	v1	job_002	2024-01-01 02:00:00
agg_daily_sales	v2	clean_orders	v2	job_003	2024-01-02 02:00:00

这个表本身也可以存在 Delta Lake 里。每次 ETL 任务跑完，自动往里面写一条记录。查询的时候，从目标表一路往上追溯，就能画出完整的血缘树。

字段级血缘：更进一步

文件级血缘够用吗？说实话，对于数据治理来说，很多时候我们需要知道「这个字段是从哪个字段来的」。比如 agg_daily_sales.total_revenue 到底对应 clean_orders 的哪个字段？

Delta Log 本身不记录字段映射，但我们可以结合 SQL 解析来做。我常用的方案是：

1. 从 commitInfo 里拿到 ETL 任务的 SQL 语句（如果有的话）。
2. 用 SQL 解析器（比如 sqlparse 或 antlr）提取 SELECT 子句中的字段映射关系。
3. 把映射关系存到另一张血缘明细表里。

举个例子，如果 ETL 的 SQL 是：

INSERT INTO agg_daily_sales
SELECT order_date, SUM(amount) AS total_revenue
FROM clean_orders
GROUP BY order_date

那字段级血缘就是：

• agg_daily_sales.order_date ← clean_orders.order_date
• agg_daily_sales.total_revenue ← clean_orders.amount（经过 SUM 聚合）

嗯，这里要注意：字段级血缘的准确性完全取决于 SQL 解析的质量。我踩过不少坑，比如子查询、CTE、CASE WHEN 这些复杂结构，解析起来特别容易出错。建议先从简单的 SELECT-FROM-WHERE 开始，逐步覆盖更多场景。

血缘数据的存储与查询

血缘数据本身也是数据资产，得好好存着。我个人推荐用 Delta Lake 来存血缘数据——没错，用 Delta Lake 管理 Delta Lake 的血缘，有点套娃的感觉，但确实好用。

表结构大概这样：

CREATE TABLE data_lineage (
  lineage_id STRING,
  source_table STRING,
  source_version BIGINT,
  target_table STRING,
  target_version BIGINT,
  operation_type STRING,
  field_mappings ARRAY<STRUCT<
    source_field: STRING,
    target_field: STRING,
    transform_expr: STRING
  >>,
  etl_job_id STRING,
  run_timestamp TIMESTAMP,
  created_at TIMESTAMP
)
USING DELTA
PARTITIONED BY (target_table)

查询的时候，用 Spark SQL 就能直接追溯：

-- 查询 agg_daily_sales 的所有上游
SELECT source_table, source_version, operation_type
FROM data_lineage
WHERE target_table = 'agg_daily_sales'
ORDER BY run_timestamp DESC

-- 反向查询：哪些表用了 raw_orders 的数据
SELECT target_table, target_version
FROM data_lineage
WHERE source_table = 'raw_orders'