4、元数据采集实战：Hive、MySQL、Kafka 元数据采集

好，咱们进入实战环节。前面讲了那么多理论，说白了，元数据采集是主数据血缘分析的「粮草先行」。粮草不到位，后面分析血缘就是空谈。

我个人习惯把元数据采集分成三类：静态元数据（表结构、字段）、操作元数据（谁在什么时候做了什么）、运行时元数据（任务执行计划、消费偏移量）。今天咱们就挑三个最典型的引擎——Hive、MySQL、Kafka，把它们的采集套路讲透。

4.1 Hive 元数据采集

Hive 的元数据存在哪儿？存在 Metastore 里。Metastore 后端通常是 MySQL 或 PostgreSQL。所以采集 Hive 元数据，本质上就是连它的 Metastore 库。

4.1.1 采集哪些表？

Metastore 里核心表就那么几张，我列个清单：

表名	存储内容	采集频率建议
DBS	数据库列表（名称、位置、描述）	每日
TBL	表信息（表名、所属库、创建时间、表类型）	每日
COLUMNS_V2	字段信息（字段名、类型、注释、序号）	每日
PARTITIONS	分区信息（分区键、分区值、分区路径）	每小时（如果分区频繁）
TBLS 中的 LAST_ACCESS_TIME	表最后访问时间（用于判断冷热数据）	每日

4.1.2 采集脚本示例

下面是一个简单的 Python 采集脚本，连 Metastore 的 MySQL 库：

import pymysql
import json
from datetime import datetime

def fetch_hive_metadata(host, port, user, password, database='hive_metastore'):
    conn = pymysql.connect(host=host, port=port, user=user, password=password, database=database)
    cursor = conn.cursor()

    # 采集数据库列表
    cursor.execute("SELECT DB_ID, NAME, DB_LOCATION_URI FROM DBS")
    dbs = cursor.fetchall()

    # 采集表信息（关联 DBS 和 TBLS）
    cursor.execute("""
        SELECT d.NAME AS db_name, t.TBL_NAME, t.TBL_TYPE, t.CREATE_TIME, t.LAST_ACCESS_TIME
        FROM TBLS t
        JOIN DBS d ON t.DB_ID = d.DB_ID
    """)
    tables = cursor.fetchall()

    # 采集字段信息（关联 TBLS、SDS、COLUMNS_V2）
    cursor.execute("""
        SELECT d.NAME AS db_name, t.TBL_NAME, c.COLUMN_NAME, c.TYPE_NAME, c.INTEGER_IDX, c.COMMENT
        FROM TBLS t
        JOIN DBS d ON t.DB_ID = d.DB_ID
        JOIN SDS s ON t.SD_ID = s.SD_ID
        JOIN COLUMNS_V2 c ON s.CD_ID = c.CD_ID
        ORDER BY d.NAME, t.TBL_NAME, c.INTEGER_IDX
    """)
    columns = cursor.fetchall()

    cursor.close()
    conn.close()

    # 组装成结构化数据
    metadata = {
        'dbs': [{'id': db[0], 'name': db[1], 'location': db[2]} for db in dbs],
        'tables': [{'db': t[0], 'name': t[1], 'type': t[2], 'create_time': str(t[3]), 'last_access': str(t[4])} for t in tables],
        'columns': [{'db': c[0], 'table': c[1], 'col': c[2], 'type': c[3], 'idx': c[4], 'comment': c[5]} for c in columns]
    }
    return metadata

# 调用示例
meta = fetch_hive_metadata('192.168.1.100', 3306, 'hive', 'hive_pass')
print(json.dumps(meta, ensure_ascii=False, indent=2))

4.2 MySQL 元数据采集

MySQL 的元数据藏在 information_schema 库里。这个库是 MySQL 自带的，里面存了所有数据库、表、字段、索引、权限等信息。你想想看，这简直就是元数据采集的「金矿」。

4.2.1 核心视图

我常用的几个视图：

• SCHEMATA：所有数据库列表
• TABLES：表信息（表名、引擎、行数、创建时间、更新时间）
• COLUMNS：字段信息（字段名、类型、是否为空、默认值、注释）
• STATISTICS：索引信息（索引名、字段、唯一性）
• TRIGGERS：触发器（用于分析数据变更逻辑）

4.2.2 增量采集策略

