Flink 基础

本章目标：能讲清 Flink 的定位、运行架构和 DataStream 编程模型，能跑一个本地 PyFlink demo，并写出一段 Kafka 入、窗口聚合、print 出的 Flink SQL。 前置知识：读完 Kafka 入门 更顺；会一点 Python 即可。

1. 为什么需要它

1.1 两种数据：有界和无界

• 有界数据：有明确开始和结束，例如一份历史订单 CSV。批处理适合它。
• 无界数据：持续不断产生，没有“全部读完”的时刻，例如 App 埋点、订单支付事件。流处理适合它。

有界：  [==== 一天的订单文件 ====]  → 读完就结束
无界：  ──●──●──●──●──●──●──●──→  永远在来

传统离线数仓是“每天凌晨把昨天的数据算一遍”，结果 T+1 才能看到。实时场景要求“支付成功后几秒内大屏就更新”，必须用流处理。

1.2 批处理和流处理

1.3 Flink 和 Spark Streaming 的区别

• Spark Streaming（老 DStream）：把流切成小批次，本质还是批处理，延迟通常秒级。
• Spark Structured Streaming：更现代的微批模型，也能做流处理，但默认仍是微批。
• Flink：原生流处理，一条一条处理，延迟可以做到毫秒级；同时支持批模式，用同一套 API/SQL 处理有界数据，这叫“流批一体”。

面试时不要说“Spark 不能流处理”。准确说法是：Spark 以微批为主，Flink 以原生流为主；延迟要求不高且已有 Spark 栈时，Structured Streaming 也是合理选择。

1.4 典型场景

实时大屏、实时风控、实时推荐特征、MySQL CDC 同步、分钟级 GMV/DAU、实时告警。数据通常从 Kafka 进来，计算后写 Kafka、Doris、ClickHouse 或消息系统。

2. 核心概念

2.1 集群架构：JobManager、TaskManager、Slot

                ┌────────────────────────────┐
  提交作业 ───→ │ JobManager                 │
                │  Dispatcher                │
                │  JobMaster（每个作业一个）  │
                │  ResourceManager           │
                └─────────────┬──────────────┘
                              │ 调度、分配 Slot
              ┌───────────────┴───────────────┐
              ↓                               ↓
      ┌────────────────┐             ┌────────────────┐
      │ TaskManager 1  │             │ TaskManager 2  │
      │ slot slot slot │             │ slot slot slot │
      │ 运行具体 Task  │             │ 运行具体 Task  │
      └────────────────┘             └────────────────┘

• JobManager：集群大脑。Dispatcher 接收作业，JobMaster 负责一个作业的调度和 checkpoint 协调，ResourceManager 管理 TaskManager 的 slot。
• TaskManager：真正干活的进程，提供 slot 运行 task。
• Slot：资源单位。一个 TaskManager 有若干 slot，slot 数决定这个进程最多同时跑多少并行子任务。

并行度（parallelism） 是一个算子被拆成多少份并行执行。一个作业并行度为 4，表示每个算子有 4 个并行子任务。并行度可以在作业级、算子级分别设置。

slot 共享：默认情况下，同一条流水线上不同算子的子任务可以共享一个 slot。例如 source、map、sink 各自并行度 4，默认只需要 4 个 slot，而不是 12 个。

算子链（operator chain）：并行度相同、数据一对一转发（forward）的相邻算子，可以合并成一个 task 执行，减少线程切换和序列化开销。遇到 keyBy、rebalance 这类需要跨节点重分布的操作，链通常会断开。

面试高频：并行度决定“切几份”，slot 决定“能同时跑几份”，一个 slot 可以运行一条算子链上的多个子任务。

2.2 DataStream API：Source → Transformation → Sink

一个 Flink 程序的基本结构：

env = StreamExecutionEnvironment.get_execution_environment()

ds = env.add_source(...)          # Source：从 Kafka、文件、集合读数据
result = ds.map(...).filter(...)  # Transformation：转换
result.add_sink(...)              # Sink：写出去

env.execute("job name")           # 提交并运行

常见转换：

key_by 会发生 hash shuffle：相同 key 的数据被路由到同一个并行子任务，这样才能做分组聚合。shuffle 有网络和序列化开销，是流作业的主要成本之一。

2.3 时间与窗口（先知道有这么回事）

