3. 元数据采集：Hive、Spark、Flink 的血缘捕获

元数据采集，说白了就是给数据湖仓装上一双「眼睛」。没有这双眼睛，血缘追踪就是空谈。我这些年做数据治理，见过太多团队上来就搞血缘分析，结果发现连最基本的表结构都没采全——那真是白费功夫。

这一章，咱们就聚焦三个核心引擎：Hive、Spark、Flink。它们代表了离线批处理、内存计算和实时流处理三种典型场景。我会把每个场景的采集原理、代码实现和避坑经验都掰开揉碎讲清楚。

3.1 Hive 元数据采集：从 Metastore 说起

Hive 的元数据都藏在 Metastore 里。不管是 MySQL 还是 Derby，表结构、分区信息、存储格式、SerDe 全在那儿。我个人习惯直接用 JDBC 连 Metastore 的底层库，这样最稳。

但注意，千万别在生产环境直接连 Metastore 的读写库。我见过有人这么干，结果一个慢查询把 Hive 的元数据操作全堵死了——那场面，啧啧，整个数仓都动不了。

下面是我常用的 Hive 元数据采集代码片段，用 Python 写的，直接查 Metastore 的 TBLS 和 COLUMNS_V2 表：

import pymysql
from datetime import datetime

def fetch_hive_metadata(host, port, user, password, db='hive_metastore'):
    conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db)
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    
    # 采集所有表的基本信息
    cursor.execute("""
        SELECT 
            t.TBL_ID, t.TBL_NAME, d.NAME AS DB_NAME,
            t.TBL_TYPE, t.CREATE_TIME, t.LAST_ACCESS_TIME,
            p.PKEY_NAME, p.PKEY_TYPE, p.PKEY_COMMENT
        FROM TBLS t
        JOIN DBS d ON t.DB_ID = d.DB_ID
        LEFT JOIN PARTITION_KEYS p ON t.TBL_ID = p.TBL_ID
        WHERE t.TBL_NAME NOT LIKE '%tmp%'
    """)
    
    tables = {}
    for row in cursor.fetchall():
        tbl_key = f"{row['DB_NAME']}.{row['TBL_NAME']}"
        if tbl_key not in tables:
            tables[tbl_key] = {
                'tbl_id': row['TBL_ID'],
                'db_name': row['DB_NAME'],
                'tbl_name': row['TBL_NAME'],
                'tbl_type': row['TBL_TYPE'],
                'create_time': datetime.fromtimestamp(row['CREATE_TIME']),
                'partitions': []
            }
        if row['PKEY_NAME']:
            tables[tbl_key]['partitions'].append({
                'name': row['PKEY_NAME'],
                'type': row['PKEY_TYPE']
            })
    
    # 采集字段信息
    cursor.execute("""
        SELECT c.CD_ID, c.COLUMN_NAME, c.TYPE_NAME, c.INTEGER_IDX, c.COMMENT
        FROM COLUMNS_V2 c
        JOIN SDS s ON c.CD_ID = s.CD_ID
        JOIN TBLS t ON s.SD_ID = t.SD_ID
    """)
    
    for row in cursor.fetchall():
        # 这里把字段挂到对应的表上
        pass
    
    cursor.close()
    conn.close()
    return tables

Hive 3.x 之后，Metastore 提供了 NotificationLog 事件流。你可以轮询 NOTIFICATION_LOG 表，拿到 CREATE_TABLE、ALTER_TABLE、DROP_TABLE 等事件。这样就能做到准实时同步。

3.2 Spark 作业监听：SparkListener 接口实战

Spark 的血缘采集，核心是 SparkListener 接口。你想想看，Spark 作业跑的时候，Driver 端会不断抛出各种事件——SparkListenerSQLExecutionStart、SparkListenerSQLExecutionEnd，里面就藏着完整的执行计划。

我最早做这块的时候，傻乎乎地去解析 Spark UI 的 REST API。后来发现官方提供了 SparkSession.listenerManager()，直接注册监听器就行，省事多了。

下面是一个自定义的 Spark 血缘监听器：

import org.apache.spark.sql.execution.ui._
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.command._

class LineageListener extends SparkListener {
  
  override def onOtherEvent(event: SparkListenerEvent): Unit = {
    event match {
      case e: SparkListenerSQLExecutionStart =>
        val plan = e.executionPlan
        // 解析 LogicalPlan，提取输入表和输出表
        val lineage = extractLineage(plan)
        // 写入元数据中心
        LineageStore.save(lineage)
        
      case _ => // 忽略其他事件
    }
  }
  
