数据采集与同步

本章目标：能说清业务库、日志、埋点、第三方数据怎么进数仓，理解全量/增量、幂等去重和文件格式选择。 前置知识：读完 Hadoop 基础 和 PySpark 实战；本章为选学，不要求搭 Kafka/Flink 集群。 预计用时：1.5 小时。

1. 为什么需要它

数仓的第一道关口是采集。源数据进不来、进错了、重复了，后面的清洗和指标全都会歪。一条典型链路：

业务库(MySQL) --Sqoop/DataX/Flink CDC--\
日志/埋点 --Flume/Filebeat/Kafka--------> HDFS/OSS/S3

                                            |
                              Hive/Spark/Flink -> ODS/DWD/DWS/ADS

                                            |
                              调度 -> 服务层(MySQL/ClickHouse)

面试被问到某个采集组件，先回答“它在链路里的位置”，再讲它解决什么问题，最后诚实说明自己有没有实操过。

2. 核心概念

2.1 四类数据源

采集前先问：数据量多大、有没有更新删除、能接受多长延迟、源库能否承受压力；答案决定批量还是 CDC、文件还是消息队列。

2.2 批量同步：Sqoop、DataX、Flink CDC

Sqoop 的典型命令（概念示例）：

sqoop import --connect jdbc:mysql://db:3306/shop --table orders \
  --username reader --password-file /user/student/.sqoop_pass \
  --target-dir /warehouse/ods/orders --incremental append \
  --check-column order_id --last-value 0 --split-by order_id --mappers 4

--split-by 决定按哪列切分 Map 任务。如果这一列分布不均，任务会倾斜；如果这一列有大量 NULL，切分也会出问题。append 只适合“只新增、不更新”的表；有更新要用 lastmodified，有删除就只能靠 CDC 或全量对比。

DataX 用 JSON 描述 reader、writer 和 channel 并发；它不记录水位，增量条件通常写在 reader 的 SQL 里，由调度传入时间范围。

Flink CDC 读 binlog 做快照 + 增量，配合 checkpoint 恢复。它不是装个插件就完事：MySQL 要开启 binlog、设置 ROW 格式、给账号复制权限，还要关注 binlog 保留时间和主从延迟。

2.3 日志采集：Flume、Filebeat、Kafka

Flume 的基本单位是 Agent：

Source（收数据） -> Channel（暂存） -> Sink（送出）

Source 常用 taildir 跟踪文件；Channel 有 memory（快但重启丢数据）和 file（可靠但慢）；Sink 可写 HDFS 或 Kafka。HDFS Sink 要控制 rollInterval、rollSize、batchSize，否则会写出大量小文件。它擅长“文件 -> HDFS/Kafka”的稳定搬运，不擅长复杂业务转换。

Filebeat 更轻量：部署在每台机器上，跟踪日志文件，记录读取偏移到 registry，输出到 Kafka、Logstash 或 Elasticsearch。机器重启后能从上次位置继续，适合大规模日志采集。

Kafka 是采集链路里的缓冲层：topic 是一类消息；partition 决定并行度，同一 partition 内有序、跨 partition 不保证顺序；消费者用 offset 记录位置，consumer group 内分摊 partition，一个 partition 同一时刻只给一个消费者。

Kafka 解决解耦、削峰和重放：埋点系统不用关心下游是谁；流量先堆在 Kafka；消费失败可以从 offset 重读。代价是管理分区数、保留时间、重复消费和消费延迟。典型链路是“应用日志 -> Filebeat/Flume -> Kafka -> Flink/Spark -> HDFS”。

2.4 全量 vs 增量

三种常见增量依据：

1. 时间戳：update_time > 上次水位。能捕获更新，但要求字段可靠、有时区统一、有索引；迟到数据可能漏掉，通常要多回溯一段窗口。
2. 自增 ID：id > 上次最大 ID。简单高效，但只能捕获新增，捕获不了更新和删除。
3. CDC：读数据库日志，insert/update/delete 都能捕获。最完整，但需要源库支持、权限和运维。

增量同步要维护水位表，记录每个表上次同步到的最大时间或 ID，流程是“读水位 -> 抽取 -> 写入 -> 更新水位”。抽取范围通常要重叠一段，重复部分靠去重解决；水位更新和业务写入最好在同一事务里，避免数据写成功但水位没更新。

注意：批处理里的“水位（watermark）”指同步进度，Flink 的 watermark 指事件时间进度。两个词字面相同、含义无关，面试时先看上下文再回答。

迟到数据按事件时间分区，不要按入库时间分区。删除数据要么用 CDC 的 delete 标记做软删除，要么定期全量对比主键。

2.5 幂等与去重

调度重跑、失败重试、Kafka 重复消费都会让同一批数据来两次。幂等指同一批数据跑一遍和跑十遍，结果一样。

常见做法：

