📚 课程目录第2章数据采集层设计

数据采集层设计：离线采集与实时采集的融合之道

数据采集层，说白了就是整个数据流水线的“水龙头”。

我做了这么多年数据架构，见过太多项目在采集层就翻车了。要么离线任务跑着跑着就挂了，要么实时通道丢数据丢得悄无声息。嗯，今天咱们就把这块掰开揉碎了讲清楚。

离线采集：Sqoop 与 DataX 的实战选择

离线采集，核心任务就是把关系型数据库里的数据，批量搬到 Hadoop 生态里。目前主流工具就两个：Sqoop 和 DataX。

核心观点： Sqoop 适合快速上手、小规模场景；DataX 适合复杂网络、异构数据源、需要精细控制的企业级场景。

Sqoop 的典型用法

Sqoop 是 Apache 的项目，用起来确实简单。一条命令就能把 MySQL 表导入 Hive。

# 全量导入
sqoop import \
  --connect jdbc:mysql://host:3306/db \
  --username user --password pass \
  --table orders \
  --hive-import \
  --hive-table ods.orders \
  --split-by id \
  --num-mappers 8

# 增量导入（基于时间戳）
sqoop import \
  --connect ... \
  --table orders \
  --check-column update_time \
  --incremental lastmodified \
  --last-value '2024-01-01 00:00:00' \
  --merge-key id

这里有个坑，我踩过好几次。Sqoop 的 --split-by 如果选错了列，比如选了分布不均匀的字段，会导致数据倾斜。我曾经有个项目，用 status 字段做 split-by，结果 8 个 mapper 里 6 个跑完了，2 个还在死扛。后来改成用主键 ID，问题就解决了。

DataX 的架构优势

DataX 是阿里巴巴开源的，它的设计思路更现代。核心是“Reader + Channel + Writer”的三层架构。

特性	Sqoop	DataX
数据源支持	主要支持 RDBMS + HDFS/Hive	50+ 数据源，包括 ES、MongoDB、HBase
断点续传	不支持	原生支持
流量控制	无	支持限速、并发控制
部署方式	单机	单机 + 分布式扩展

我个人习惯，新项目直接上 DataX。原因很简单：DataX 的断点续传功能，在数据量大的时候能救命。你想想看，一个跑了 6 小时的离线任务，最后 5 分钟挂了，Sqoop 得从头再来，DataX 能从断点接着跑。这差距，做过运维的都懂。

实时采集：Canal 与 Debezium 的选型逻辑

实时采集，核心是 CDC（Change Data Capture，变更数据捕获）。说白了，就是监听数据库的 binlog，把每一次增删改都实时抓出来。

目前主流方案就两个：阿里巴巴的 Canal 和 Red Hat 的 Debezium。

Canal：MySQL 生态的首选

Canal 是阿里巴巴为了解决 MySQL 和 Redis 同步问题而开发的。它伪装成 MySQL 的从库，拉取 binlog 数据。

# Canal 配置示例（instance.properties）
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset=UTF-8
# 监听所有表
canal.instance.filter.regex=.*\\..*
# 解析后投递到 Kafka
canal.mq.topic=ods_binlog
canal.mq.partitionsNum=8

我记得有个电商项目，高峰期每秒 2 万笔订单。Canal 扛住了，延迟控制在 200 毫秒以内。但要注意，Canal 只支持 MySQL，而且对 MariaDB 的兼容性不太好。如果你用的是 PostgreSQL，那就得看 Debezium 了。

Debezium：多数据库的通用方案

Debezium 基于 Kafka Connect 构建，天生就是为流式架构设计的。它支持 MySQL、PostgreSQL、MongoDB、SQL Server、Oracle 等多种数据库。

我的建议： 如果团队已经上了 Kafka，直接选 Debezium。它和 Kafka Connect 深度集成，配置管理、监控、扩缩容都方便。如果只是 MySQL 单库，Canal 更轻量。

采集策略：全量、增量、实时，怎么搭？

实际项目中，很少只用一种策略。我总结了一个“三段式”采集模型：

首次全量： 用 DataX 把历史数据一次性拉过来
增量补录： 用 Sqoop 或 DataX 的增量模式，补上全量到实时之间的数据缺口
实时同步： 用 Canal/Debezium 持续监听 binlog

为什么会这样设计？因为 binlog 只保留一段时间（通常是 7 天）。如果你要同步 3 年前的数据，只能走离线全量。全量跑完后，再开启实时通道，中间用增量补录做衔接。

# 伪代码：采集策略调度逻辑
if is_first_run:
    run_datax_full_import()          # 全量
    run_sqoop_incremental_catchup()  # 补录
    start_canal_binlog_listener()    # 实时
else:
    if has_data_gap:
        run_sqoop_incremental_catchup()
    else:
        # 正常实时处理
        process_binlog_stream()

容错机制：数据不丢、不重、不乱

容错是采集层最头疼的问题。我见过最惨的一次，实时通道因为网络抖动断了 3 分钟，恢复后 binlog 已经过期了，丢了 10 万条订单数据。从那以后，我定了几条铁律。

避坑指南： 我曾经因为没做 offset 持久化，导致 Canal 重启后重复消费了 2 小时的 binlog，下游的实时报表数据翻了一倍。老板开会时指着大屏问：“今天订单量怎么暴涨了？” 场面一度非常尴尬。

离线采集容错

断点续传： DataX 原生支持，Sqoop 需要自己记录 last-value
重试机制： 网络超时、连接断开时自动重试 3 次，间隔指数退避
数据校验： 全量导入后，对比源库和目标库的行数、checksum

实时采集容错

offset 持久化： Canal 和 Debezium 都支持将消费位点写入 ZK 或 Kafka
幂等写入： 目标端（Kafka/HBase）要支持去重，防止重复数据
死信队列： 解析失败的 binlog 事件，丢到死信队列，人工排查

整体架构图

下面这张图，是我在项目中常用的采集层架构。它把离线、增量、实时三条路径整合在一起，通过统一的调度中心管理。

总结一下

数据采集层设计，核心就三件事：

选对工具： 离线用 DataX，实时看场景选 Canal 或 Debezium
定好策略： 全量 + 增量 + 实时，三段式衔接
做足容错： 断点续传、offset 持久化、幂等写入，一个都不能少

嗯，这一块内容比较多，但都是实打实的经验。你把这些吃透了，数据采集层基本不会出大问题。