4. 元数据采集引擎：设计可插拔的采集框架

元数据采集，说白了就是给数据目录「喂数据」的管道。没有它，目录就是个空壳子。

我见过不少团队，一开始只接了个 MySQL，后来要接 Hive，再后来要接 Kafka，每次都得改采集代码。改着改着就乱了，采集脚本散落在各个项目里，没人敢动。嗯，这就是典型的「硬编码采集」的坑。

这一章，我们就来聊聊怎么设计一个可插拔的采集框架。让新增数据源像插 U 盘一样简单。

4.1 核心设计思路：插件化与SPI

可插拔的核心，就是「面向接口编程」。Java 里最常用的机制是 SPI（Service Provider Interface）。你定义好采集接口，每个数据源自己实现，然后通过配置文件注册进来。框架启动时自动加载所有实现，互不干扰。

我习惯把采集器抽象成三层：

  • 连接层：负责建立与数据源的连接，处理认证、网络、连接池。
  • 抽取层：负责拉取元数据，比如表结构、分区信息、Topic 列表。
  • 转换层：把不同数据源的元数据格式，统一成内部标准模型。

这样做的好处很明显——你改 Kafka 采集器，不会影响到 JDBC 采集器。各管各的，出了问题也好排查。

关键原则：采集器之间不能有共享状态。每个采集器都是独立的「黑盒」，只通过标准模型与框架通信。

4.2 采集接口定义

我们先定义一个最核心的接口。我通常会这样设计：

public interface MetadataCollector {

    /**
     * 采集器唯一标识，比如 "jdbc-mysql", "hive", "kafka"
     */
    String type();

    /**
     * 初始化采集器，传入连接配置
     */
    void init(Map<String, String> config);

    /**
     * 执行采集，返回统一元数据模型
     */
    CollectResult collect(CollectContext context);

    /**
     * 增量采集，只返回变更的部分
     */
    CollectResult collectIncremental(CollectContext context);

    /**
     * 测试连接是否可用
     */
    boolean testConnection(Map<String, String> config);
}

你可能会问，为什么要有 collectIncremental？

我在项目中遇到过，全量采集一张大表，几万个字段，每次跑十几分钟。业务方等不及。后来加了增量采集，只抓上次采集后新增或修改的表，效率提升 80%。

4.3 三种典型采集器的实现

我们来看看三种最常用的采集器是怎么实现的。我挑几个关键点说。

4.3.1 JDBC 采集器

JDBC 是最通用的关系型数据库采集方式。核心逻辑就是查 information_schema。

public class JdbcCollector implements MetadataCollector {

    private DataSource dataSource;

    @Override
    public void init(Map<String, String> config) {
        // 用 HikariCP 创建连接池
        HikariConfig hikariConfig = new HikariConfig();
        hikariConfig.setJdbcUrl(config.get("url"));
        hikariConfig.setUsername(config.get("username"));
        hikariConfig.setPassword(config.get("password"));
        this.dataSource = new HikariDataSource(hikariConfig);
    }

    @Override
    public CollectResult collect(CollectContext context) {
        CollectResult result = new CollectResult();
        try (Connection conn = dataSource.getConnection()) {
            DatabaseMetaData metaData = conn.getMetaData();
            // 获取所有表
            ResultSet tables = metaData.getTables(
                null, null, "%", new String[]{"TABLE", "VIEW"}
            );
            while (tables.next()) {
                String schema = tables.getString("TABLE_SCHEM");
                String tableName = tables.getString("TABLE_NAME");
                // 获取字段信息
                ResultSet columns = metaData.getColumns(
                    null, schema, tableName, "%"
                );
                // ... 组装成标准模型
            }
        } catch (SQLException e) {
            result.setSuccess(false);
            result.setErrorMsg(e.getMessage());
        }
        return result;
    }
}

这里有个坑。不同数据库的 information_schema 字段名不一样。比如 MySQL 用 TABLE_SCHEM，Oracle 用 OWNER。我建议在配置里加一个 dbType 参数，根据类型做适配。

4.3.2 Hive 采集器

Hive 的元数据其实存在 Metastore 里，通常是 MySQL 或 PostgreSQL。你可以直接连 Metastore 的库查 TBLS、COLUMNS_V2 这些表。

但我更推荐用 Hive Metastore Client 的 API。为什么呢？因为直接查库，万一 Metastore 的表结构升级了，你的采集器就挂了。用 API 更稳定。

public class HiveCollector implements MetadataCollector {

    private IMetaStoreClient client;

    @Override
    public void init(Map<String, String> config) {
        HiveMetaStoreClient.Builder builder = new HiveMetaStoreClient.Builder();
        builder.setMetaStoreUris(config.get("metastore.uris"));
        this.client = builder.build();
    }

    @Override
    public CollectResult collect(CollectContext context) {
        CollectResult result = new CollectResult();
        try {
            List<String> databases = client.getAllDatabases();
            for (String db : databases) {
                List<String> tables = client.getAllTables(db);
                for (String table : tables) {
                    Table hiveTable = client.getTable(db, table);
                    // 提取字段、分区、存储格式等信息
                    // ... 组装成标准模型
                }
            }
        } catch (TException e) {
            result.setSuccess(false);
            result.setErrorMsg(e.getMessage());
        }
        return result;
    }
}

