流处理与实时管道：从 Kafka 到 Flink Exactly-Once

0. 元信息

• 主题路径：docs/topics/data-engineering-basics/subtopics/streaming-and-realtime/README.md
• 父主题：data-engineering-basics
• 适合对象：会写 Python/SQL、希望从”批处理思维”过渡到”实时数据流”的工程师
• 建议周期：2.5～3 周，每周 10～14 小时
• 前置知识：SQL、Python、Linux；建议先完成父主题第 3 阶段
• 最终目标：能为给定业务（点击流/订单/IoT）设计一条 Kafka → Flink → Iceberg 的实时管道，掌握 Exactly-Once、Watermark、State Backend、Window、迟到数据处理

1. 学习路线

Kafka 基础（broker / topic / partition / offset）→ Exactly-Once 语义 → Flink 架构（JobManager / TaskManager）→ DataStream API → Watermark 与 Window → State Backend（RocksDB）→ 两阶段提交 Sink → 流批一体 SQL

2. 阶段周数分配

每天 1.5～2 小时。2.5 周方案聚焦核心 5 阶段；3 周方案多 2 天做 EOS 与流批一体。

3. 核心知识 / 产出 / 标准表

4. 第一周（每天 1.5~2 小时）

Day 1 约定：本计划使用 Docker Compose 起 Kafka（3 broker）+ Flink（Standalone）+ Iceberg on MinIO。Kafka 测试用 kafka-console-producer / kafka-console-consumer；Flink 跑 flink run；SQL 客户端 sql-client.sh。所有 Job 必须可重跑：清空 topic + 重置 offset 后跑出相同结果。

Day 7 执行次序

步骤 A —— 最小可用流管道（60~90 分钟）

1. kafka-topics --create --topic orders；
2. Producer 写 1 万条订单到 Kafka；
3. Flink Job：Kafka Source → keyBy(user_id) → TumblingEventTimeWindow(5min) → reduce(SumAmount) → IcebergSink；
4. SELECT * FROM iceberg.dws_user_order_5min 查汇总结果；
5. 端到端 P95 < 30s。

步骤 B —— 4 类边界用例（30~45 分钟）

Day 7 当天必完成步骤 A；步骤 B 至少完成 B1、B3。

5. 阶段通用验收

1. 不看答案独立重写一条 Kafka → Flink → Iceberg 的端到端管道 + 一次 EOS 验证；
2. 用自己的话解释”为什么流处理需要 Watermark""为什么 Flink 与 Kafka 各自管一份 EOS""为什么 State 必须可序列化”；
3. 画一张图：流处理架构图 + Watermark 时间线 + State 后端选型决策树（三件套之一）；
4. 测试乱序、迟到、节点宕机、Schema 漂移、状态膨胀 5 类边界；
5. 准备至少 3 组自定义事件流（正常 / 乱序 / 迟到）并贴出实际输出与端到端 P95；
6. 记录端到端 P50 / P99 延迟、吞吐（条/秒）、checkpoint 时长、state 大小 4 个核心指标；
7. 能修改已有 Flink Job（加 Window、改 Watermark 策略、换 State 后端）并用同样指标验证。

交付存放：第 3 项的图、第 5 项的输出、第 6 项的指标，统一存到 week1/notes/ 或 week1/<job>/README.md。

6. 最终验收

• 独立设计并交付一条端到端实时管道（Kafka → Flink → Iceberg），覆盖 EOS、Watermark、State、Window、迟到处理；

• 至少完成 18 个实战项（分布建议）：

  子阶段	题目数量	难度	平台建议
  Kafka 基础与 EOS	3	Easy / Medium	Confluent Quickstart
  Flink DataStream API	4	Medium	Flink Forward Talks
  Watermark 与 Window	4	Medium / Hard	Streaming Systems 案例
  State Backend 与容错	4	Medium	Flink 官方训练
  流批一体 SQL	3	Hard	Flink SQL + Iceberg 案例

  约束：至少 9 项达到 Medium，至少 3 项达到 Hard；每项必须留可运行 Job + checkpoint 恢复证据。

