数据集成层：Kafka + Flink CDC 入湖

本层定位：两条入湖通道

湖仓一体的存储底座（对象存储 + 开放表格式）只回答"数据以什么形态放"， 数据集成层回答"数据从哪里来、多久到、以什么一致性保证到"。工程上收敛为两条通道：

两条通道最终汇入同一份湖仓存储，整体分层如下：

Apache Kafka：实时主干

核心机制：分区 / 副本 / ISR

Kafka 是分布式提交日志（Commit Log），实时链路的主动脉。三个核心概念决定了它的吞吐、 顺序与可靠语义：

• 分区（Partition）：Topic 的并行单位。同一分区内的消息严格有序， 分区数直接决定生产与消费的并行度上限；key 相同的消息哈希到同一分区，是"按 key 保序"的基础。
• 副本（Replica）：每个分区有 1 个 Leader 副本与 N-1 个 Follower 副本， 读写都走 Leader，Follower 异步拉取同步；副本分布在不同 broker 以容忍机器故障。
• ISR（In-Sync Replicas）：与 Leader 保持同步的副本集合（含 Leader）。 当 Follower 落后超过 replica.lag.time.max.ms（默认 30s）即被踢出 ISR； 消息只有被 ISR 全部写入才视为"已提交"（committed），这决定了 acks=all 的真实含义。

Leader 故障时，Controller 从 ISR 中选举新 Leader——只有 ISR 内的副本才有资格当选， 因此 ISR 数量与 min.insync.replicas 共同构成持久化下限： replication.factor=3, min.insync.replicas=2 表示"最多容忍 1 台 broker 宕机仍可安全写入"。

broker 核心配置（server.properties）

# Kafka broker 核心配置（server.properties）
broker.id=1
listeners=PLAINTEXT://kafka-1:9092
log.dirs=/data/kafka1,/data/kafka2

# 网络与 IO 线程：万兆网络高吞吐场景的常规起点
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576

# 新建 topic 的默认分区数与副本数（兜底值，建 topic 时显式指定更稳）
num.partitions=6
default.replication.factor=3
min.insync.replicas=2

# 数据保留：时间与大小双上限，命中任一即删除旧 segment
log.retention.hours=72
log.retention.bytes=214748364800
log.segment.bytes=1073741824          -- segment 1GB，滚动后才能被过期清理
log.retention.check.interval.ms=300000

# 副本拉取线程与日志压缩（用于留痕/对账类 topic）
num.replica.fetchers=4
log.cleaner.enable=true

生产者可靠性：acks 与幂等

acks 决定"写入多少副本才算成功"：0（发出即忘，可能丢）、 1（Leader 写入即成功，Leader 宕机可能丢）、all（ISR 全部写入才成功，配合 min.insync.replicas≥2 可做到不丢）。配合幂等生产者 （enable.idempotence=true，broker 按 PID + 序列号去重）与有限重试， 可在"不丢"与"不重"之间取得工程上的最优解：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092,kafka-3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 可靠性三件套：acks=all + 幂等 + 有限重试（单分区幂等有序）
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("retries", "100");
props.put("max.in.flight.requests.per.connection", "5");   // 幂等开启时 ≤5 即可保序

// 吞吐三件套：批量 + 压缩 + 攒批等待
props.put("batch.size", 131072);
props.put("linger.ms", "20");
props.put("compression.type", "lz4");
props.put("buffer.memory", 67108864);

// 跨事务幂等（如按 order_id 建补偿）由下游保证，Kafka 只保证单分区不重不丢
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("ods.shop.orders", orderId, payload),
              (meta, e) -> { if (e != null) e.printStackTrace(); });

容量规划：分区数估算公式

分区不是越多越好（更多分区意味着更多文件句柄、更久的 leader 切换与 rebalance、更大的 controller 元数据压力），规划的核心公式只有一个：

分区数 = 目标峰值吞吐 ÷ 单分区预期吞吐 × 预留系数

其中：
- 单分区生产吞吐经验值：10 MB/s（顺序写、页缓存充裕时的保守估计）
- 预留系数：1.5 ~ 2.0（业务增长 + 大促脉冲 + 故障重放）

数值示例：业务库 binlog 解析后峰值 100 MB/s，规划 2 倍冗余：

