大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载本文围绕 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.mdFlattenWith 被定义为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.从这一描述可以提炼出两个关键信息结果是一个逻辑统一的 PCollection合并后元素内容不变只是汇聚到同一条管道中继续流转输入来源多样既可以合并已经构造好的 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); PCollectionString wordsStartingWithA pipeline.apply(Words starting with A, Create.of(apple, ant, arrow)); PCollectionString wordsStartingWithB pipeline.apply(Words starting with B, Create.of(ball, book, bow)); PCollectionString output applyTransform(wordsStartingWithA, wordsStartingWithB); output.apply(Log.ofElements()); pipeline.run(); } static PCollectionString applyTransform( PCollectionString words1, PCollectionString words2) { PTransformPCollectionString, PCollectionString 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 变换对象PTransformPCollectionString, PCollectionString flattenTransform Flatten.with(words2);Flatten.with(PCollectionT)返回的是一个PTransformPCollectionT, PCollectionT。注意这里并没有立即执行合并——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其内部实现是一个ParDoDoFn对每个元素调用LOG.info(message)输出元素内容若元素所在窗口不是GlobalWindow还会附加窗口信息随后原样out.output(element)透传。四、测试验证合并行为的精确断言测试代码位于 TaskTest.javaTest public void flattenWith() { PCollectionString wordsStartingWithA testPipeline.apply(Words starting with A, Create.of(apple, ant, arrow)); PCollectionString wordsStartingWithB testPipeline.apply(Words starting with B, Create.of(ball, book, bow)); PCollectionString results Task.applyTransform(wordsStartingWithA, wordsStartingWithB); PAssert.that(results) .containsInAnyOrder(APPLE, ANT, ARROW, ball, book, bow); testPipeline.run().waitUntilFinish(); }该测试清晰地印证了两个事实合并成功6 个元素全部出现在输出 PCollection 中顺序无保证containsInAnyOrder表明 Flatten 输出的元素不保证顺序——这正是分布式数据处理的常态也是 Beam 编程模型的重要理念变换作用范围前三个单词为大写来自被MapElements处理过的 words1后三个保持小写来自 words2精确验证了链式 Flatten 的位置语义。五、源码级原理Flatten.with() 到底做了什么要真正理解 FlattenWith需要进入 SDK 核心实现 Flatten.java 一探究竟。5.1 with(PCollection) 的等价实现Flatten.with(PCollectionT other)工厂方法Flatten.java#L102-L104返回一个内部类FlattenWithPCollection其expand方法Flatten.java#L116-L119只有一行核心逻辑Override public PCollectionT expand(PCollectionT 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(PTransformPBegin, PCollectionT other)重载Flatten.java#L144-L159public static T PTransformPCollectionT, PCollectionT with( PTransformPBegin, PCollectionT other) { return new PTransformPCollectionT, PCollectionT() { Override public PCollectionT expand(PCollectionT 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-L207WindowFn 兼容性遍历所有输入若任一输入的WindowFn与第一个不兼容抛出IllegalStateExceptionInputs to Flatten had incompatible window windowFnsTrigger 兼容性若任一输入的触发器与第一个不兼容同样抛出IllegalStateExceptionInputs to Flatten had incompatible triggers有界性聚合输出 PCollection 的IsBounded由所有输入的isBounded做 AND 运算得出——即只要有一个输入是无界流输出就是无界的Coder 继承输出使用第一个输入 PCollection 的 Coder若输入列表为空则 Coder 保持未指定。5.4 窗口与时间戳语义合并产生的输出元素保留原输入元素所属的窗口与时间戳输出 PCollection 的 WindowFn 与所有输入一致。这在流式场景中非常重要Flatten 不会重组窗口边界只是把多条流的元素汇合到同一处理逻辑中。六、FlattenWith 与传统 Flatten 的对比维度Flatten.with(other)FlattenWithFlatten.pCollections()传统 Flatten输入形式单个 PCollection otherPCollection 或 PTransformPCollectionListT可含任意多个 PCollection链式调用支持可直接内联在变换链中间需先把多个 PCollection 打包成列表合并根变换支持with(PTransform)重载需手动 apply 后再打包适用场景两路合并、边变换边合并、链式编写多路2合并、批量聚合底层实现内部仍是PCollectionList.of(input).and(other).apply(pCollections())直接展开为单一输出可以这样理解FlattenWith 是传统 Flatten 的链式友好语法糖其底层与pCollections()完全同源源码见 Flatten.java。当需要合并 3 个及以上 PCollection 时依然建议使用PCollectionList.of(pc1).and(pc2).and(pc3).apply(Flatten.pCollections())。七、运行方式与学习环境本 Kata 属于 learning/katas/java 训练项目的一部分。按照其 README 的指引推荐以下方式搭建运行环境使用 IntelliJ IDEA 的 Education 版本或安装 EduTools 插件选择Open打开learning/katas/java目录在弹出的提示中选择Import Gradle project配置 Gradle等待 Gradle 构建完成在Project Structure中设置项目 SDK例如 JDK 8在Project工具窗口切换到Course视图即可看到 FlattenWith 等各课题目录完成填空并运行测试验证。由于FlattenWith的PTransform支持被保存与复用你也可以把它封装成工具方法在真实业务管道中实现两路数据源汇合后再统一处理的模式。完成本练习后建议继续尝试 Core Transforms 模块中的其他 Katas如 Combine、GroupByKey它们与 Flatten 共同构成了多 PCollection 协同处理的基础能力。赞分享大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载相关推荐如何用 graph.v1 的 train_schema 与 predict_schema 自定义 Rasa 执行图如何用 graph.v1 的 train_schema 与 predict_schema 自定义 Rasa 执行图 Rasa 默认用 default.v1 配方大数据批处理流处理数据工程Apache Beam 核心变换实战使用 Partition 将 PCollection 拆分为多个输出集合Java Kata 详解Apache Beam 核心变换实战使用 Partition 将 PCollection 拆分为多个输出集合Java Kata 详解 Partition大数据批处理流处理数据工程如何检查 PEP 723 内联元数据脚本ty 与 uv --with-requirements 怎么用如何检查 PEP 723 内联元数据脚本ty 与 uv with requirements 怎么用 假设你写了一个带 PEP 723 内联元数据头的 Pyt大数据批处理流处理数据工程上一篇SeleniumBasic终极指南VB生态的浏览器自动化革命下一篇轻松掌握Nginx反向代理图形化管理工具完全指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考