📚 课程目录第4章数据源接入策略

4、数据源接入策略：关系型数据库、NoSQL数据库、消息队列、文件系统接入

数据目录系统好不好用，接入能力是第一个坎。我见过太多项目，架构图画得漂亮，结果一接真实数据源就卡壳。说白了，数据源接入不是简单的「连上就行」，你得考虑增量同步、Schema 变更、性能影响、权限管控……这一章，我把四种主流数据源的接入策略掰开揉碎讲清楚。

核心原则：接入策略 = 连接方式 + 元数据采集 + 增量机制 + 权限适配。缺一个，后面都会补坑。

4.1 关系型数据库接入

关系型数据库是企业数据的老本营。MySQL、PostgreSQL、Oracle、SQL Server，基本绕不开。我个人习惯把它们分成两类：OLTP 型（业务库）和 OLAP 型（数仓）。接入策略差别很大。

4.1.1 连接方式

JDBC 是标配。但要注意，直接拿业务库的 JDBC 去扫全表，DBA 会找你喝茶。我建议用 只读副本 或 CDC（Change Data Capture） 方式。

JDBC 直连：适合开发环境或小表，配置简单，但频繁查询会影响主库性能。
CDC（Debezium + Kafka）：生产环境首选。监听 Binlog/Redo Log，对业务库零侵入。
只读副本：如果公司有从库，优先连从库，别碰主库。

我的经验：有一次接 Oracle 生产库，DBA 只给了个只读账号，结果我们 JDBC 查询太频繁，直接把归档日志撑爆了。后来改成 CDC 方式，再没出过问题。

4.1.2 元数据采集

采集什么？表结构、字段类型、注释、索引、主外键、分区信息。这些是数据目录的「骨架」。

采集项	SQL 示例（MySQL）	用途
表列表	`SHOW TABLES`	发现数据资产
字段信息	`DESCRIBE table_name`	构建字段级血缘
索引信息	`SHOW INDEX FROM table_name`	查询优化参考
分区信息	`SHOW CREATE TABLE`	数据生命周期管理

4.1.3 增量同步

全量采集一天一次就够了。增量采集才是关键。我常用的策略：

时间戳字段：如果表有 updated_at，按时间戳增量拉取。简单，但依赖业务代码规范。
Binlog 监听：通过 Debezium 捕获变更事件，实时推送到 Kafka，再由数据目录消费。实时性最好。
版本号/自增ID：适合只追加不更新的表，比如日志表。

避坑指南：我曾经遇到一个项目，表没有 updated_at，也没有自增ID，最后只能每天全量比对。10亿行的表，全量一次跑6小时。后来我强制要求业务方加时间戳字段，才解决。

4.2 NoSQL 数据库接入

NoSQL 种类太多，MongoDB、Elasticsearch、HBase、Cassandra……接入方式各不相同。但核心思路一致：通过 API 或 SDK 获取元数据，而不是 SQL。

4.2.1 MongoDB 接入

MongoDB 的元数据藏在 admin 库的系统集合里。我一般用 listCollections 和 getIndexes 命令。

// 获取所有集合
use admin
db.runCommand({ listCollections: 1 })

// 获取集合的文档结构（采样）
db.collection.aggregate([
  { $sample: { size: 100 } },
  { $project: { _id: 0 } }
])

注意，MongoDB 是 Schemaless 的，字段类型可能不一致。我建议做 Schema 推断：采样1000条文档，统计每个字段出现频率和类型分布，生成一个「近似 Schema」。

4.2.2 Elasticsearch 接入

ES 本身就是元数据管理的好手。接入时直接调用 _mapping API 获取索引映射。

GET /your_index/_mapping

返回的 JSON 里包含了字段名、类型、分词器等信息。我习惯把这些映射信息转成数据目录的字段模型，同时记录索引的 settings（分片数、副本数）。

小技巧：ES 的 mapping 一旦创建，修改成本很高。接入时最好记录 mapping 版本，方便后续对比变更。

4.2.3 HBase 接入

HBase 的元数据在 hbase:meta 表里。通过 Java API 或 Thrift 接口获取表名、列族、Region 分布。

// Java API 示例
Admin admin = connection.getAdmin();
TableName[] tableNames = admin.listTableNames();
for (TableName tn : tableNames) {
    HTableDescriptor desc = admin.getTableDescriptor(tn);
    // 获取列族信息
    ColumnFamilyDescriptor[] families = desc.getColumnFamilies();
}

HBase 的 Region 分布信息对数据目录很有价值，可以辅助做数据 locality 分析。

4.3 消息队列接入

消息队列接入，很多人会忽略。但数据目录如果能把 Kafka Topic、RocketMQ 的 Topic 结构管起来，对数据链路追踪帮助巨大。

4.3.1 Kafka 接入

Kafka 的元数据通过 AdminClient 获取。我一般采集以下信息：

Topic 列表：listTopics()
分区数、副本数：describeTopics()
消息 Schema：如果用了 Avro/Protobuf，从 Schema Registry 拉取
消费者组：listConsumerGroups()，记录消费延迟

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
try (AdminClient admin = AdminClient.create(props)) {
    ListTopicsResult topics = admin.listTopics();
    topics.names().get().forEach(System.out::println);
}

