4、元数据采集适配器：插件架构与三种采集模式

元数据采集这件事，说白了就是要把各种数据源里的表结构、字段信息、分区信息给捞出来。听起来简单，但真做起来你会发现——每个数据源的脾气都不一样。

MySQL 用 JDBC 就能搞定，Hive 得走 MetaStore API，Kafka 得解析日志，还有些自研系统连文档都不全。所以我一直觉得，采集框架必须得灵活。怎么灵活？插件化。

4.1 基于插件架构的采集框架

我最早做元数据平台的时候，犯过一个错：把所有采集逻辑写在一个大类里。结果每次加新数据源，都要改核心代码，测试回归一遍，上线还提心吊胆。后来我重构了一次，改成插件架构，世界清净了。

插件架构的核心思路就一句话：定义好接口，让每个数据源自己实现。

public interface MetadataCollector {
    // 采集器初始化
    void init(Map<String, String> config);
    
    // 获取所有数据库/目录
    List<Catalog> listCatalogs();
    
    // 获取指定库下的所有表
    List<Table> listTables(String catalog, String schema);
    
    // 获取表的详细元数据（字段、分区、注释等）
    TableDetail getTableDetail(TableIdentity identity);
    
    // 判断是否支持增量采集
    boolean supportIncremental();
    
    // 增量采集（返回变更时间戳之后的数据）
    List<ChangeEvent> incrementalCollect(long lastTimestamp);
}

每个数据源写一个实现类，打成 jar 包，放到采集器的插件目录下。框架启动时自动扫描加载。这样加新数据源，不需要改一行框架代码。

嗯，这里要注意：插件和框架之间的版本兼容性。我建议用 SPI 机制（ServiceLoader）或者 OSGi，别用 Class.forName 硬加载，否则类冲突够你喝一壶的。

4.2 三种采集模式

根据数据源的类型和开放能力，我把采集模式分成三种。每种都有它的适用场景和坑。

采集模式	适用数据源	优点	缺点
JDBC 模式	MySQL、PostgreSQL、Oracle、SQL Server	标准 SQL，通用性强	大表查询慢，对数据库有压力
API 模式	Hive MetaStore、Glue、Atlas、DataHub	效率高，支持批量	依赖第三方 API 版本
日志模式	Kafka、Flume、自研系统	实时性好，不侵入源系统	解析复杂，需要维护日志格式

4.2.1 JDBC 模式

这是最基础的模式。说白了就是连上数据库，执行 information_schema 或者 SHOW CREATE TABLE 之类的 SQL。

我在项目中遇到过一个问题：某个 MySQL 实例上有 2 万多张表，全量采集一次要跑 40 分钟。后来怎么优化的？分页 + 并行。把库拆成多个线程，每个线程查一部分，时间降到了 8 分钟。

4.2.2 API 模式

对于 Hive、Iceberg 这类数据湖组件，直接调 MetaStore API 比 JDBC 快得多。Hive 的 MetaStore Thrift API 可以一次拿到所有表的分区信息，不用一条条查。

我记得有一次对接阿里云的 DataWorks，他们的 API 有频率限制，每秒最多 10 次。我一开始没注意，结果采集到一半被限流了。后来加了重试和退避策略，才稳定下来。

// Hive MetaStore API 示例（伪代码）
val client = new HiveMetaStoreClient(conf)
val databases = client.getAllDatabases()
databases.foreach { db =>
    val tables = client.getAllTables(db)
    tables.foreach { table =>
        val tableObj = client.getTable(db, table)
        // 解析 tableObj 的字段、分区、存储格式等
    }
}

4.2.3 日志模式

有些系统不提供 JDBC 也不提供 API，比如自研的元数据服务。这时候只能从日志里捞。

日志模式的核心是解析。你得知道日志的格式，是 JSON 还是 CSV，字段怎么对齐。我建议统一用 JSON 格式，解析起来最省事。

举个例子，Kafka 里每条消息可能长这样：

{
  "event_type": "CREATE_TABLE",
  "table_name": "orders",
  "schema": "public",
  "columns": [
    {"name": "id", "type": "bigint", "comment": "主键"},
    {"name": "amount", "type": "decimal(10,2)", "comment": "金额"}
  ],
  "timestamp": 1700000000000
}

采集器消费 Kafka 消息，解析后写入元数据存储。这种方式实时性最好，但有个前提——源系统必须把变更事件发到消息队列里。

4.3 增量与全量同步策略

全量采集和增量采集，不是二选一，而是配合使用。

全量采集：第一次接入数据源时，把所有元数据拉一遍。之后定期（比如每天凌晨）跑一次，作为兜底。

增量采集：在全量采集的基础上，监听变更事件，只拉取变化的部分。

我个人的习惯是：全量保底，增量提效。

具体策略如下：

• 首次接入：全量采集，写入基线数据
• 日常运行：增量采集，监听变更事件（DDL 日志、API 变更通知等）
• 定期校验：每天凌晨跑一次全量，和增量数据做对比，发现不一致就修复
• 异常恢复：如果增量采集断了（比如 Kafka 挂了），恢复后先补全量，再切回增量

你想想看，如果没有增量采集，每天全量跑一次，2 万张表的数据量，光网络开销就够呛。有了增量，大部分时间只处理几十张变更的表，效率提升不是一点半点。

4.4 整体架构图

下面这张图展示了采集适配器的整体架构。从插件加载到三种采集模式，再到全量/增量策略的配合，一目了然。

4.5 实战中的几个建议

最后，分享几个我在实战中踩过的坑，希望能帮你少走弯路。

• 插件版本管理：每个采集器 jar 包都要带版本号，框架里做版本兼容检查。我曾经因为插件版本和框架不匹配，导致 ClassNotFoundException，排查了半天。
• 采集超时处理：JDBC 查询设置 Statement Timeout，API 调用设置 HTTP Timeout。别让一个慢查询拖垮整个采集任务。
• 增量采集的幂等性：同一个变更事件可能被消费多次，采集器要能去重。我一般用事件 ID + 时间戳做唯一键。
• 监控告警：采集任务的成功率、延迟、数据量都要上报。如果连续 3 次采集失败，自动告警并切换为全量模式。