📚 课程目录第3章数据采集与接入

3、数据采集与接入：批式采集、实时采集与数据源适配器设计

数据采集是整个 MLOps 数据底座的第一个环节。说白了，就是把散落在各个业务系统里的数据，统一搬到我们的数据平台上来。这一步做不好，后面所有模型训练、特征工程都是空中楼阁。

我个人习惯把数据采集分成两条线：批式采集和实时采集。两条线各有各的适用场景，也各有各的坑。今天我们就来聊聊这两条线怎么搭，以及怎么用数据源适配器把它们统一管理起来。

3.1 批式采集：Sqoop 与 DataX

批式采集，顾名思义，就是按固定时间窗口（比如每天凌晨、每小时）把数据批量搬过来。适用于离线报表、历史数据回溯、模型训练样本生成等场景。

3.1.1 Sqoop：传统 Hadoop 生态的搬运工

Sqoop 是 Apache 旗下的老牌工具，专门用来在关系型数据库和 Hadoop（HDFS、Hive、HBase）之间做批量数据传输。我在项目中遇到过不少团队还在用 Sqoop 1.4.x，虽然版本老，但胜在稳定。

一个典型的 Sqoop 导入命令长这样：

sqoop import \
  --connect jdbc:mysql://192.168.1.100:3306/orders_db \
  --username reader \
  --password-file /path/to/pwd.file \
  --table order_info \
  --target-dir /data/raw/orders/2025-03-01 \
  --fields-terminated-by '\001' \
  --num-mappers 8 \
  --split-by order_id \
  --incremental lastmodified \
  --check-column update_time \
  --last-value '2025-03-01 00:00:00'

这里有几个关键点：

--split-by：指定分片字段，Sqoop 会根据这个字段把数据拆成多个 mapper 并行拉取。我建议用数字类型的主键，别用字符串，否则分片不均匀。
--incremental lastmodified：增量模式。第一次全量之后，后续只拉取变更数据。嗯，这里要注意：如果源表没有 update_time 字段，这个模式就用不了。
--num-mappers：并行度。不是越大越好，我曾经调成 32 个 mapper，结果把源库的连接池打满了，业务方直接投诉。

避坑指南：我曾经在生产环境遇到过 Sqoop 导出时字符集乱码的问题。原因是源库是 UTF-8，但 Sqoop 默认用 Latin1 读取。解决方案是在 jdbc URL 里显式加上 ?useUnicode=true&characterEncoding=UTF-8。

3.1.2 DataX：阿里巴巴的异构数据同步利器

DataX 是阿里巴巴开源的异构数据源同步框架。相比 Sqoop，它的优势在于数据源插件化——官方支持 MySQL、Oracle、SQL Server、HDFS、Hive、HBase、MongoDB、Elasticsearch 等几十种数据源。我在做数据中台项目时，经常用它来打通不同团队的数据孤岛。

DataX 的作业配置是 JSON 格式，清晰易懂：

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "connection": [
              {
                "jdbcUrl": ["jdbc:mysql://192.168.1.100:3306/orders_db"],
                "table": ["order_info"],
                "where": "create_time >= '2025-03-01 00:00:00'"
              }
            ],
            "username": "reader",
            "password": "xxxxxx",
            "column": ["order_id", "user_id", "amount", "create_time"],
            "splitPk": "order_id"
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://namenode:8020",
            "fileType": "text",
            "path": "/data/raw/orders/2025-03-01",
            "fileName": "order_info",
            "writeMode": "append",
            "fieldDelimiter": "\001"
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 8
      }
    }
  }
}

DataX 的 channel 数控制并发度。我建议从 4 开始试，逐步往上加，直到源库的 CPU 或 IO 达到 60% 左右为止。别一上来就开 32 个 channel，容易出事。

3.1.3 Sqoop vs DataX 对比

维度	Sqoop	DataX
数据源支持	主要支持 RDBMS → Hadoop	几十种数据源，双向同步
配置方式	命令行参数	JSON 配置文件
增量同步	支持 lastmodified 和 append	需通过 where 条件自行实现
性能调优	通过 mapper 数控制	通过 channel 数控制
社区活跃度	较低（Apache 已停止维护）	较高（阿里持续更新）
适用场景	传统 Hadoop 生态	异构数据源、数据中台

我个人建议：如果团队技术栈偏 Hadoop 原生，且数据源主要是 RDBMS，Sqoop 够用。但如果涉及多种数据源（比如 MySQL + MongoDB + Elasticsearch），或者需要频繁调整同步逻辑，DataX 更灵活。

3.2 实时采集：Canal 与 Kafka

批式采集的延迟通常在分钟到小时级别。但有些场景——比如实时风控、实时推荐、在线特征计算——需要秒级甚至毫秒级的数据接入。这时候就得靠实时采集了。

3.2.1 Canal：MySQL 的 Binlog 监听器

Canal 是阿里巴巴开源的 MySQL Binlog 解析工具。它的原理很简单：伪装成 MySQL 的从库，拉取主库的 Binlog，然后解析成结构化数据推送给下游。你想想看，这样就能实时感知到数据库的每一行变更——插入、更新、删除，一个不落。

一个典型的 Canal 部署架构：

MySQL Master → Canal Server → Kafka Topic → Flink/Spark Streaming → 实时特征存储