目标峰值吞吐 = 100 MB/s × 2 = 200 MB/s
分区数       = 200 ÷ 10 = 20 → 向上取整到 broker 数的整数倍 → 24 个分区

校验消费端：24 分区 × 每消费者并行处理 8 MB/s = 192 MB/s ≈ 达标
校验副本流量：200 MB/s × 3 副本 = 600 MB/s 集群写入带宽
  → 8 台 broker 均摊 75 MB/s/台，万兆网卡余量充足，千兆网卡会成为瓶颈

retention 策略

• 按时间（log.retention.hours）：CDC 主干 topic 常用 24~72 小时—— 留够"消费端宕机 + 故障排查"的时间窗，又不至于磁盘被历史数据占满。
• 按大小（log.retention.bytes）：与时间上限叠加使用，防御流量突发把磁盘写爆。
• 日志压缩（compact）：对"每个 key 只保留最新值"的留痕 topic（如维表快照、位点对账） 使用 cleanup.policy=compact，可无限期保留 key 的最新状态。
• 与 Iceberg 的关系：Kafka 只是缓冲主干而非数据归档层， 湖仓才是唯一的事实源（Source of Truth）——retention 到期删除不影响数据完整性， 因为数据已提交至 Iceberg 快照。

Flink CDC：无锁数据捕获

无锁快照 + binlog 原理

Flink CDC 底层复用了 Debezium 的 binlog 解析能力，但把"采集"与"计算"合并在 Flink 引擎内完成。 其革命性在于增量快照算法（无锁快照），读取存量数据不再需要锁表：

1. 分块（Chunking）：按主键把表逻辑切分为多个 chunk（如每块 8 万行）；
2. 并发快照：多个 chunk 并行读取，读的是 MVCC 一致性视图（REPEATABLE READ）， 不加全局锁、不阻塞业务写入；
3. 高低水位（high-low watermark）：每块快照前后各记录一次当前 binlog 位点， 快照结束后回放这段时间的 binlog 增量，并按 op_ts 与快照数据合并去重；
4. 无缝切换：全部分块完成后，source 优雅切换为纯 binlog 订阅模式，全程不断流。

与早期 CDC 工具的对比：

Exactly-Once 到 Iceberg：端到端语义拆解

端到端链路 MySQL → Flink CDC → Kafka → Flink → Iceberg 上， "精确一次"要逐段拆解，工程上有两条路线：

• 路线一：两阶段提交（2PC / XA）：Kafka 事务（transactional.id + isolation.level=read_committed）+ Flink checkpoint 两阶段提交协议 + Iceberg sink 在 checkpoint 时提交快照。语义最强，但吞吐受事务协调开销拖累，链路任何一环抖动都会放大延迟。
• 路线二：at-least-once + 幂等写入（推荐）：各段允许重复投递， 依赖 Iceberg 的 upsert（v2 表 equality delete）或主键去重语义把重复记录在提交时合并掉。 吞吐高、实现简单，最终状态精确一次（effectively-once）。

断点续传与位点管理

Flink 的 checkpoint 是位点管理的载体：每次 checkpoint 把各 source 的消费位点 （Kafka 分区 offset / binlog 文件名+位置）连同算子状态（维表缓存、聚合中间态）一起持久化到状态后端。 作业故障重启后从最近一次成功的 checkpoint 恢复，数据既不丢也不重算：

• Kafka source：offset 存在 checkpoint 状态中（而非依赖 consumer group 提交）， scan.startup.mode=group-offsets 仅决定首次启动位点，之后完全由 checkpoint 接管。
• MySQL CDC source：binlog 位点（gtid 或 file+pos）随 chunk 进度一起写入状态， 增量快照的分块进度也可恢复，全量阶段崩溃重启后从断点 chunk 继续，无需重新全量。
• Iceberg sink：数据文件先落对象存储，快照提交与 checkpoint 对齐—— 未确认的提交在恢复后按同名文件幂等覆盖，保证"一个 checkpoint 周期的数据恰好一次可见"。