• 完成 1 个综合项目：CDC + 实时 dashboard（Debezium → Kafka → Flink → Iceberg → Superset）；

• 能用 15 分钟讲清 Flink / Kafka 各自 EOS 实现、Watermark 与迟到数据处理、State 后端选型、checkpoint 与 savepoint 区别。

7. 综合项目

首选：CDC + 实时 dashboard（必做：MySQL → Kafka → Flink → Iceberg → Superset）。
备选：点击流实时管道（Nginx → Kafka → Flink → Iceberg → 实时 UV/PV 看板）。
备选：IoT 设备告警（MQTT → Kafka → Flink → Iceberg + 规则告警）。

CDC + 实时 dashboard 必做要求：

• 输入：MySQL orders 表（开启 binlog row 模式）+ 模拟写入工具（atlas / 自写脚本）；
• 输出：必输出（1）Debezium 抓取 binlog 到 Kafka topic cdc.orders；（2）Flink Job 消费 cdc.orders → 写 Iceberg DWD + DWS（含 5 分钟汇总）；（3）Superset 看板：实时订单量 / GMV / 转化率；（4）一次故障演练：故意延迟 1 小时、kill TaskManager、Schema 加列；
• 算法 / 工程：用 flink-connector-debezium + IcebergSink；用 WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))；Checkpoint 60 秒 + EOS；
• 进阶可选：用 StarRocks / ClickHouse 替代 Superset 做 OLAP；接 OpenLineage 输出血缘事件。

任何综合项目都必须包含：

1. 需求说明与数据契约（源 schema、目标 schema、SLA、迟到容忍）；
2. 流处理架构图与命名规范；
3. 核心代码（Debezium 配置 + Flink Job + Iceberg DDL + Superset 看板配置）；
4. 边界测试（broker 宕机、TaskManager OOM、迟到 1 小时、Schema 漂移、状态膨胀）；
5. 可观测（Prometheus 抓 Flink metrics + Grafana 看板）；
6. README（设计取舍、已知限制、下一步）；
7. 复盘记录（模拟 1 次 P0 故障，写 retrospective.md）；
8. notes/ 规范：

本主题贡献

实时管道的核心难点是”消息天然会丢会重 vs 业务要求 exactly-once”。本子主题专门讲清 Kafka → Flink 如何用两阶段提交 + 幂等 Producer + Watermark 把这条边界守住——以 Kafka 为消息总线，Flink 为计算引擎，KStream / DataStream API 为编程抽象，watermark 与 checkpoint 双驱动 exactly-once；不重复父主题的 Airflow 编排与 Kafka 基础。

3 职责

1. 用 Kafka 事务 Producer（transactional.id + enable.idempotence=true + acks=all）保证”发出去的消息要么全成功要么全失败”；Flink Kafka Source 用 setStartOffsets(OffsetsInitializer.committedOffsets()) + isolation.level=read_committed 只消费已提交消息。
2. 用 Flink Watermark（WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))）+ TumblingEventTimeWindow + Side Output 处理乱序与迟到；两阶段提交 Sink（TwoPhaseCommitSinkFunction + Iceberg IcebergSink）保证端到端 exactly-once。
3. 用 Flink Checkpoint（60 秒，CheckpointingMode.EXACTLY_ONCE）+ EmbeddedRocksDBStateBackend 把状态做可恢复；kill TaskManager 后从最近 checkpoint 续跑，Kafka offset 与 RocksDB state 双一致。

4 交付物