流数据会乱序，不能按“到达顺序”直接切窗口。Flink 用事件时间表示数据实际发生的时间，用水位线（Watermark） 表示“比某个时间更早的数据基本到齐了”。窗口负责把无限流切成有限块，例如每分钟一个窗口。

这三个概念是实时计算的核心，Flink 进阶 会详细展开；本章只需要知道：没有时间语义和窗口，流处理只能做逐条转换，做不了“每分钟统计”。

2.4 Flink SQL 和 Table API

Flink 也支持用 SQL 描述流任务。Kafka 表、结果表通过 CREATE TABLE ... WITH (...) 定义，查询写法和离线 SQL 很像，但多出了水位线、时间窗口和持续查询的概念。

一个持续查询会一直运行：新数据进来就更新结果。INSERT INTO sink SELECT ... FROM source 提交后不会自己结束。

3. 动手示例

3.1 安装 PyFlink

PyFlink 依赖 Java，建议 Java 11 或 17，Python 用 3.8 到 3.11 之间。先检查：

java -version
python --version
pip install apache-flink

Python 代码通过 StreamExecutionEnvironment.get_execution_environment() 启动本地 mini cluster，不需要单独部署 Flink 集群。

注意：PyFlink 安装包较大，版本对 Java/Python 有要求。两小时装不上就停止，用 Docker 版 Flink 或只看概念；本章验收不依赖你一定能跑通。

3.2 PyFlink DataStream 本地 demo

新建 realtime_demo/flink_stream_demo.py：

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

rows = [
    ("O1001", "手机", 2999.0),
    ("O1002", "图书", 59.0),
    ("O1003", "手机", 1999.0),
    ("O1004", "家电", 399.0),
    ("O1005", "图书", 129.0),
]

ds = env.from_collection(
    rows,
    type_info=Types.ROW([Types.STRING(), Types.STRING(), Types.DOUBLE()]),
)

result = (
    ds.filter(lambda r: r[2] >= 100)
    .map(lambda r: (r[1], r[2]), output_type=Types.TUPLE([Types.STRING(), Types.DOUBLE()]))
    .key_by(lambda x: x[0])
    .reduce(lambda a, b: (a[0], a[1] + b[1]))
)

result.print()
env.execute("category amount demo")

运行：

python realtime_demo/flink_stream_demo.py

你会看到每个品类在收到元素后不断更新的累计金额。from_collection 是有界源，跑完就结束，适合验证语法；真实场景把 source 换成 Kafka。

3.3 Flink SQL：Kafka source → 分钟聚合 → print

假设 Kafka 的 order_events 消息格式如下，ts 是毫秒时间戳：

{"order_id":"O1001","category":"手机","amount":2999.0,"ts":1700000000000}

在 Flink SQL Client 或 PyFlink Table API 中执行：