# 提交作业：开启 checkpoint、RocksDB 状态后端、保留多点恢复
$FLINK_HOME/bin/sql-client.sh \
  -f sql/ods_orders_to_iceberg.sql \
  -Dexecution.checkpointing.interval=60s \
  -Dstate.backend=rocksdb \
  -Dstate.checkpoints.dir=s3://lake/flink/ckpt/ods_orders \
  -Dstate.checkpoints.num-retained=10 \
  -Drestart-strategy=fixed-delay,attempts=10,delay=30s

# 故障后从指定 checkpoint 恢复（断点续传）
$FLINK_HOME/bin/flink run -s s3://lake/flink/ckpt/ods_orders/chk-5832 ...

完整 Flink SQL 作业：mysql-cdc → 维表关联 → Iceberg

-- ============================================================
-- 作业：shop.orders CDC 实时入湖（含维表打宽），目标 Iceberg ODS/DWD
-- ============================================================
SET 'execution.checkpointing.interval' = '60s';
SET 'execution.checkpointing.min-pause' = '30s';
SET 'table.exec.sink.upsert-materialize' = 'AUTO';
SET 'pipeline.auto-watermark-interval' = '3s';

-- 1. 源表：MySQL CDC（增量快照无锁读取存量 + binlog 增量）
CREATE TABLE ods_orders_src (
    order_id   BIGINT,
    user_id    BIGINT,
    sku_id     BIGINT,
    amount     DECIMAL(12, 2),
    status     TINYINT,
    pay_time   TIMESTAMP(3),
    proc_time  AS PROCTIME(),                          -- 处理时间：维表 lookup 用
    op_ts      TIMESTAMP(3) METADATA FROM 'op_ts' VIRTUAL,  -- 变更发生时间
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql-master.shop.svc',
    'port' = '3306',
    'username' = 'flink_cdc',
    'password' = '******',
    'database-name' = 'shop',
    'table-name' = 'orders',
    'scan.incremental.snapshot.enabled' = 'true',       -- 无锁增量快照
    'scan.incremental.snapshot.chunk.size' = '8096',    -- 按主键分块，控制单块大小
    'scan.startup.mode' = 'initial',                    -- 存量 + 增量
    'server-time-zone' = 'Asia/Shanghai'
);

