企业官网建设流程全解析

• 大数据
• 批处理
• 流处理
• 数据工程

本文围绕 Apache Beam 官方 Java Kata 训练项目中的 FlattenWith 练习展开，讲解如何通过Flatten.with(...)把两个 PCollection 合并成单个 PCollection，并借助源码剖析其底层实现与窗口约束。读完本文，你将掌握 FlattenWith 的链式调用写法、与传统Flatten.pCollections()的差异，以及合并时窗口/触发器兼容性的校验逻辑，可以直接上手完成该 Kata 并理解其设计意图。

一、FlattenWith 是什么

FlattenWith 是 Apache Beam 中Flatten变换家族的一员，核心作用是把多个 PCollection 对象合并成一个逻辑上的单一 PCollection。与经典 Flatten 相比，它的独特之处在于：它允许把产生根 PCollection 的变换（如 Create、Read）与已存在的 PCollection一起参与合并。

在官方文档中（见 task.md），FlattenWith 被定义为：

FlattenWith is a Beam transform that merges multiple PCollection objects into a single logical PCollection. It allows for the combination of both root PCollection-producing transforms (like Create and Read) and existing PCollections.

从这一描述可以提炼出两个关键信息：

1. 结果是一个逻辑统一的 PCollection：合并后元素内容不变，只是汇聚到同一条"管道"中继续流转；
2. 输入来源多样：既可以合并已经构造好的 PCollection，也可以直接把Create、Read这类产生根 PCollection 的变换当作合并对象（Flatten.with(PTransform)重载即为此设计）。

二、Kata 任务拆解

本练习位于 Java 版 Beam Katas 的 Core Transforms 模块，其任务原文为：

Kata:Implement a FlattenWith transform that merges two PCollection of words into a single PCollection, optimized for chaining operations.

即：实现一个 FlattenWith 变换，把两个单词 PCollection 合并为一个 PCollection，并且针对链式调用（chaining）进行优化。

任务元数据（见 task-info.yaml）显示该练习：

• 复杂度级别：BASIC（基础）；
• 分类：Combiners、Flatten、Core Transforms；
• 标签：transforms、join、strings。

输入数据

Kata 提供了两组待合并的单词集合：

• 以 A 开头的单词：apple、ant、arrow
• 以 B 开头的单词：ball、book、bow

核心练习点

练习的关键在于"optimized for chaining operations"——即不要先手动把两个 PCollection 打包成PCollectionList再整体 apply，而是利用Flatten.with(other)返回一个可复用的PTransform对象，把它内联到已有的变换链中间，实现边变换边合并。

三、完整实现与源码走读

3.1 解题实现

Kata 的参考实现位于 Task.java，核心代码如下：

public class Task { public static void main(String[] args) { PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create(); Pipeline pipeline = Pipeline.create(options); PCollection<String> wordsStartingWithA = pipeline.apply("Words starting with A", Create.of("apple", "ant", "arrow")); PCollection<String> wordsStartingWithB = pipeline.apply("Words starting with B", Create.of("ball", "book", "bow")); PCollection<String> output = applyTransform(wordsStartingWithA, wordsStartingWithB); output.apply(Log.ofElements()); pipeline.run(); } static PCollection<String> applyTransform( PCollection<String> words1, PCollection<String> words2) { PTransform<PCollection<String>, PCollection<String>> flattenTransform = Flatten.with(words2); return words1 .apply("Transform A to Uppercase", MapElements.into(TypeDescriptors.strings()) .via((String word) -> word.toUpperCase())) .apply("Flatten with words2", flattenTransform); } }

3.2 逐行解读

第一步：构建两个源 PCollection

使用Create.of(...)分别生成wordsStartingWithA和wordsStartingWithB。Create是 Beam 中最常用的"产生根 PCollection"的变换，负责把内存中的静态数据注入管道。

第二步：构造 FlattenWith 变换对象

PTransform<PCollection<String>, PCollection<String>> flattenTransform = Flatten.with(words2);

Flatten.with(PCollection<T>)返回的是一个PTransform<PCollection<T>, PCollection<T>>。注意这里并没有立即执行合并——PTransform 只是一个"蓝图"，真正执行发生在它被 apply 到某个输入 PCollection 时。因此它可以被保存为变量、按需复用，这正是"针对链式调用优化"的含义。

第三步：链式应用

return words1 .apply("Transform A to Uppercase", MapElements.into(TypeDescriptors.strings()) .via((String word) -> word.toUpperCase())) .apply("Flatten with words2", flattenTransform);

先把words1中每个单词转成大写，再在同一链条的末尾把words2合并进来。这种写法让 Flatten 成为链条中的一环，而不是独立的"收尾动作"。

值得注意的语义细节：因为 Flatten 是作用在"大写转换之后"的结果上，所以只有 words1 的元素会被大写化，words2 中的ball、book、bow保持原样。这一细节正是测试用例要验证的行为（见下文第四节）。

3.3 输出打印

合并结果通过Log.ofElements()打印到日志。Log是 Katas 提供的通用工具（见 Log.java），其内部实现是一个ParDo+DoFn：对每个元素调用LOG.info(message)输出元素内容，若元素所在窗口不是GlobalWindow还会附加窗口信息，随后原样out.output(element)透传。

四、测试验证：合并行为的精确断言

测试代码位于 TaskTest.java：