Canal 的配置核心在 instance.properties 里：

# 源库信息
canal.instance.master.address=192.168.1.100:3306
canal.instance.dbUsername=canal_user
canal.instance.dbPassword=canal_pwd

# 监听哪些表
canal.instance.filter.regex=orders_db\\..*

# 解析后的数据投递到 Kafka
canal.mq.topic=orders_binlog
canal.mq.partitionsNum=8
canal.mq.partitionHash=order_id

经验之谈：我在项目中遇到过 Canal 消费延迟的问题。排查后发现是 Binlog 的 binlog_row_image 设置成了 MINIMAL（只记录变更列），导致 Canal 解析出的数据不完整。建议设置为 FULL，这样每条变更记录都包含完整行数据，下游处理更灵活。

3.2.2 Kafka：实时数据的中枢神经

Kafka 在实时采集链路中扮演的是消息总线的角色。Canal 把 Binlog 数据推送到 Kafka，下游的 Flink、Spark Streaming 或者特征服务再从 Kafka 消费。这样做的好处是解耦——生产者和消费者互不影响，而且 Kafka 本身的高吞吐、持久化特性保证了数据不丢失。

我建议的 Topic 设计规范：

命名规则：{数据源}.{数据库}.{表}.{事件类型}，例如 mysql.orders_db.order_info.insert
分区策略：按主键或业务 ID 哈希分区，保证同一行数据的变更顺序
保留策略：实时特征场景保留 7 天，离线回溯场景保留 30 天

3.2.3 批式 vs 实时：怎么选？

维度	批式采集	实时采集
延迟	分钟~小时级	秒级~毫秒级
数据量	适合大批量（GB~TB）	适合持续小批量
一致性	最终一致，可做全量对账	最终一致，可能出现乱序
资源消耗	集中式高消耗	持续低消耗
典型工具	Sqoop、DataX	Canal、Debezium、Kafka Connect
适用场景	离线报表、模型训练样本	实时风控、在线特征、实时监控

其实很多场景是两者结合的。比如实时特征用 Canal + Kafka 接入，每天凌晨再用 DataX 跑一次全量对账，确保数据一致性。我管这叫批流一体。

3.3 数据源适配器设计

随着业务发展，数据源会越来越多。MySQL、Oracle、MongoDB、Elasticsearch、Kafka、文件系统……如果每个数据源都单独写一套采集逻辑，维护成本会爆炸。这时候就需要一个数据源适配器来统一管理。

3.3.1 适配器模式的核心思想

说白了，就是定义一个统一的接口，让每种数据源都实现这个接口。上层调度系统只关心接口，不关心底层是 MySQL 还是 Kafka。这样新增一个数据源时，只需要写一个新的适配器，不需要改调度逻辑。

接口定义示例（Java）：

public interface DataSourceAdapter {
    // 初始化连接
    void init(DataSourceConfig config);
    
    // 读取数据，返回统一的 Record 对象
    List<Record> read(ReadContext context);
    
    // 写入数据
    void write(List<Record> records, WriteContext context);
    
    // 获取数据源元信息（表结构、分区等）
    DataSourceMeta getMeta();
    
    // 关闭连接
    void close();
}

3.3.2 适配器注册与路由

我习惯用一个适配器注册中心来管理所有适配器实例。注册中心维护一个 Map，key 是数据源类型（如 "mysql"、"kafka"、"hdfs"），value 是对应的适配器工厂类。调度系统根据任务配置里的数据源类型，从注册中心拿到适配器，然后调用统一的 read/write 接口。

这样做的好处：

扩展性：新增数据源只需实现接口并注册，不影响现有任务
隔离性：每个适配器独立管理自己的连接池、重试策略、错误处理
可测试性：可以 mock 适配器进行单元测试

3.3.3 适配器配置管理

每个数据源的连接信息、认证方式、限流策略都不一样。我建议把配置统一放在配置中心（比如 Apollo、Nacos 或 Consul）里，适配器启动时从配置中心拉取。这样修改配置不需要重启服务。

配置示例（YAML）：

adapters:
  - type: mysql
    version: 8.0
    config:
      jdbcUrl: jdbc:mysql://192.168.1.100:3306/orders_db
      username: reader
      password: ${MYSQL_PASSWORD}
      maxConnections: 10
      queryTimeout: 30s
      retryCount: 3
      
  - type: kafka
    version: 3.2
    config:
      bootstrapServers: kafka01:9092,kafka02:9092
      groupId: mlops_data_collector
      autoOffsetReset: earliest
      maxPollRecords: 500

  核心原则：适配器只负责数据格式的转换和传输，不负责业务逻辑。业务逻辑应该在上层的 ETL 或特征工程层处理。我曾经见过有人把数据清洗逻辑写在适配器里，结果换数据源时清洗逻辑也得重写，得不偿失。

3.4 本章小结：数据采集的整体架构

最后，我用一张图来总结本章的核心逻辑。这张图展示了批式采集和实时采集两条链路，以及数据源适配器如何统一管理它们。

这张图里，从上到下分了三层：数据源层、适配器层、采集方式层和目标存储层。适配器层是核心，它让上层采集任务与底层数据源解耦。批式采集和实时采集两条链路并行，互不干扰，但最终数据都会汇聚到统一的数据底座中。