1. 一条 Kafka → Flink → Iceberg 的端到端管道（FlinkKafkaConsumer + keyBy(user_id) + TumblingEventTimeWindow(5min) + IcebergSink），含 Watermark 策略与迟到容忍 5 秒。
2. 一份 exactly-once 验证脚本（事务 Producer 抛异常 → abort_transaction → 消费侧看不到部分消息；kill TaskManager → 自动重启 → offset 与 state 完全一致），含 checkpoint 恢复证据目录。
3. 一份 RocksDB State Backend 配置（state.backend=rocksdb + state.checkpoints.dir=s3://flink/checkpoints/），跑 1GB 状态不 OOM，checkpoint 时长 < 30 秒。
4. 一份 Flink metrics + Prometheus 抓取配置（latencyMarker / numRecordsIn / numRecordsOut / checkpointDuration），4 个核心指标基线：端到端 P50/P99 延迟 / 吞吐（条/秒） / checkpoint 时长 / state 大小。

3 指标

1. exactly-once 验证 0 重复 0 丢失（事务 abort 后消费侧消息数为 0；checkpoint 恢复后 Kafka offset 与 RocksDB state 完全一致）。
2. 端到端 P99 延迟 < 30 s（含 Kafka produce + Flink 处理 + Iceberg commit）。
3. Checkpoint 时长 < 30 s（1GB 状态 + RocksDB 增量 checkpoint）。

8. 推荐开源资料

许可证提示：Kafka、Flink、Iceberg 都是 Apache-2.0；Confluent 部分组件（ksqlDB、Schema Registry）有社区版与商业版之分，使用前查 LICENSE。默认做法是读思路后自己实现，不要直接复制 Ververica 商业版代码到生产。

默认使用顺序：先读 Akidau《Streaming Systems》第 13 章建立”流处理心智模型” → 装 Kafka 跑通 Producer/Consumer → 读 Narkhede 第 14 章 → 跑第一个 Flink Job → 读 Flink 官方文档 concepts → 加 Watermark + Window → 换 RocksDB State 后端 → 接 Iceberg Sink → 用 EXPLAIN 看 Flink SQL 计划 → 对比批流同一查询结果 → 写复盘到 notes/retrospective.md。

9. 学习资料汇聚（v0.3 自包含）

本节由本计划生成。链接指向原始材料或作者公开内容。规范会演进，记录时务必写明版本与日期。

9.1 背景与动机

Lambda 架构 2011 年由 Nathan Marz 提出，用”批处理层 + 速度层”双路处理实时数据，但维护两套代码。2014 年 Jay Kreps 提出 Kappa 架构，主张用单一日志流处理一切。Google 2015 年发表 Dataflow 论文（VLDB 2015），把 Watermark、Trigger、Accumulation 三大概念写进流处理理论。Apache Flink 2014 年从柏林大学开源，定位为”原生流处理引擎”；Spark Streaming 2013 年用微批（micro-batch）模拟流；Kafka 2011 年由 LinkedIn 开源，今天是事实标准消息层。三者共同构成实时数据流的事实标准技术栈。

9.2 概念地图

flowchart LR
  Src[源 DB/日志/IoT] --> K[Kafka topic]
  K --> F[Flink Job]
  K --> SS[Spark Streaming]
  F --> S1[Window]
  F --> S2[State]
  F --> S3[Watermark]
  F --> Sink[Sink]
  Sink --> Ice[Iceberg]
  Sink --> K2[Kafka topic]
  Sink --> DB[PG/MySQL]
  F --> Checkpoint[Checkpoint]
  F --> EOS[Exactly-Once]

9.3 基础知识讲解

9.3.1 经典论文

9.3.2 经典书籍

9.3.3 优秀博客与文档

9.3.4 核心人物

9.3.5 开发方法

9.4 经典问题与经典案例（≥5 道）

9.5 学习难点

概念难点

思维难点

工程难点

| 难点 | 突破路径 | |---|---|---| | RocksDB 调优 | 调 write buffer、block cache | | Checkpoint 失败 | 检查 state 大小、磁盘、网络 | | 反压 | 调 buffer pool、并行度 |

9.6 技术标准与接口

