3、技术元数据采集：Hive/Spark表结构采集、血缘关系解析、数据质量指标采集

技术元数据采集，说白了就是给数据湖仓做「CT扫描」。你得知道表长什么样、数据从哪来、质量过不过关。这一章我重点讲三个核心动作：表结构怎么抓、血缘怎么理、质量指标怎么收。

嗯，先别急着写代码。咱们得先搞清楚采集的目标是什么。我个人习惯把技术元数据采集分成三层：

  • 静态层：表结构、字段类型、分区信息、存储格式。这些基本不变。
  • 动态层：血缘关系、ETL脚本、作业依赖。这些随着调度跑。
  • 评估层：数据质量指标，比如空值率、重复率、波动率。

你想想看，如果这三层都打通了，元数据平台才算真正能用。

核心观点：技术元数据采集不是一次性任务，而是持续增量同步。我见过太多团队一次性全量采集完就再也不管了，结果三个月后元数据全是过时的。

3.1 Hive/Spark表结构采集

先说表结构。Hive和Spark SQL的表结构其实都存储在Hive Metastore里。所以采集入口很明确——直接连Metastore的数据库，或者通过Hive Metastore Client API。

我在项目中遇到过一种情况：团队用Spark写了一大堆临时表，结果Metastore里积累了上万张表，其中一半是没人管的垃圾表。所以采集的时候，我建议加上过滤条件：只采集最近30天有更新的表，或者只采集指定数据库的表。

下面是我常用的采集脚本核心逻辑：

-- 从Hive Metastore直接查询表结构信息
SELECT 
    d.NAME AS database_name,
    t.TBL_NAME AS table_name,
    t.TBL_TYPE AS table_type,  -- MANAGED_TABLE / EXTERNAL_TABLE
    tp.PARAM_VALUE AS comment,
    s.LOCATION AS hdfs_path,
    t.CREATE_TIME,
    t.LAST_ACCESS_TIME
FROM TBLS t
JOIN DBS d ON t.DB_ID = d.DB_ID
LEFT JOIN TABLE_PARAMS tp ON t.TBL_ID = tp.TBL_ID 
    AND tp.PARAM_KEY = 'comment'
LEFT JOIN SDS s ON t.SD_ID = s.SD_ID
WHERE d.NAME IN ('ods', 'dwd', 'dws', 'ads')
  AND t.LAST_ACCESS_TIME > UNIX_TIMESTAMP(DATE_SUB(CURRENT_DATE, 30));

字段级别的信息藏在COLUMNS_V2表里：

SELECT 
    c.COLUMN_NAME,
    c.TYPE_NAME,
    c.INTEGER_IDX AS col_order,
    c.COMMENT
FROM COLUMNS_V2 c
WHERE c.CD_ID = ?;  -- 通过SDS表的CD_ID关联

避坑指南：我曾经直接连Metastore的MySQL库做采集，结果有一次Metastore升级改了表结构，采集程序直接挂了。后来我改用Hive Metastore Thrift Client，虽然慢一点，但兼容性好得多。

采集完的数据，我一般会存成JSON格式，方便后续解析：

{
  "database": "ods",
  "table": "order_detail",
  "type": "MANAGED_TABLE",
  "location": "hdfs://nameservice/user/hive/warehouse/ods.db/order_detail",
  "columns": [
    {"name": "order_id", "type": "string", "comment": "订单ID", "position": 0},
    {"name": "user_id", "type": "bigint", "comment": "用户ID", "position": 1},
    {"name": "amount", "type": "decimal(18,2)", "comment": "金额", "position": 2}
  ],
  "partition_keys": [
    {"name": "dt", "type": "string", "comment": "日期分区"}
  ],
  "create_time": "2024-01-15 10:30:00"
}

3.2 血缘关系解析

血缘关系是技术元数据里最难搞的部分。为什么？因为SQL写法千奇百怪，子查询、CTE、UDF、动态表名，什么花样都有。

我建议分两条线走：

  • 解析SQL脚本：从调度系统拿到ETL脚本，用SQL解析器提取输入表和输出表。
  • 运行时采集：通过Spark/Hive的Listener或Hook，在作业运行时捕获血缘信息。

先说说SQL解析这条路。我个人习惯用Apache Calcite或者开源的sqlparse库。核心逻辑是：

-- 假设有一条ETL SQL
INSERT OVERWRITE TABLE dwd.order_detail_daily PARTITION(dt='2024-01-15')
SELECT 
    o.order_id,
    o.user_id,
    o.amount,
    u.user_name
FROM ods.order_detail o
JOIN ods.user_info u ON o.user_id = u.user_id
WHERE o.dt = '2024-01-15';

-- 解析结果应该是：
-- 输入表: ods.order_detail, ods.user_info
-- 输出表: dwd.order_detail_daily
-- 字段级血缘: dwd.order_detail_daily.order_id <- ods.order_detail.order_id
--             dwd.order_detail_daily.user_id <- ods.order_detail.user_id
--             dwd.order_detail_daily.amount <- ods.order_detail.amount
--             dwd.order_detail_daily.user_name <- ods.user_info.user_name