关键点：消息队列的元数据采集频率可以低一些，比如每小时一次。Topic 变更不像数据库表那么频繁。

4.3.2 RocketMQ 接入

RocketMQ 通过 MQAdminExt 获取元数据。采集项类似 Kafka，但多了 Message Queue 分配 信息。

DefaultMQAdminExt admin = new DefaultMQAdminExt();
admin.setNamesrvAddr("localhost:9876");
admin.start();

// 获取所有 Topic
Set<String> topics = admin.fetchAllTopicList().getTopicList();

// 获取 Topic 路由信息
TopicRouteData route = admin.examineTopicRouteInfo("your_topic");

4.4 文件系统接入

文件系统是最容易被忽视的数据源。HDFS、S3、NAS、FTP……文件本身也是数据资产。我见过一个公司，90% 的数据以 CSV/Parquet 文件形式躺在 HDFS 上，数据目录却只管了数据库。

4.4.1 HDFS 接入

HDFS 接入的核心是 目录结构扫描 和 文件格式识别。

// 使用 Hadoop FileSystem API
FileSystem fs = FileSystem.get(new URI("hdfs://namenode:8020"), new Configuration());
Path path = new Path("/data");
RemoteIterator<LocatedFileStatus> files = fs.listFiles(path, true);
while (files.hasNext()) {
    LocatedFileStatus file = files.next();
    // 记录路径、大小、块信息、修改时间
    System.out.println(file.getPath() + " | " + file.getLen());
}

我建议做 分层扫描：先扫目录结构，再按需深入子目录。避免一次性扫描整个 HDFS，那会触发 NameNode 的 RPC 风暴。

4.4.2 文件格式解析

不同文件格式，元数据提取方式不同：

格式	元数据提取方式	注意事项
Parquet	读取 footer 中的 Schema	支持嵌套结构，字段类型丰富
ORC	读取 stripe 统计信息	包含行数、列统计值
Avro	读取文件头部的 Schema JSON	Schema 可嵌入文件内
CSV/JSON	采样推断 Schema	需要处理脏数据、空值

我曾经踩过的坑：有一次扫描 HDFS 上的 Parquet 文件，没做采样限制，结果一个文件 2TB，读 footer 花了 10 分钟。后来我加了文件大小阈值，超过 1GB 的文件只读前 10 个 row group 的 footer。

4.4.3 对象存储（S3/MinIO）接入

S3 的接入通过 ListObjectsV2 API。注意分页和前缀过滤。

// AWS SDK 示例
ListObjectsV2Request req = ListObjectsV2Request.builder()
    .bucket("your-bucket")
    .prefix("data/")
    .maxKeys(1000)
    .build();

ListObjectsV2Response resp = s3Client.listObjectsV2(req);
resp.contents().forEach(obj -> {
    // 记录 key、大小、ETag、最后修改时间
});

对象存储的元数据还包括 存储类型（Standard/Glacier）和标签（Tagging），这些对数据生命周期管理很有用。

4.5 接入架构总览

下面这张图，是我在实际项目中总结的数据源接入架构。你可以看到，所有数据源通过统一的 接入适配层 进入数据目录，屏蔽了底层差异。

4.6 接入策略选择矩阵

不同场景下，接入策略怎么选？我整理了一个矩阵，方便你快速决策：

数据源类型	推荐接入方式	实时性	对源系统影响	适用场景
MySQL（OLTP）	CDC（Debezium）	秒级	低	生产业务库
PostgreSQL	逻辑复制 + JDBC	秒级	低	实时数仓
Oracle	OGG 或 LogMiner	分钟级	中	核心交易系统
MongoDB	Change Streams	秒级	低	文档型数据
Elasticsearch	_mapping API	分钟级	无	搜索索引管理
Kafka	AdminClient + Schema Registry	分钟级	无	消息链路追踪
HDFS	FileSystem API + 分层扫描	小时级	中	离线数仓
S3	ListObjectsV2 + 事件通知	分钟级	低	数据湖

我的建议：刚开始做数据目录时，先接关系型数据库和 HDFS，这两个是数据资产的大头。消息队列和 NoSQL 可以放在第二期。别想一口吃成胖子。

4.7 接入的通用注意事项

最后，说几个我踩过的坑，希望能帮你省点时间：

连接池管理：每个数据源单独配置连接池，避免一个数据源挂了拖垮整个接入层。
熔断与降级：接入层必须做熔断。比如某个数据源连续 3 次采集失败，自动跳过，不影响其他数据源。
元数据版本管理：每次采集的元数据打上版本号，方便回滚和对比变更。
权限最小化：接入账号只给只读权限。我见过有人用 root 账号接数据目录，结果误操作删了表……
采集频率差异化：核心业务库 5 分钟一次，离线数仓 1 小时一次，归档数据一天一次。别一刀切。

血的教训：我曾经没做熔断，结果一个 HDFS 集群挂了，接入层线程全部阻塞，导致其他 20 个数据源全部采集失败。后来加了超时和熔断，再也没出过类似问题。

数据源接入是数据目录的「地基」。地基不稳，上层再漂亮的功能都是空中楼阁。把这一章的内容吃透，后面做血缘分析、数据质量、搜索优化，都会顺手很多。