@Test public void flattenWith() { PCollection<String> wordsStartingWithA = testPipeline.apply("Words starting with A", Create.of("apple", "ant", "arrow")); PCollection<String> wordsStartingWithB = testPipeline.apply("Words starting with B", Create.of("ball", "book", "bow")); PCollection<String> results = Task.applyTransform(wordsStartingWithA, wordsStartingWithB); PAssert.that(results) .containsInAnyOrder("APPLE", "ANT", "ARROW", "ball", "book", "bow"); testPipeline.run().waitUntilFinish(); }

该测试清晰地印证了两个事实：

1. 合并成功：6 个元素全部出现在输出 PCollection 中；
2. 顺序无保证：containsInAnyOrder表明 Flatten 输出的元素不保证顺序——这正是分布式数据处理的常态，也是 Beam 编程模型的重要理念；
3. 变换作用范围：前三个单词为大写（来自被MapElements处理过的 words1），后三个保持小写（来自 words2），精确验证了链式 Flatten 的位置语义。

五、源码级原理：Flatten.with() 到底做了什么

要真正理解 FlattenWith，需要进入 SDK 核心实现 Flatten.java 一探究竟。

5.1 with(PCollection) 的等价实现

Flatten.with(PCollection<T> other)工厂方法（Flatten.java#L102-L104）返回一个内部类FlattenWithPCollection，其expand方法（Flatten.java#L116-L119）只有一行核心逻辑：

@Override public PCollection<T> expand(PCollection<T> input) { return PCollectionList.of(input).and(other).apply(pCollections()); }

这意味着Flatten.with(other)在功能上完全等价于"把 input 与 other 组成 PCollectionList 再套用Flatten.pCollections()"，差异仅仅在于：前者可以作为链上的一环内联使用，而后者需要先把集合打包。源码注释也明确指出：

This is equivalent to creating a PCollectionList containing both the input andotherand then applying pCollections(), but has the advantage that it can be more easily used inline.

getKindString()返回"Flatten.With"，用于调试与命名时的区分。

5.2 with(PTransform) 重载：合并根变换输出

除PCollection重载外，Flatten还提供了with(PTransform<PBegin, PCollection<T>> other)重载（Flatten.java#L144-L159）：

public static <T> PTransform<PCollection<T>, PCollection<T>> with( PTransform<PBegin, PCollection<T>> other) { return new PTransform<PCollection<T>, PCollection<T>>() { @Override public PCollection<T> expand(PCollection<T> input) { return PCollectionList.of(input) .and(input.getPipeline().apply(other)) .apply(pCollections()); } ... }; }

这个重载正是 task.md 中"combine both root PCollection-producing transforms (like Create and Read) and existing PCollections"的实现基础：它先把other（一个Create、Read等变换）apply 到管道上产生新的 PCollection，再与输入合并。这意味着你可以写出形如words.apply(Flatten.with(Create.of("newWord")))的代码，把"从无到有"的变换与既有集合一步合并。

5.3 合并时的窗口与触发器校验

PCollections.expand在构造输出 PCollection 时执行了一系列构造期校验（Flatten.java#L173-L207）：

• WindowFn 兼容性：遍历所有输入，若任一输入的WindowFn与第一个不兼容，抛出IllegalStateException："Inputs to Flatten had incompatible window windowFns"；
• Trigger 兼容性：若任一输入的触发器与第一个不兼容，同样抛出IllegalStateException："Inputs to Flatten had incompatible triggers"；
• 有界性聚合：输出 PCollection 的IsBounded由所有输入的isBounded做 AND 运算得出——即只要有一个输入是无界流，输出就是无界的；
• Coder 继承：输出使用第一个输入 PCollection 的 Coder；若输入列表为空，则 Coder 保持未指定。

5.4 窗口与时间戳语义

合并产生的输出元素保留原输入元素所属的窗口与时间戳，输出 PCollection 的 WindowFn 与所有输入一致。这在流式场景中非常重要：Flatten 不会重组窗口边界，只是把多条流的元素汇合到同一处理逻辑中。

六、FlattenWith 与传统 Flatten 的对比

可以这样理解：FlattenWith 是传统 Flatten 的"链式友好"语法糖，其底层与pCollections()完全同源（源码见 Flatten.java）。当需要合并 3 个及以上 PCollection 时，依然建议使用PCollectionList.of(pc1).and(pc2).and(pc3).apply(Flatten.pCollections())。

七、运行方式与学习环境

本 Kata 属于 learning/katas/java 训练项目的一部分。按照其 README 的指引，推荐以下方式搭建运行环境：

1. 使用 IntelliJ IDEA 的 Education 版本（或安装 EduTools 插件），选择"Open"打开learning/katas/java目录；
2. 在弹出的提示中选择"Import Gradle project"，配置 Gradle；
3. 等待 Gradle 构建完成；
4. 在"Project Structure"中设置项目 SDK（例如 JDK 8）；
5. 在"Project"工具窗口切换到"Course"视图，即可看到 FlattenWith 等各课题目录，完成填空并运行测试验证。

由于FlattenWith的PTransform支持被保存与复用，你也可以把它封装成工具方法，在真实业务管道中实现"两路数据源汇合后再统一处理"的模式。完成本练习后，建议继续尝试 Core Transforms 模块中的其他 Katas（如 Combine、GroupByKey），它们与 Flatten 共同构成了多 PCollection 协同处理的基础能力。

• 大数据
• 批处理
• 流处理
• 数据工程