CREATE TABLE order_events (
  order_id STRING,
  category STRING,
  amount DECIMAL(10, 2),
  ts BIGINT,
  event_time AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'order_events',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'flink_gmv_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

CREATE TABLE print_sink (
  window_start TIMESTAMP_LTZ(3),
  category STRING,
  order_cnt BIGINT,
  gmv DECIMAL(12, 2)
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink
SELECT
  TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
  category,
  COUNT(*) AS order_cnt,
  SUM(amount) AS gmv
FROM order_events
GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE), category;

几个关键点：

• event_time AS TO_TIMESTAMP_LTZ(ts, 3) 把毫秒时间戳变成事件时间；
• WATERMARK ... - INTERVAL '5' SECOND 表示允许 5 秒乱序，水印推进到窗口结束时间后窗口才输出；
• TUMBLE 是滚动窗口，每分钟一个、不重叠；
• print 连接器只用于开发调试，生产要写 Kafka、Doris 等；
• Kafka 连接器 JAR 需要单独加入 Flink 的依赖，版本与 Flink 主版本匹配，否则会报找不到 connector。

提示：本地只验证 SQL 语法时，可以把 source 换成 datagen 连接器或 VALUES 表，不依赖 Kafka。真正上线时再补连接器和 checkpoint 配置。

3.4 提交方式和运行模式

• 本地模式：PyFlink 在 IDE 里直接跑，适合开发和学习。
• Standalone / YARN / Kubernetes：生产常见部署方式，由集群管理资源。
• Flink SQL Client：交互式提交 SQL，适合验证和运维；也可以用 sql-client.sh -f job.sql 批量执行。

本地 demo 不涉及高可用和资源调度，面试时要说清“我跑的是本地 mini cluster，生产部署了解概念”。

4. 常见坑

1. 只设了事件时间，没设水位线。事件时间窗口可能永远不触发，因为 Flink 不知道“什么时候算到齐”。
2. Kafka 从 latest 开始消费，又没有新数据。看起来“任务没输出”，其实在等新消息。
3. 忘了加 Kafka connector JAR。SQL 提交时报 Could not find any factory for identifier 'kafka'。
4. 并行度大于可用 slot。作业一直处于等待资源状态，不会报错退出。
5. key_by 的 key 为空。空 key 无法正常分组，埋点数据要先过滤或补默认值。
6. 把 print 当生产 sink。它把结果打到标准输出，重启就丢，不能给下游用。
7. 没开 checkpoint 就谈故障恢复。没有 checkpoint，作业挂了只能从头重跑，状态也恢复不了。
8. 本地 from_collection 当成真实流。它是有界数据，跑完即结束，不能用来验证长期运行和乱序处理。

5. 练习

练习 1：slot 和并行度

集群有 3 个 TaskManager，每个 4 个 slot。一个作业并行度设为 6，默认 slot 共享，需要几个 slot？还能同时跑第二个并行度为 6 的作业吗？如果两个作业并行度都是 8 呢？

练习 2：哪里发生了 shuffle，哪里能形成算子链

流水线为 source → map → filter → keyBy → window → reduce → sink，并行度相同。指出哪些算子可以链在一起，哪里发生 shuffle。

练习 3：作业没有输出，怎么排查

Kafka 里有数据，Flink SQL 作业也显示 RUNNING，但 print 没有结果。列出至少四个排查方向。

练习 4：设计一个实时品类 GMV 作业

订单事件在 Kafka，要求每分钟按品类统计支付金额，结果写入 Doris。写出作业的输入、处理步骤、时间语义和输出。

练习 5：什么时候用批，什么时候用流

判断下列场景更适合批还是流，并说明理由：每日财务对账、实时风控、月度经营报表、App 崩溃告警。

练习 6：Flink 与 Spark Streaming

用自己的话说明两者的核心区别，并说出各自适合的场景。

6. 面试延伸

面试题：Flink 和 Spark Streaming 的区别？

是什么：Spark Streaming 是微批流处理，Flink 是原生流处理，同时支持批模式。

解决什么问题：低延迟、乱序事件和故障恢复要求高时，原生流更合适；已有 Spark 批处理体系时，微批更容易复用。

我的实操边界：我用 PyFlink 本地跑过 DataStream 和 Flink SQL demo，了解 slot、并行度和算子链；没有在生产集群做过 Flink 作业调优，Spark 侧也只做过离线 SQL 和小规模 PySpark。

面试题：并行度、slot 和算子链是什么关系？

是什么：并行度是算子拆成多少份，slot 是 TaskManager 提供的资源单位，算子链是把可合并的相邻算子放进同一个 task。

解决什么问题：并行度决定吞吐，slot 决定资源上限，算子链减少线程切换和序列化开销。

我的实操边界：我在本地观察过并行度变化对输出文件数和运行方式的影响，理解默认 slot 共享；没有做过大规模 slot 隔离和资源队列配置。

面试题：Flink 怎么保证精确一次？

是什么：Flink 通过 checkpoint 保存状态和输入位置，配合可重放的数据源和事务/幂等 sink，实现端到端精确一次。

解决什么问题：作业失败恢复后，不能重复计算，也不能漏算。

我的实操边界：我理解 checkpoint 和两阶段提交的原理，本地只验证过普通流任务；没有在生产环境配置过 Kafka 事务 sink 和状态后端。详细机制见 Flink 进阶。

7. 本章验收

• 能用自己的话区分有界数据和无界数据，并说出批处理和流处理各适合什么场景。
• 能画出 JobManager、TaskManager、slot 的关系，并解释并行度和 slot 的区别。
• 能说明算子链的作用，以及 keyBy 为什么会打断算子链并产生 shuffle。
• 能独立运行本章的 PyFlink DataStream demo，并改动过滤条件和聚合逻辑。
• 能写出 Kafka source、1 分钟滚动窗口、print sink 的 Flink SQL 骨架。
• 能列出至少四个“作业在跑但没有输出”的排查方向。
• 能回答 Flink 和 Spark Streaming 的核心区别，不把微批说成“不能流处理”。

下一章：Flink 进阶，讲时间、水位线、状态和精确一次。实时链路如何落地见 实时数仓。