夜雨聆风核心文件：DataStream.java, KeyedStream.java, KeyGroupRangeAssignment.java
从哪里调用
上一课深入了 flatMap 的 5 层包装。现在你在 words（SingleOutputStreamOperator）上调用 .keyBy(value -> value.f0)。
这一课解决什么问题
words.keyBy(value -> value.f0)上一课的 flatMap 创建了 OneInputTransformation 并加入 DAG。keyBy 呢？它创建了什么？
答案是：几乎什么都没创建。 只返回了一个 KeyedStream 包装器。分区效果延迟到下游的 sum 调用时才生效。这听起来很"懒"，但这种"懒"是经过精心设计的。
第一关：keyBy()——就两行代码
打开 DataStream.java，跳到第 274-277 行：
public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key){ Preconditions.checkNotNull(key);returnnew KeyedStream<>(this, clean(key));}就两行！ 没有 addOperator，没有创建 Transformation，没有修改 env.transformations List。和 flatMap 的 5 层包装相比，keyBy 简直不像个正经方法。
但这就是它最精妙的地方——**"什么都不做"也是有意义的**。
实践环节
在 keyBy() 这行打断点，step into 看 new KeyedStream<>(this, clean(key)) 内部干了什么。你会发现它只是保存了 keySelector 引用。
第二关：KeyedStream 构造函数——继承 DataStream，挂载 keySelector
KeyedStream.java 第 114-118 行：
publicKeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector){this(dataStream, keySelector, TypeExtractor.getKeySelectorTypes(keySelector, dataStream.getType()));}传入的 dataStream 是 words，其 transformation 指向 Tokenizer 的 OneInputTransformation(id=2)。KeyedStream 继承 DataStream，所以 this.transformation 仍然指向 id=2。
关键细节①：从 DAG 的角度看，keyBy 没有创建新节点。KeyedStream 和之前 words 共享同一个 Transformation。区分不过是 KeyedStream 额外记录了 keySelector 和 keyType。
这意味着什么？DAG 图还是 4 个节点（Source + Tokenizer + Counter + Sink），没有 "KeyByNode"。
第三关（关键！）：分区的真正创建处——ReduceTransformation
分区效果在哪里体现？Ctrl/Cmd + 点击keyed.sum(1) → 追到 KeyedStream.reduce() 方法（KeyedStream.java:723-740）：
public SingleOutputStreamOperator<T> reduce(ReduceFunction<T> reducer){ ReduceTransformation<T, KEY> reduce =new ReduceTransformation<>("Keyed Reduce", environment.getParallelism(), transformation, // ← words.transformation (id=2, Tokenizer) clean(reducer), // ← SumAggregator keySelector, // ← value -> value.f0 ★ 在这里传入！ getKeyType(), // ← Stringfalse);// ...}在 ReduceTransformation 内部，因为传入了 keySelector，会自动创建 PartitionTransformation：
// ReduceTransformation 内部逻辑（简化）：if (keySelector != null) {// ★ 创建 PartitionTransformation 包裹父 Transformationthis.input = new PartitionTransformation<>( inputTransformation, // id=2new KeyGroupStreamPartitioner<>(keySelector, // ★ 根据 key 路由 KeyGroupRangeAssignment.DEFAULT_LOWER_BOUND_MAX_PARALLELISM));}关键细节②：PartitionTransformation 不是"DAG 中的一个真正节点"。在 StreamGraph 生成时，它被解析为虚拟节点——边的属性（Partitioner），而不是一个 StreamNode。这就是为什么 keyBy 的效果最终体现在"边"上，而不是"节点"上。
第四关：KeyGroup 路由——hash → KeyGroup → SubtaskIndex
KeyGroupStreamPartitioner 的核心逻辑：
// KeyGroupStreamPartitioner.selectChannel(record):publicintselectChannel(SerializationDelegate<StreamRecord<T>> record){ T element = record.getInstance().getValue(); K key = keySelector.getKey(element); // ① 提取 key// ② key → KeyGroupint keyGroup = KeyGroupRangeAssignment.assignToKeyGroup( key, maxParallelism); // hash(key) % maxParallelism// ③ KeyGroup → SubtaskIndexreturn KeyGroupRangeAssignment.computeOperatorIndexForKeyGroup( maxParallelism, parallelism, keyGroup); // 查表映射}最终的数据路由路径：
key = "hello" → hash("hello") = 99162322 → 99162322 % 128 = 42 (KeyGroup 42) → 如果 parallelism=4, maxParallelism=128: Subtask 0: KeyGroups [0..31] Subtask 1: KeyGroups [32..63] ← KeyGroup 42 在这里 Subtask 2: KeyGroups [64..95] Subtask 3: KeyGroups [96..127] → "hello" 数据去 Subtask 1实践环节
在 selectChannel 方法中打断点，运行 WordCount。看每次调用 keySelector.getKey(element) 返回的 key 值，以及 computeOperatorIndexForKeyGroup 返回的 subtask index。验证 "hello" 是否每次都去同一个 Subtask。
第五关：为什么用 maxParallelism 而不是 parallelism 做取模？
这是 keyBy 设计中最值得你记住的一点。
KeyGroupRangeAssignment.assignToKeyGroup(key, maxParallelism) = MathUtils.murmurHash(key.hashCode()) % maxParallelism;如果 parallelism 做取模：扩容时取模结果会变。比如 parallelism 从 2 变成 4：
"hello".hashCode() % 2 = 0→ Subtask 0扩容后 "hello".hashCode() % 4 = ?→ 可能不是 Subtask 0 了
这会导致所有 key 重新分布，状态需要全部迁移——存量作业升级时是一场灾难。
用 maxParallelism 做取模：KeyGroup 的范围是固定的（[0, maxParallelism)）。扩容只改变 KeyGroup 到 Subtask 的分配方案，不改变 key 到 KeyGroup 的映射。状态只需要按 KeyGroup 范围整体搬迁，不需要逐 key 重新计算。
类比：想象一个拥有 128 个固定编号储物柜的小区（maxParallelism=128）。每个住户的钥匙上刻着储物柜编号。当前有 4 个管理员，每人管 32 个柜子。扩容到 8 个管理员，每人管 16 个柜子。住户的钥匙不需要换——只是管理员的分工变了。这就是 KeyGroup 设计的精妙之处。
数据流可视化
keyBy 之前（ForwardPartitioner，默认）： Tokenizer 实例 0 → 结果直接发给 Sum 实例 0 Tokenizer 实例 1 → 结果直接发给 Sum 实例 1 数据流向：1对1，无重新分布keyBy 之后（KeyGroupStreamPartitioner）： Tokenizer 实例 0 → ("hello",1) → hash→KG42→Sum 实例 1 → ("world",1) → hash→KG78→Sum 实例 2 Tokenizer 实例 1 → ("hello",1) → hash→KG42→Sum 实例 1 → ("flink",1) → hash→KG127→Sum 实例 3 数据流向：按 key 重新分布，同一 key 去同一实例总结
┌─────────────────────────────────────────────────────────────────┐│ keyBy() 核心要点： ││ ││ 1. keyBy 不创建 Transformation，只创建 KeyedStream 包装器 ││ 2. KeyedStream 额外记录 keySelector 和 keyType ││ 3. 分区效果延迟到下游有状态算子（sum/reduce）创建时生效 ││ 4. KeyGroupStreamPartitioner 在 ReduceTransformation 内自动创建 ││ 5. hash(key) % maxParallelism → KeyGroup → SubtaskIndex ││ 6. 用 maxParallelism 取模保证扩容时状态迁移最小化 ││ 7. 这就是为什么 keyBy 之后的算子不能和前驱 Chain ││ → 边上的 Partitioner 从 ForwardPartitioner 变成了 ││ KeyGroupStreamPartitioner，违反了 Chain 的条件 │└─────────────────────────────────────────────────────────────────┘下一课预告
sum(1) 的本质是 reduce(SumAggregator)，创建 ReduceTransformation。这个 Transformation 内部维护着 ValueState<T>——每次累加都从状态中读上一次的结果，计算后再写回状态。这是 Flink 有状态处理的最简例子。