📚 课程目录第4章数据源接入实战

4、数据源接入实战：关系型数据库、数据湖、API、消息队列的元数据采集

好，咱们进入实战环节。前面讲了数据目录的概念和架构，这一章我带你亲手把元数据“拽”进来。

说白了，数据目录要是没有数据源接入，就是个空壳子。我见过不少团队，搭好平台后发现连不上业务库，或者采集效率低到被DBA投诉。嗯，这里头门道不少。

咱们今天重点攻克四种主流数据源：关系型数据库、数据湖、API 和 消息队列。每种我都会给出具体的采集策略和代码片段。

核心原则：元数据采集不能影响业务系统的正常运行。读库优先，增量优先，采样优先。

4.1 关系型数据库：JDBC 直连与增量采集

关系型数据库是元数据采集的“第一站”。MySQL、PostgreSQL、Oracle 这些，基本都支持 JDBC 或 ODBC 协议。

我个人习惯用 JDBC 元数据 API 来获取表结构。DatabaseMetaData 这个类，能拿到几乎所有你需要的信息：表名、列名、类型、主键、索引。

// 核心代码片段：获取表元数据
Connection conn = DriverManager.getConnection(url, user, password);
DatabaseMetaData meta = conn.getMetaData();

// 获取所有表
ResultSet tables = meta.getTables(null, null, "%", new String[]{"TABLE"});
while (tables.next()) {
    String tableName = tables.getString("TABLE_NAME");
    String tableComment = tables.getString("REMARKS");
    
    // 获取列信息
    ResultSet columns = meta.getColumns(null, null, tableName, "%");
    while (columns.next()) {
        String colName = columns.getString("COLUMN_NAME");
        int colType = columns.getInt("DATA_TYPE");
        String colComment = columns.getString("REMARKS");
        // 写入元数据存储
    }
}

小技巧：对于千万级大表，全量扫描元数据可能会锁表。我建议用 information_schema 视图代替 JDBC 元数据 API，性能更好，而且不会产生锁。

增量采集怎么做？嗯，这里有个坑。很多团队一开始就全量采集，结果每天跑几小时，数据还没更新完。

我推荐两种增量策略：

基于时间戳：记录每张表的 LAST_ALTERED 时间，只采集变更过的表。
基于事件通知：用数据库的 binlog 或 CDC（Change Data Capture）工具，实时捕获 DDL 变更。

我曾经在一个金融项目里，用 Debezium 监听 MySQL binlog，把元数据变更延迟控制在秒级。DBA 一开始还担心性能，实测下来对主库影响不到 1%。

4.2 数据湖：Hive 与 Iceberg 的元数据扫描

数据湖的元数据采集，跟关系型数据库不太一样。Hive 的元数据存在 Metastore 里，Iceberg 的元数据存在表目录里。

说白了，你得先搞清楚数据湖的“元数据存储层”在哪。

数据湖类型	元数据存储位置	采集方式	注意事项
Hive	Metastore（MySQL/PostgreSQL）	JDBC 直连 Metastore 库	注意版本兼容性
Iceberg	表目录（HDFS/S3）	读取 metadata.json	快照隔离级别
Delta Lake	_delta_log 目录	解析 JSON 日志	事务日志回放
Hudi	.hoodie 目录	读取 timeline 元数据	区分 MOR 和 COW

对于 Hive，我习惯直接连 Metastore 的数据库。但要注意，Metastore 的版本必须跟 Hive 版本匹配，否则字段映射会出错。嗯，我踩过这个坑。

# 采集 Hive 表元数据（Python 示例）
from pyhive import hive

conn = hive.Connection(host='metastore_host', port=10000)
cursor = conn.cursor()

cursor.execute("SHOW DATABASES")
databases = cursor.fetchall()

for db in databases:
    cursor.execute(f"SHOW TABLES IN {db[0]}")
    tables = cursor.fetchall()
    for table in tables:
        cursor.execute(f"DESCRIBE {db[0]}.{table[0]}")
        columns = cursor.fetchall()
        # 解析列名、类型、注释
        for col in columns:
            print(f"{db[0]}.{table[0]}.{col[0]}: {col[1]}")

警告：直接扫描 HDFS/S3 上的文件目录来推断元数据，效率极低。一定要通过 Metastore 或表格式的元数据 API 来采集。