小技巧：Hive 采集时，记得把分区字段和普通字段区分开。很多目录工具把分区字段当普通字段展示，导致用户查询时总报错。

4.3.3 Kafka 采集器

Kafka 的元数据相对简单，主要是 Topic 列表、分区数、副本数、以及 Topic 的配置信息。

public class KafkaCollector implements MetadataCollector {

    private AdminClient adminClient;

    @Override
    public void init(Map<String, String> config) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.get("bootstrap.servers"));
        this.adminClient = AdminClient.create(props);
    }

    @Override
    public CollectResult collect(CollectContext context) {
        CollectResult result = new CollectResult();
        try {
            ListTopicsResult topicsResult = adminClient.listTopics();
            Set<String> topicNames = topicsResult.names().get();
            for (String topic : topicNames) {
                DescribeTopicsResult describeResult = adminClient.describeTopics(
                    Collections.singletonList(topic)
                );
                TopicDescription description = describeResult.all().get().get(topic);
                // 提取分区数、副本分配等信息
                // ... 组装成标准模型
            }
        } catch (Exception e) {
            result.setSuccess(false);
            result.setErrorMsg(e.getMessage());
        }
        return result;
    }
}

Kafka 采集有个特殊点：Topic 的 Schema 信息通常不在 Kafka 本身，而在 Schema Registry 里。所以采集器需要额外对接 Schema Registry，把 Avro/Protobuf 的字段定义拉下来。

4.4 统一元数据模型

不同数据源的元数据千差万别，但最终都要映射成一套标准模型。我一般这样设计：

数据源类型 原始概念 统一模型映射
MySQL Database → Table → Column Catalog → Schema → Table → Field
Hive Database → Table → Column / Partition Catalog → Schema → Table → Field + PartitionKey
Kafka Cluster → Topic → Schema Field Catalog → Schema → Table → Field
Elasticsearch Cluster → Index → Mapping Field Catalog → Schema → Table → Field

你看，不管底层多复杂，最终都抽象成「Catalog → Schema → Table → Field」这四层。这样上层应用（比如数据地图、血缘分析）就不用关心底层是什么数据源了。

4.5 采集调度与增量策略

采集器写好了，怎么跑？总不能每次都全量扫一遍吧。

我建议采用「全量 + 增量」的混合策略：

  • 首次采集：全量拉取，写入目录库。
  • 定时增量：每隔 5-10 分钟，只拉取变更的部分。
  • 定期全量：每天凌晨做一次全量对账，修复增量可能遗漏的数据。

增量采集怎么判断变更？不同数据源方法不同：

  • JDBC：查 information_schema 的 UPDATE_TIME 字段。
  • Hive：对比 Metastore 中表的 LAST_ACCESS_TIME 或 CREATE_TIME。
  • Kafka：通过 DescribeConfigs 看配置版本号是否变化。

注意：增量采集不是万能的。有些数据源（比如老版 Hive）不支持记录变更时间。这时候只能退而求其次，缩短全量采集的周期，或者用事件通知机制（比如监听 Metastore 的 binlog）。

4.6 框架整体架构

说了这么多，我们来看一张整体架构图。它把采集框架的核心流程串起来了。

可插拔元数据采集框架架构 采集调度器 JDBC 采集器 Hive 采集器 Kafka 采集器 元数据转换器（统一模型映射） 元数据存储（图数据库/关系库） 数据目录 API / 搜索 / 血缘分析 新增数据源只需实现 MetadataCollector 接口，注册到 SPI 配置即可，无需修改框架代码

从图上你能看到，整个流程是单向的：调度器触发采集器，采集器拉取原始元数据，转换器统一格式，最后存入存储层供上层使用。每个环节都可以独立扩展。

4.7 避坑指南

最后，分享几个我踩过的坑：

  • 连接泄漏：采集器用完连接一定要关闭。我曾经有个 JDBC 采集器忘了关 ResultSet，导致连接池被占满，线上数据库被打挂。后来加了 try-with-resource 才解决。
  • 超时处理：有些数据源（比如 Hive）元数据查询很慢，动辄几十秒。一定要给采集设置超时时间，超时了就跳过，别卡住整个调度。
  • 版本兼容：Kafka 2.x 和 3.x 的 AdminClient API 有差异。我建议在采集器里做版本检测，或者干脆维护两套实现。
  • 数据一致性：增量采集时，如果数据源在采集过程中发生了变更，可能会漏掉或重复。我的做法是加一个「对账任务」，每天凌晨全量扫一遍，修复不一致的数据。

我的习惯：每个采集器上线前，先在测试环境跑一周，观察资源消耗和采集延迟。没问题了再推到生产。别一上来就全量跑，容易出事。

好了，这一章的内容就到这里。可插拔采集框架的设计，说白了就是「接口 + 实现 + 调度」的组合。把接口定义好，实现各自独立，调度统一管理，剩下的就是填坑了。嗯，填坑也是工程师的必修课。

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