  private def extractLineage(plan: String): LineageInfo = {
    // 这里用 Catalyst 的解析器反序列化执行计划
    // 实际项目中可以用 Spark 的 QueryExecution 对象
    // 我这里简化了，真实场景要处理 Subquery、CTE、Union 等复杂结构
    LineageInfo(
      inputTables = extractInputTables(plan),
      outputTables = extractOutputTables(plan),
      queryText = plan,
      timestamp = System.currentTimeMillis()
    )
  }
}

// 注册监听器
spark.sparkContext.addSparkListener(new LineageListener())

这里有个关键点：解析执行计划。Spark 的 LogicalPlan 是一棵树，你得遍历它，找到 UnresolvedRelation（输入表）和 InsertIntoTable、SaveIntoDataSourceCommand（输出表）。

我建议你用 SparkSession.sessionState.sqlParser 把 SQL 解析成 LogicalPlan，然后用 Tree Pattern Matching 提取表名。别自己去写正则解析 SQL——那玩意儿太容易出错了，尤其是遇到子查询和 CTE 的时候。

3.3 Flink 血缘捕获：实时流的挑战

Flink 的血缘捕获比 Hive 和 Spark 都麻烦。为什么？因为 Flink 是流处理，作业可能 7x24 小时不停，你没法等作业跑完再采集。而且 Flink 的 Table API 和 DataStream API 混用时，血缘信息会散落在不同的地方。

我做过一个实时数仓项目，Flink 作业有 200 多个算子，光靠人工梳理血缘根本不可能。后来我用了 Flink 的 Catalog 机制 和 Checkpoint 元数据 来捕获。

先说 Catalog。Flink 1.11 之后，TableEnvironment 会维护一个 CatalogManager，里面注册了所有的表。你可以通过 catalogManager.getCatalog() 拿到表结构：

import org.apache.flink.table.api._
import org.apache.flink.table.catalog._

def captureFlinkCatalog(tEnv: TableEnvironment): List[CatalogTable] = {
  val catalogManager = tEnv.getCatalogManager
  val catalogs = catalogManager.listCatalogs()
  
  catalogs.flatMap { catalogName =>
    val catalog = catalogManager.getCatalog(catalogName).get
    val databases = catalog.listDatabases()
    
    databases.flatMap { db =>
      val tables = catalog.listTables(db)
      tables.map { tableName =>
        val table = catalog.getTable(new ObjectPath(db, tableName))
        CatalogTableInfo(
          catalog = catalogName,
          database = db,
          tableName = tableName,
          schema = table.getUnresolvedSchema,
          comment = table.getComment
        )
      }
    }
  }
}

但 Catalog 只能拿到静态的表结构，拿不到作业内部的算子血缘。这时候就得靠 Checkpoint 元数据 了。Flink 的 Checkpoint 里保存了 OperatorID 和 State 的映射关系。你可以解析 Checkpoint 的元数据文件，反推出数据流路径。

下面是一个简化版的 Flink 算子血缘提取思路：

// 通过 REST API 获取 JobGraph
val jobManagerUrl = "http://flink-jobmanager:8081"
val jobs = requests.get(s"$jobManagerUrl/jobs").text
// 解析 JSON，拿到每个 job 的 vertices 和 edges

// 每个 vertex 包含 OperatorID、算子名称、输入/输出类型
// 每个 edge 包含 source 和 target 的 OperatorID

// 构建血缘图：
// 1. 遍历 vertices，建立算子节点
// 2. 遍历 edges，建立算子间的数据流边
// 3. 结合 Catalog 信息，把 Source/Sink 算子映射到物理表

3.4 三种采集方式的对比与选型

说了这么多，我把三种方式的核心差异整理成了一张表，方便你选型时参考：

维度	Hive 元数据采集	Spark 作业监听	Flink 血缘捕获
采集方式	JDBC 直连 Metastore + NotificationLog 事件	SparkListener 接口 + 执行计划解析	Catalog API + Checkpoint 元数据 + REST API
实时性	分钟级（轮询）或秒级（事件）	作业执行时实时触发	分钟级（依赖 Checkpoint 周期）
采集内容	表结构、分区、存储格式、DDL 变更	SQL 文本、输入/输出表、字段级血缘	算子链、Source/Sink 表、状态映射
复杂度	低（成熟稳定）	中（需解析 LogicalPlan）	高（流作业状态复杂）
适用场景	离线数仓、Hive 表治理	Spark SQL、ETL 作业血缘	实时数仓、Flink SQL 作业

选型的时候，我建议你遵循一个原则：先采能采的，再优化采得好的。Hive 的元数据采集最简单，可以最先落地；Spark 的监听器能给你作业级的血缘，适合中期建设；Flink 的血缘捕获最复杂，但一旦做成了，实时数仓的治理就稳了。