4.3 API 数据源：RESTful 与 GraphQL 的元数据发现

API 的元数据采集，是最容易被忽视的一块。很多团队只关注数据库和数据湖，结果 API 接口成了“数据黑洞”。

API 元数据包括：端点路径、请求参数、响应结构、认证方式、限流策略。这些信息通常藏在 OpenAPI/Swagger 文档 里。

我建议的做法是：

从 API 网关或注册中心获取服务列表。
拉取每个服务的 OpenAPI 规范（JSON/YAML）。
解析 paths、components/schemas 等字段。
将端点、参数、响应模型写入数据目录。

// 解析 OpenAPI 3.0 规范
OpenAPI openAPI = new OpenAPIParser()
    .read("https://api-gateway/v3/api-docs", null, null);

Map<String, PathItem> paths = openAPI.getPaths();
for (Map.Entry<String, PathItem> entry : paths.entrySet()) {
    String path = entry.getKey();
    PathItem item = entry.getValue();
    
    // 提取 GET 方法的响应模型
    Operation getOp = item.getGet();
    if (getOp != null) {
        Schema<?> responseSchema = getOp
            .getResponses()
            .get("200")
            .getContent()
            .get("application/json")
            .getSchema();
        // 递归解析 schema 字段
        extractFields(responseSchema, path);
    }
}

对于没有 OpenAPI 文档的“野接口”，怎么办？我曾经在一个遗留系统里，用 流量录制 的方式，从网关日志里反向推导出接口结构。虽然不完美，但至少能覆盖 80% 的常用接口。

4.4 消息队列：Kafka Topic 的 Schema 注册与采样

消息队列的元数据，核心是 Topic 和 Schema。Kafka 生态里，Schema Registry 是元数据采集的关键入口。

我个人习惯的采集流程：

连接 Schema Registry，获取所有 subject（对应 Topic）。
拉取每个 subject 的最新 schema 版本。
解析 Avro/Protobuf/JSON Schema 定义。
记录 Topic 的分区数、副本数、消息保留策略。

# 从 Schema Registry 采集 Avro Schema
import requests

registry_url = "http://schema-registry:8081"
subjects = requests.get(f"{registry_url}/subjects").json()

for subject in subjects:
    # 获取最新版本
    version = requests.get(
        f"{registry_url}/subjects/{subject}/versions/latest"
    ).json()
    
    schema = version["schema"]
    # schema 是 JSON 字符串，包含 type、fields、doc 等
    print(f"Subject: {subject}")
    print(f"Schema: {schema[:200]}...")

技巧：如果 Topic 没有注册 Schema，可以采样最近几条消息来推断字段结构。但要注意，采样只能覆盖部分字段，空值字段可能会被遗漏。

嗯，这里有个常见问题：Kafka 消息的 key 和 value 可能用不同的 schema。采集时一定要分别处理，否则数据目录里的字段会不完整。

4.5 统一采集调度与元数据合并

四种数据源都接进来了，怎么统一管理？我建议建一个 采集调度层，负责：

定时触发各数据源的采集任务。
处理采集失败的重试和告警。
将异构的元数据模型转换为统一格式。

统一元数据模型长什么样？我一般用这个结构：

字段	说明	示例
source_type	数据源类型	mysql / hive / api / kafka
source_name	数据源名称	prod-mysql-01
entity_type	实体类型	table / topic / endpoint
entity_name	实体名称	orders / user-events
fields	字段列表（JSON）	[{"name":"id","type":"int"}]
last_updated	采集时间戳	2025-03-20T10:30:00Z

我曾经在一个项目里，把四种数据源的元数据统一成这个模型后，下游的数据发现和血缘分析一下子就通了。之前各团队各自为政，数据目录里一堆重复和冲突的元数据，搞得没人敢用。

关键点：元数据合并时，以“数据源 + 实体名”作为唯一键。如果同一个实体出现在多个数据源（比如 Kafka Topic 和 Hive 表对应同一份数据），用 lineage 关联，不要硬合并。

好了，四种数据源的接入实战就讲到这里。每种都有不同的坑和技巧，但核心思路是一致的：读库优先、增量优先、统一模型。你动手做的时候，先从关系型数据库开始，这个最成熟，踩坑成本最低。