全量采集 information_schema 对性能有影响，尤其是 TABLES 视图，它会扫描所有表的数据文件来统计行数。我建议：

1. 首次全量：把库、表、字段、索引全部拉下来，存到元数据中心。
2. 后续增量：利用 UPDATE_TIME 字段。MySQL 5.6+ 的 TABLES 视图有 UPDATE_TIME，记录表结构或数据的最后变更时间。
3. 变更捕获：如果 UPDATE_TIME 不可靠（比如 MyISAM 引擎不支持），可以用 SHOW TABLE STATUS 配合 INFORMATION_SCHEMA.TABLES 做交叉验证。

-- 增量采集示例：只采集最近24小时变更的表
SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_ROWS, UPDATE_TIME
FROM information_schema.TABLES
WHERE UPDATE_TIME >= NOW() - INTERVAL 1 DAY
  AND TABLE_SCHEMA NOT IN ('mysql', 'performance_schema', 'sys');

4.3 Kafka 元数据采集

Kafka 的元数据跟 Hive、MySQL 不太一样。它没有关系型数据库那样的表结构，它的元数据是分布式的，存在 ZooKeeper 或 KRaft 里。说白了，你要采集的是：Topic、Partition、Consumer Group、Offset 这些信息。

4.3.1 采集哪些信息？

元数据项	来源	用途
Topic 列表	Kafka Broker API	知道有哪些数据流
Partition 数量与副本	Kafka Broker API	了解数据分布和容错
Consumer Group 列表	Kafka Consumer API	知道谁在消费数据
Offset 信息	Kafka Consumer API	分析消费延迟、数据积压
消息 Schema	Schema Registry（如果用了 Avro/Protobuf）	字段级血缘分析

4.3.2 采集脚本示例

用 kafka-python 库采集 Topic 和 Consumer Group 信息：

from kafka import KafkaAdminClient, KafkaConsumer
from kafka.admin import ConfigResource, ConfigResourceType

def fetch_kafka_metadata(bootstrap_servers):
    admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)

    # 采集 Topic 列表
    topics = admin_client.list_topics()
    print(f"Topics: {topics}")

    # 采集每个 Topic 的配置（分区数、副本因子）
    topic_configs = {}
    for topic in topics:
        config_resource = ConfigResource(ConfigResourceType.TOPIC, topic)
        configs = admin_client.describe_configs([config_resource])
        topic_configs[topic] = {
            'partitions': configs[topic].get('partitions', 'N/A'),
            'replication_factor': configs[topic].get('replication.factor', 'N/A')
        }

    # 采集 Consumer Group 列表
    consumer_groups = admin_client.list_consumer_groups()
    print(f"Consumer Groups: {consumer_groups}")

    admin_client.close()
    return {'topics': topics, 'topic_configs': topic_configs, 'consumer_groups': consumer_groups}

# 调用示例
meta = fetch_kafka_metadata(['localhost:9092'])
print(meta)

4.4 三种采集的对比与整合

好，三种引擎的采集方法都讲完了。你可能会问：它们之间有什么异同？我画了一张图，帮你理清思路。

从图上你能看到，三种采集方式最终汇入同一个元数据中心。这样做的好处是：血缘分析引擎只需要对接一个地方，不用管底层是 Hive 还是 Kafka。

4.4.1 采集策略对比

维度	Hive	MySQL	Kafka
采集方式	JDBC 直连 Metastore	SQL 查询 information_schema	AdminClient API + Consumer API
增量支持	通过 LAST_ACCESS_TIME 或分区时间	通过 UPDATE_TIME	通过 Consumer Group Offset 变化
性能风险	低（Metastore 是独立库）	中（information_schema 扫描影响大）	高（API 调用频繁会压 Broker）
字段级血缘	直接支持（COLUMNS_V2）	直接支持（COLUMNS）	需要 Schema Registry 配合

4.5 本章小结

嗯，这一章内容不少。咱们把 Hive、MySQL、Kafka 三种典型引擎的元数据采集方法都过了一遍。核心要点就三个：

• Hive：连 Metastore，采 DBS、TBLS、COLUMNS_V2、PARTITIONS。
• MySQL：查 information_schema，注意增量策略，别扫全表。
• Kafka：用 AdminClient 和 Consumer API，采 Topic、Partition、Consumer Group、Offset。