3. 元数据采集实战：采集 Kafka Topic 元数据、采集 Flink 作业元数据、采集 Hive/ClickHouse 表元数据

元数据采集是整个数据目录的「水源」。水源不干净，下游的治理、血缘、质量监控全是白搭。我见过太多团队，一上来就搞数据地图、搞血缘分析，结果连 Topic 的 Schema 都采不全，最后做出来的东西没人敢信。

这一章，咱们就动手把三个核心数据源的元数据采上来：Kafka Topic、Flink 作业、Hive/ClickHouse 表。我会把每个采集点的关键字段、采集方式、避坑经验都拆开讲。

3.1 采集 Kafka Topic 元数据

Kafka 是实时数仓的「主动脉」。Topic 的元数据如果采不全，下游的 Flink 作业和 Hive 表就没办法做血缘关联。我个人习惯把 Kafka 元数据分成两类：集群级和Topic 级。

3.1.1 采集哪些字段？

3.1.2 怎么采？

我建议用 Kafka AdminClient API 做全量采集，配合 Schema Registry REST API 拉取 Schema。代码很简单，但要注意几个坑。

// 全量采集 Topic 列表
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
try (AdminClient admin = AdminClient.create(props)) {
    ListTopicsResult topics = admin.listTopics();
    Set<String> topicNames = topics.names().get();
    
    for (String name : topicNames) {
        // 采集分区和副本信息
        DescribeTopicsResult desc = admin.describeTopics(List.of(name));
        TopicDescription td = desc.values().get(name).get();
        
        // 采集配置
        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, name);
        DescribeConfigsResult configs = admin.describeConfigs(List.of(resource));
        Config config = configs.values().get(resource).get();
        
        // 写入元数据表
        saveTopicMeta(name, td, config);
    }
}

3.1.3 增量采集怎么做？

全量采集每天跑一次就够了。但增量变更（比如新建 Topic、修改分区数）需要实时捕获。我常用的方案是：

• 监听 Kafka 内部 __consumer_offsets 或 broker 日志：但这种方式侵入性强，不推荐。
• 定时轮询对比：每 5 分钟拉一次 Topic 列表，跟上次快照做 diff。简单可靠，我一直在用。
• Hook 方式：在 Topic 创建/修改的入口（比如自研平台）主动发一条变更消息到元数据采集队列。

3.2 采集 Flink 作业元数据

Flink 作业是实时数仓的「计算引擎」。作业的 DAG、状态、Checkpoint 信息，直接决定了数据加工的链路是否健康。我记得有一次线上事故，就是因为 Flink 作业的并行度被改小了，导致数据积压，但元数据系统没有及时感知到。

3.2.1 采集哪些字段？

3.2.2 怎么采？

Flink 提供了 REST API，直接请求 JobManager 的接口就能拿到所有信息。我个人习惯封装一个 FlinkMetaClient，统一处理 API 调用和异常重试。

// 采集 Flink 作业 DAG
String jobManagerUrl = "http://flink-jobmanager:8081";
String jobsUrl = jobManagerUrl + "/jobs";

// 1. 获取所有作业列表
JSONArray jobs = httpGet(jobsUrl).getJSONArray("jobs");

for (int i = 0; i < jobs.length(); i++) {
    String jid = jobs.getJSONObject(i).getString("id");
    
    // 2. 获取作业详情（包含 DAG）
    String detailUrl = jobsUrl + "/" + jid;
    JSONObject detail = httpGet(detailUrl);
    
    // 3. 解析算子链
    JSONArray vertices = detail.getJSONObject("plan")
                               .getJSONArray("nodes");
    for (int j = 0; j < vertices.length(); j++) {
        JSONObject vertex = vertices.getJSONObject(j);
        String operatorName = vertex.getString("description");
        int parallelism = vertex.getInt("parallelism");
        // 保存算子信息
        saveOperatorMeta(jid, operatorName, parallelism);
    }
    
    // 4. 获取 Checkpoint 信息
    String cpUrl = jobsUrl + "/" + jid + "/checkpoints";
    JSONObject cp = httpGet(cpUrl);
    saveCheckpointMeta(jid, cp);
}