• 分区覆盖写：每次重写目标分区，而不是追加；重跑同一天结果不变。
• MERGE/upsert：按主键匹配，存在则更新，不存在则插入；需要 Delta/Iceberg/Hudi 或支持 MERGE 的表。
• 主键去重：按主键和更新时间排序，取最新一条。

分区覆盖写模板（${biz_date} 由调度传入）：

INSERT OVERWRITE TABLE dwd_orders PARTITION (dt = '${biz_date}')
SELECT * FROM ods_orders_inc WHERE dt = '${biz_date}';

同一分区重跑多次，结果一致。注意不要用普通 INSERT INTO 追加重跑，否则数据翻倍。

按主键去重的写法见练习 3：按 order_id 分组，update_time 相同就用 ingest_time 兜底；如果还相同，说明源数据本身有重复，要和上游确认。

面试高频：至少一次（at-least-once）不丢但可能重复；精确一次（exactly-once）需要源端可重放 + 端到端事务或幂等写入。实际工程里更常见的是“至少一次 + 幂等去重”，不要张口就说“我们保证 exactly-once”。

2.6 文件格式与压缩

列式格式只读用到的列，配合谓词下推能大幅减少 I/O，所以 DWD/DWS 层通常用 Parquet 或 ORC。CSV/JSON 适合贴源和交换，不适合反复扫描。

压缩选择：

Parquet/ORC 按块压缩，整个文件仍可按 row group 并行读；单个 gzip 文本文件在 HDFS 上不能切分，只能一个 Task 从头读到尾，大日志尤其要注意。

3. 动手示例

3.1 模拟增量抽取与去重

用 pandas 模拟“按时间水位抽增量”的过程。保存为 scripts/ingest_inc_demo.py：

# -*- coding: utf-8 -*-
import pandas as pd
orders = pd.read_csv("data/sample/orders_demo.csv",
                     parse_dates=["order_date"])
# 假设上次已同步到这一天，真实项目从水位表读取
WATERMARK = pd.Timestamp("2024-01-05")
END = pd.Timestamp("2024-01-11")   # 右开区间，覆盖到 01-10 全天
inc = orders[(orders.order_date > WATERMARK)
             & (orders.order_date < END)]
inc = inc.sort_values("order_date").drop_duplicates("order_id", keep="last")
print("增量订单数:", len(inc))

在示例数据上应输出 增量订单数: 5。这 5 条是水位之后、采集截止日之前的订单。右边界用 < 2024-01-11 而不是 <= 2024-01-10，否则会漏掉截止日当天 00:00 之后的所有订单。

注意：这里用 order_date 只是演示，它捕获不了“订单状态从 shipped 变成 delivered”这类更新。真实订单表要用 update_time 或 CDC；只用下单日期做增量，会漏掉所有后续变更。

4. 常见坑

5. 在线练习

自测：

• 能根据数据量、时效和更新方式，为商品维表、日志、订单表各选一种采集方式。
• 能解释全量同步、增量同步和 CDC 的适用场景。
• 能说出同步水位和 binlog 位点在增量同步中的作用。
• 能设计一个“重跑不重复”的落库方案。

6. 面试延伸

面试题：全量和增量怎么选？怎么保证不重复？

是什么：全量每次抽整表，增量只抽变化部分；不重复靠幂等写入和主键去重。

解决什么问题：全量简单、能发现删除，但大表成本高；增量省资源，但依赖可靠的水位和变更捕获。工程上用“增量 + 定期全量校准”，并用分区覆盖或 MERGE 保证重跑安全。

我的实操边界：我在本地模拟过按时间水位抽增量和主键去重，理解全量/增量的取舍；没有在生产上维护过 CDC 链路和水位表，Kafka/Flink 只了解概念。

面试题：Kafka 在采集链路里解决什么问题？

是什么：Kafka 是分布式消息队列，按 topic 和 partition 组织消息，消费者用 offset 记录位置，消费者组分摊分区。

解决什么问题：解耦上游和下游、削峰、缓冲和重放。埋点系统只管写 Kafka，下游可以按自己的速度消费；失败可以从 offset 重读。

我的实操边界：我理解 topic/partition/offset/consumer group 和“分区内有序”，没有在生产环境部署和运维过 Kafka，也没调过 ISR、副本和分区扩容。

7. 本章验收

• 能画出“业务库/日志/埋点 -> 采集 -> 存储 -> 计算 -> 分层”的链路
• 能说出 Sqoop、DataX、Flink CDC 各自适合什么场景
• 能说出 Flume 的 Source/Channel/Sink 和 Kafka 的 topic/partition/offset
• 能解释全量和增量的取舍，说出时间戳、自增 ID、CDC 三种增量依据
• 能写按主键去重的 ROW_NUMBER SQL
• 能说出至少两种幂等写入方式，并解释为什么普通 INSERT INTO 不安全
• 能为日志、明细表、Kafka 事件分别选择合适的格式和压缩
• 完成 3 道在线练习，至少 2 道能独立写出答案