-- 2. 维表：商品维表 lookup JOIN（缓存 1 万行 / 60 秒）
CREATE TABLE dim_sku (
    sku_id    BIGINT,
    sku_name  STRING,
    category  STRING,
    PRIMARY KEY (sku_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://mysql-master:3306/shop',
    'table-name' = 'sku',
    'username' = 'flink_cdc',
    'password' = '******',
    'lookup.cache.max-rows' = '10000',
    'lookup.cache.ttl' = '60s'
);

-- 3. 目标表：Iceberg（v2 + upsert，写入与 checkpoint 对齐提交）
CREATE TABLE dwd_orders_lake (
    order_id   BIGINT,
    user_id    BIGINT,
    sku_id     BIGINT,
    sku_name   STRING,
    category   STRING,
    amount     DECIMAL(12, 2),
    status     TINYINT,
    pay_time   TIMESTAMP(3),
    dt         STRING,
    PRIMARY KEY (order_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'lakehouse',
    'catalog-type' = 'hive',
    'uri' = 'thrift://hive-meta:9083',
    'warehouse' = 's3://lake/warehouse',
    'format-version' = '2',                             -- v2 支持 equality delete（upsert）
    'write.format' = 'parquet',
    'write.upsert.enabled' = 'true',                    -- 主键 upsert：CDC 语义核心开关
    'write.target-file-size-bytes' = '134217728',       -- 目标文件 128MB，抑制小文件
    'write.metadata.metrics.default' = 'truncate(16)'   -- 列统计，供下游查询文件跳过
);

-- 4. 消费写入：CDC 变更流 + 维表打宽 → 按天分区入湖
INSERT INTO dwd_orders_lake
SELECT o.order_id, o.user_id, o.sku_id,
       s.sku_name, s.category,
       o.amount, o.status, o.pay_time,
       DATE_FORMAT(o.pay_time, 'yyyy-MM-dd') AS dt
FROM ods_orders_src AS o
LEFT JOIN dim_sku FOR SYSTEM_TIME AS OF o.proc_time AS s
  ON o.sku_id = s.sku_id;

实时入湖链路架构

分层链路全景

生产级实时入湖链路的标准形态：CDC 采集层解耦业务库，Kafka 主干削峰与多播， Flink 加工层承担清洗与打宽，Iceberg 以快照粒度落地。前一节的"直连"示例适合单表简单场景； 一旦表数量、下游消费者增多，中间的 Kafka 主干就不可或缺：

• 采集层（Flink CDC）：每库一个采集作业，按表拆分到算子并行读取， 统一写入 Kafka（按主键 key 分区，保证同主键变更有序）。
• 主干层（Kafka）：以 ODS 命名规范组织 topic（如 ods.shop.orders）， 72h retention 兜底故障重放；schema 演进通过 Schema Registry（Avro/Protobuf）协同。
• 加工层（Flink 消费）：从 Kafka group-offsets 起读，完成清洗、维表 lookup JOIN、 轻度聚合，写入 Iceberg DWD/DWS 分区表。
• 落地层（Iceberg）：v2 表 + upsert，快照与 checkpoint 对齐； ODS 保留原始 CDC 事件（含 before/after），DWD 是合并后的最新状态——事件与状态分离，重算有据可依。

延迟与吞吐的权衡

乱序与回撤流处理

• 乱序（out-of-order）：多分区消费、上游重放、网络抖动都会让事件乱序到达。 窗口聚合用事件时间 + Watermark 对齐： WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND； 超窗的迟到数据走侧输出（side output）单独落一张"迟到事实表"，由离线任务次日修正，避免无限等窗。
• 回撤流（retract）：CDC 更新会让下游聚合结果失效，Flink 以 +I/-U/+U/-D 变更日志语义向下游传播"撤回旧值、下发新值"。 目标表必须支持 upsert（Iceberg v2 equality delete / Doris Unique 表）， 纯 append 的 sink（如 Kafka append-only topic）会因不支持回撤而报错。
• 端到端顺序：按主键 key 写 Kafka 同分区，保证同一订单的变更序列不乱， 这是 upsert 幂等正确性的前提。

Schema Evolution：新加列的上下游协同

上游业务库加列（如 orders 新增 coupon_id）时，链路各环节需按序协同：

1. 业务库：执行向后兼容的 DDL——ADD COLUMN coupon_id BIGINT DEFAULT NULL， 避免直接 rename/not null 这类破坏性变更；
2. Schema Registry：注册新 schema 版本，校验兼容性（向后兼容）， Kafka 消息按新 schema 写入，老消费者仍可跳过未知字段；
3. Iceberg：同步执行 ALTER TABLE ... ADD COLUMN（元数据操作，秒级完成， 不重写数据文件）；
4. Flink 作业：更新 SELECT 列表与目标表定义，从 savepoint 重启 恢复位点与状态，历史数据新列自动补 NULL；
5. 下游消费：BI/查询引擎通过 catalog 刷新感知新列，报表按需纳管。

-- 湖仓侧：加列（元数据操作，不重写文件）
ALTER TABLE lakehouse.dwd.dwd_orders
ADD COLUMN coupon_id BIGINT AFTER amount;

-- Flink 作业侧：改 SELECT 后从 savepoint 原地重启，位点与状态无损迁移
-- bin/flink stop --savepointPath s3://lake/flink/sp/ dwd_orders_job && 重启

批式集成

并非所有数据都值得上实时链路。批式通道以"定时全量/增量抽取 + 批量写湖"的方式， 用更低的复杂度覆盖了大部分数据源，是湖仓数据供给的基座：

DataX / Sqoop 的定位

• DataX：阿里开源的离线异构数据同步工具，插件化支持 MySQL/Oracle ↔ HDFS/各类 RDBMS，单机多线程跑批，适合中小规模、无 Spark 集群团队的定点同步。 定位是"数据库 ↔ 数据库/文件"的点对点搬运工，不承担计算。
• Sqoop：Hadoop 时代的 RDBMS ↔ HDFS/Hive 批量导入工具（基于 MapReduce）， 已停止社区演进，新项目建议用 Spark JDBC / Flink JDBC 批模式 / DataX 替代。
• 选型口径：湖仓体系内优先 Spark 批作业（与计算层同栈、可直接写 Iceberg、 支持复杂转换）；异构库到库的小规模搬运可用 DataX 补位。

-- Spark 批同步：每日快照写 Iceberg 天分区（overwrite 指定分区，幂等重跑）
spark-sql \
  --conf spark.sql.sources.partitionOverwriteMode=dynamic \
  -e "
INSERT OVERWRITE lake.ods.sku_snapshot PARTITION (dt = '2026-11-04')
SELECT sku_id, sku_name, category, price
FROM jdbc_src.sku
WHERE updated_at >= '2026-11-03';"

容量规划方法

容量规划是把"业务流量"翻译成"中间件与集群资源"的系统方法，四步闭环：

步骤一：采集峰值流量

从业务库监控取 binlog 生成速率的 P99 峰值（而非平均值），并统计表数量与单表大小， 得到入口流量画像。示例采集结论：平均 15 MB/s，大促峰值 60 MB/s，4 张核心大表。

步骤二：确定预留倍数

在峰值之上再预留增长与缓冲：×3 是常见工程系数 （1 倍当前峰值 + 1 倍年度增长 + 1 倍故障重放/回溯）。 本例目标吞吐 = 60 MB/s × 3 = 180 MB/s。

步骤三：反推分区与资源

Kafka 分区数  = 180 ÷ 10（单分区 10 MB/s）= 18 → 取 24（broker 数整数倍）
Kafka 副本带宽 = 180 × 3 = 540 MB/s（集群口径），8 broker 均摊 67.5 MB/s → 万兆达标
Kafka 磁盘    = 180 MB/s × 72h × 3 副本 ÷ 4(lz4) ≈ 33 TB → 每台 4.2 TB × 1.3 余量 ≈ 5.5 TB

Flink source 并行度 = 24（= 分区数，再大无增益）
Flink 处理能力      = 每 parallelism 6 MB/s → 24 × 6 = 144 MB/s
                     -- 若需扛满 180 MB/s，并行度提到 32（Kafka 分区同步扩）
Flink 资源          = 每 parallelism 1 core + 2 GB 堆 → 约 32 core / 64 GB（含 JM 冗余）

Iceberg 小文件     = 60s checkpoint × 32 并行度 ≈ 每分钟 32~64 个文件
                     → 每日 4.6~9.2 万文件 → 必须规划每小时 compaction 作业

步骤四：压测验证

1. Kafka：kafka-producer-perf-test 以 180 MB/s 打流 1 小时， 观察 broker 延迟（request latency）与 ISR 稳定性；
2. 端到端：录制真实 binlog 回放至 2 倍峰值，观察 Flink 反压（backpressure）、 checkpoint 时长（应 < 间隔的 1/3，即 < 20s）与 consumer lag 是否归零；
3. 落地：检查 Iceberg 提交延迟、manifest 数量与查询引擎的文件跳过效率， 确认 compaction 节奏跟得上；
4. 留档：把压测结论固化为容量基线，每季度或大促前复测。

适用边界

何时直接 CDC 入湖，而非经 Kafka

• 表少且下游单一：只有 3~5 张表、唯一消费方就是湖仓 ODS 时， Flink CDC 直写 Iceberg 少一跳，延迟更低、成本更小，Kafka 主干的价值（多订阅/重放）无从体现。
• 延迟极度敏感：秒级可见的监控类指标，多一跳 Kafka（攒批 + 消费拉取） 会引入不可控的排队延迟，直连链路更可控。
• 团队无 Kafka 运维能力：小团队维护一套分布式消息系统的成本 （分区规划、ISR 告警、磁盘水位）可能高于其收益。

何时不该上实时链路

• 时效没有消费方：报表次日更新即可满足业务，实时化只产出成本不产出价值—— 先有实时消费场景，再有实时链路，顺序不可颠倒。
• 上游是"批量重算"型：数据每天全量修正、频繁回刷历史， CDC 的行级增量反而放大写入量；这类源用批式快照更干净。
• 逻辑重而变更轻：变更量不大但每次都要多表复杂 JOIN 与重算， 实时链路的状态与回撤处理会让作业脆弱难维护，交给离线批更稳。
• 质量要求人工把关：财务对账类数据需要人工审核后发布， 强行实时化会绕过审核节点，制造合规风险。