Entity

Scope

Kafka 是消息层，Flink/Spark 是流处理引擎；Pulsar 是消息+存储一体；Beam 是多引擎抽象。

Structure

• Kafka：Producer → Broker → Topic → Partition → Consumer Group。
• Flink：JobManager + TaskManager + Slot + Operator Chain + Checkpoint + Savepoint。
• Spark Streaming：DStream → Batch Interval → RDD/Dataset。

Ecosystem

• 消息：Kafka、Pulsar、NATS、RabbitMQ、Redis Streams。
• 流处理：Flink、Spark Streaming、Beam、ksqlDB。
• Sink：Iceberg、Delta、Hudi、PostgreSQL、Elasticsearch、Kafka。
• Schema：Schema Registry、Avro、Protobuf。

Depth Tiers

本子主题目标：L3。

Source

• Apache Kafka 官方文档：引用快照 2026-07-30。
• Apache Flink 官方文档：引用快照 2026-07-30。
• Streaming Systems：引用快照 2026-07-30。

9.7 关键代码

9.7.1 Kafka 幂等 Producer + 事务

# 9.7.1 Kafka 事务消息（Python confluent-kafka）
from confluent_kafka import Producer, Consumer
from confluent_kafka.admin import AdminClient

producer = Producer({
    "bootstrap.servers": "kafka:9092",
    "enable.idempotence": True,
    "transactional.id": "tx-orders-1",
    "acks": "all",
})

producer.init_transactions()
producer.begin_transaction()
try:
    for order in orders:
        producer.produce("orders", key=str(order["id"]), value=json.dumps(order))
    producer.commit_transaction()
except Exception:
    producer.abort_transaction()
    raise

9.7.2 Flink DataStream Job（含 Watermark）

// 9.7.2 Flink Java DataStream：Kafka -> Iceberg with Watermark
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE);

DataStream<Order> orders = env
    .addSource(new FlinkKafkaConsumer<>("orders", new OrderDeserializer(), kafkaProps))
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((o, ts) -> o.getEventTime())
    );

orders
    .keyBy(Order::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .reduce(new SumAmount())
    .addSink(new IcebergSink.Builder<OrderSummary>()
        .setTable(TableIdentifier.of("dw", "dws_user_order_5min"))
        .setUpsert(true)
        .build());

env.execute("orders-windowed-aggregation");

9.7.3 Flink SQL：流批一体查询

-- 9.7.3 Flink SQL：Kafka 表 + Iceberg 表 JOIN（流批一体）
CREATE TABLE kafka_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10, 2),
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'format' = 'json',
    'properties.bootstrap.servers' = 'kafka:9092'
);

CREATE TABLE iceberg_dwd_orders (...) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'iceberg_catalog',
    'database-name' = 'dw'
);

INSERT INTO iceberg_dwd_orders
SELECT * FROM kafka_orders WHERE amount > 0;

10. 常见误区

• 流处理当批处理写（无 Watermark）；
• Kafka 用 acks=0 当 Exactly-Once；
• Flink 用 HashMapState 处理大状态 OOM；
• 迟到数据不配 Allowed Lateness 直接丢；
• 双流 Join 不设状态 TTL，状态无限膨胀；
• Checkpoint 间隔设太短（< 1 分钟），影响吞吐；
• 反压靠加机器不调并行度；
• Flink Job 不设 RestartStrategy；
• Savepoint 与 Checkpoint 混用；
• Kafka topic 分区数拍脑袋；
• Stream SQL 当流式 ETL 写，不设状态；
• 不用 Schema Registry，事件结构漂移。

11. 所有知识点分类

1. 编程语言
2. 数据结构与算法
3. 计算机基础
4. 工程技术
5. Web 与后端
6. 前端与客户端
7. 数据与人工智能
8. 项目与职业能力
9. 安全与可靠性

本计划归属：数据与人工智能 主 + 工程技术 辅。