运行时采集这条路，我推荐用Spark的QueryExecutionListener。在Spark作业执行完后，通过Spark的LogicalPlan可以拿到完整的血缘信息：

// Spark Listener示例（Scala）
class LineageCollectorListener extends QueryExecutionListener {
  override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
    val plan = qe.optimizedPlan
    // 遍历plan，提取Read和Write节点
    val inputTables = plan.collectLeaves().collect {
      case r: HiveTableRelation => r.tableMeta.qualifiedName
      case r: LogicalRelation if r.catalogTable.isDefined => 
        r.catalogTable.get.identifier.unquotedString
    }
    val outputTable = plan match {
      case InsertIntoTable(_, _, _, _, _) => extractOutputTable(plan)
      case _ => ""
    }
    // 写入血缘存储
    saveLineage(inputTables, outputTable, funcName)
  }
}

注意：运行时采集会带来性能开销。我曾经在一个每天跑5000个Spark作业的集群上开启全量血缘采集，结果作业平均耗时增加了8%。后来我改成只采集核心数仓层的作业，把非核心作业的血缘通过SQL解析兜底。

血缘数据存储，我推荐用图数据库或者关系型数据库存成边表：

字段 类型 说明
id bigint 主键
source_table string 源表全名（db.table）
target_table string 目标表全名
job_name string 产生血缘的作业名
field_mapping json 字段级映射关系
execution_time timestamp 作业执行时间
lineage_type string SQL解析 / 运行时采集

3.3 数据质量指标采集

数据质量指标，说白了就是给数据打分。我一般把质量指标分成三类：

  • 完整性：空值率、缺失率
  • 准确性：重复率、值域校验、格式校验
  • 一致性：跨表关联校验、汇总校验

采集方式有两种：一种是跑SQL统计，另一种是用数据质量框架（比如Apache Griffin、Deequ）。我个人更倾向于用SQL，因为灵活，而且不引入额外组件。

下面是我常用的质量指标采集SQL模板：

-- 完整性指标：空值率
SELECT 
    'ods.order_detail' AS table_name,
    'order_id' AS column_name,
    'null_rate' AS metric_name,
    ROUND(SUM(CASE WHEN order_id IS NULL THEN 1 ELSE 0 END) / COUNT(*), 4) AS metric_value,
    CURRENT_TIMESTAMP AS collect_time
FROM ods.order_detail
WHERE dt = '${yesterday}'

UNION ALL

-- 准确性指标：重复率
SELECT 
    'ods.order_detail' AS table_name,
    'order_id' AS column_name,
    'duplicate_rate' AS metric_name,
    ROUND((COUNT(*) - COUNT(DISTINCT order_id)) / COUNT(*), 4) AS metric_value,
    CURRENT_TIMESTAMP AS collect_time
FROM ods.order_detail
WHERE dt = '${yesterday}'

UNION ALL

-- 一致性指标：汇总校验
SELECT 
    'ods.order_detail' AS table_name,
    'amount' AS column_name,
    'sum_check' AS metric_name,
    ROUND(SUM(amount), 2) AS metric_value,
    CURRENT_TIMESTAMP AS collect_time
FROM ods.order_detail
WHERE dt = '${yesterday}';

经验之谈：我曾经遇到一个坑——质量指标采集跑得太频繁，把Hive Metastore打挂了。后来我加了限流和缓存策略：每天凌晨跑一次全量指标，白天只跑增量表的指标。另外，指标结果要存到独立的数据库里，别跟元数据混在一起。

采集结果我习惯存成时序数据格式：

字段 类型 说明
table_name string 表名
column_name string 字段名（表级指标可为空）
metric_name string 指标名（null_rate / duplicate_rate / sum_check）
metric_value double 指标值
threshold double 告警阈值
status string 正常 / 告警 / 严重
collect_time timestamp 采集时间

3.4 技术元数据采集架构总览

最后，我把这三块整合到一起，画一张架构图。你一看就明白整体流程了。

技术元数据采集架构 Hive Metastore Spark作业 / ETL脚本 数据质量SQL / 框架 表结构采集器 血缘解析引擎 质量指标采集器 元数据存储（关系型 / 图数据库 / 时序库） 数据地图 / 表检索 血缘可视化 / 影响分析 质量看板 / 告警 采集频率建议：表结构每天1次，血缘每次作业执行后采集，质量指标每天1次

从架构图能看出来，技术元数据采集不是孤立的三个模块，而是从数据源到存储再到应用的完整链路。表结构是基础，血缘是脉络，质量指标是体检报告。三者缺一不可。

嗯，这一章的内容就到这里。记住一个原则：采集要自动化、增量式、可追溯。别想着一次搞定，元数据管理是个持续优化的过程。

← 上一章 📖 返回目录 下一章 →