3.2.3 增量采集怎么做？

Flink 作业的状态变化比较频繁（RUNNING -> FAILED -> RESTARTING）。我建议：

• 全量采集：每 5 分钟拉一次所有作业列表，更新状态。
• 增量采集：通过 Flink 的 Webhook 或 Metrics Reporter，把作业状态变更推送到消息队列。这样能做到秒级感知。
• DAG 变更：DAG 一般只在作业重启时变化，所以每次全量采集时对比上次的 DAG 哈希值即可。

3.3 采集 Hive/ClickHouse 表元数据

Hive 和 ClickHouse 是实时数仓的「存储层」。表结构、分区、存储量这些信息，是数据治理的基础。说白了，没有表元数据，你连「这张表是谁建的、数据量多大」都说不清楚。

3.3.1 采集哪些字段？

3.3.2 怎么采？

Hive 我推荐直接连 HiveMetaStore Thrift API，不走 JDBC。因为 JDBC 只能查表结构，拿不到存储量、文件数这些物理信息。

// 通过 HiveMetaStore 客户端采集
IMetaStoreClient client = new HiveMetaStoreClient(hiveConf);

// 获取所有数据库
List<String> databases = client.getAllDatabases();
for (String db : databases) {
    // 获取所有表
    List<String> tables = client.getAllTables(db);
    for (String tableName : tables) {
        // 获取表结构
        Table table = client.getTable(db, tableName);
        String ddl = table.getSd().getCols().toString();
        long createTime = table.getCreateTime();
        
        // 获取分区信息
        List<Partition> partitions = client.listPartitions(db, tableName, (short)-1);
        int partitionCount = partitions.size();
        
        // 获取存储量（需要 HDFS 配合）
        long totalSize = getHdfsSize(table.getSd().getLocation());
        
        saveHiveTableMeta(db, tableName, ddl, partitionCount, totalSize);
    }
}

ClickHouse 就简单多了，直接查 system 库的系统表。

-- 采集 ClickHouse 表元数据
SELECT 
    database,
    name AS table_name,
    engine,
    create_table_query,
    metadata_modification_time
FROM system.tables
WHERE database NOT IN ('system', 'INFORMATION_SCHEMA');

-- 采集列信息
SELECT 
    database,
    table,
    name AS column_name,
    type,
    comment,
    position
FROM system.columns
WHERE database NOT IN ('system', 'INFORMATION_SCHEMA');

-- 采集分区和存储量
SELECT 
    database,
    table,
    count() AS part_count,
    sum(bytes_on_disk) AS total_bytes,
    sum(rows) AS total_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table;

3.3.3 增量采集怎么做？

Hive 和 ClickHouse 的表结构变更频率不高，但分区是每天都在变的。我的策略是：

• 表结构：每天全量采集一次，对比上次的建表语句哈希值。
• 分区信息：每小时采集一次，只采集新增的分区。Hive 可以通过 listPartitionsByFilter 按时间过滤。
• 存储量：每 6 小时采集一次，因为 HDFS 的 du 操作比较重。

3.4 统一采集调度与存储

三个数据源的采集逻辑写好了，但怎么统一调度？我个人习惯用 Quartz + 配置中心 的方式，每个采集任务是一个 Job，通过配置中心控制采集频率和开关。

// 统一采集调度配置示例（YAML）
采集任务:
  kafka_meta:
    调度: "0 */5 * * * ?"   # 每5分钟
    超时: 120秒
    重试: 3次
  flink_meta:
    调度: "0 */5 * * * ?"   # 每5分钟
    超时: 60秒
    重试: 2次
  hive_meta:
    调度: "0 0 1 * * ?"     # 每天凌晨1点
    超时: 600秒
    重试: 1次
  clickhouse_meta:
    调度: "0 0 */6 * * ?"   # 每6小时
    超时: 300秒
    重试: 2次

采集到的元数据统一写入 MySQL/PostgreSQL，表结构设计成 meta_kafka_topics、meta_flink_jobs、meta_hive_tables、meta_clickhouse_tables。每张表都带 snapshot_time 和 change_type 字段，方便做增量对比和历史回溯。