大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载Apache Beam 的PTransformTransform变换/转换是管道中表示数据处理操作的核心抽象它接收零个或多个PCollection作为输入产出零个或多个PCollection作为输出是所有批处理与流式处理逻辑的承载单元。本文将基于 Beam 官方编程指南中关于 PTransform 的基础说明结合本仓库 Java 与 Python SDK 的源码实现系统讲解 PTransform 的定义、四大关键特征、常见变换类型、用户代码User Code概念并给出可直接运行的ParDo示例与自定义复合变换的实战建议。什么是 PTransform在 Apache Beam 的统一编程模型中一条管道Pipeline本质上是由一系列数据处理步骤串联而成的有向图而每一步就是一个 PTransform。官方文档05_basic_ptransforms.md给出的定义是APTransform或 transform代表 Apache Beam 管道中的一次数据处理操作或一个步骤。一个 transform 被应用到零个或多个PCollection对象上并产生零个或多个PCollection对象。这条定义包含两个关键点输入与输出都是 PCollectionPCollection 是 Beam 中对分布式数据集有界批数据或无界流数据的抽象PTransform 不直接操作底层存储而是以 PCollection 为边界进行数据流动零个或多个的灵活性有的 transform 没有输入如数据源读取有的没有输出如数据写出但绝大多数 transform 接收一个或多个 PCollection 并产出一个或多个 PCollection。在 Java SDK 中这一抽象对应org.apache.beam.sdk.transforms.PTransformInputT, OutputT泛型抽象类定义于 PTransform.java其类型参数InputT extends PInput、OutputT extends POutput分别约束输入与输出必须是PCollection等管道值。在 Python SDK 中则对应apache_beam.transforms.ptransform.PTransform类定义于 ptransform.py该类同时继承了类型提示WithTypeHints与展示数据HasDisplayData能力。PTransform 的四大关键特征官方文档总结了 PTransform 的四个核心特征它们是理解 Beam 编程模型设计意图的钥匙特征含义实践体现Versatility多功能性能够对 PCollection 执行多种多样的操作从简单的逐元素映射ParDo到按键分组GroupByKey、聚合Combine再到读写外部系统TextIO.Read/WriteComposability可组合性可以组合成复杂的数据处理管道一个 transform 的输出 PCollection 可以作为下一个 transform 的输入形成处理序列Parallel execution并行执行专为分布式处理设计可在多个 worker 上同时执行Beam 会将 PCollection 切分为 bundle由 runner 调度到多台机器并行处理Scalability可扩展性能处理海量数据同时适用于批处理与流式数据同一套 PTransform 抽象同时覆盖有界与无界数据无需区分编写其中可组合性是 Beam 编程模型区别于传统单机框架的关键在 Python SDK 中组合通过管道符|完成其底层由PCollection.__or__与PTransform.__or__实现见 core.py在 Java SDK 中则通过apply()方法完成官方 javadoc 明确说明 transform 的调用方式统一为apply()调用方视角下原生实现与复合实现在用法上没有区别。常见 PTransform 类型Beam SDK 内置了大量开箱即用的 PTransform官方文档将其划分为四类对应关系与仓库源码位置如下1. 数据源变换Source Transforms概念上无输入TextIO.Read从文本文件读取数据对应 Java SDK 的org.apache.beam.sdk.io.TextIOCreate从内存中的可迭代对象直接创建 PCollection是最常用的测试与调试工具。Create在 Python SDK 中定义于 core.py源码显示它有两点值得注意的行为其一拒绝将字符串/字节串当作可迭代对象Refusing to treat string as an iterable避免用户把单个字符串误当成字符列表其二若传入字典会转换为items()视图即键值对元组序列。Create还支持reshuffle参数用于控制是否在创建后重新打乱数据分布以优化并行度。Java 对应实现为 Create.java。2. 处理与转换操作Processing ConversionParDo对每个元素应用用户定义的函数DoFn是 Beam 中最核心、最灵活的逐元素变换GroupByKey按键对K, V对进行分组产出PCollectionKVK, IterableVJava 实现见 GroupByKey.javaPython 实现见 core.pyCoGroupByKey对多个按键分组的 PCollection 执行联合分组join 操作Combine对每个键或整个 PCollection 执行聚合操作如求和、求最大值Java 实现见 Combine.javaCount统计元素个数或按键统计频次Java 实现见 Count.java。3. 输出变换Outputting TransformsTextIO.Write将 PCollection 写出到文本文件。这类 transform 概念上通常没有 PCollection 输出而是将数据落盘或写入外部系统。4. 用户自定义复合变换Composite Transforms用户可以基于已有 transform 组合出面向特定业务场景的复合变换详见下文自定义复合变换一节。从 PTransform.java 的 javadoc 可以确认大部分 PTransform 实际上都是其他 PTransform 的复合体只有少数 transform 由 SDK 原生实现用户被鼓励用这种机制模块化自己的代码复合变换会获得自己的名称并且监控界面支持在复合层次结构中导航。用户代码User Code与 Beam 模型约束PTransform 的处理逻辑以函数对象的形式提供Beam 术语中称之为用户代码user code。官方文档明确了两点用户代码会被应用到输入 PCollection 的每一个元素或来自多个 PCollection 的元素用户代码必须满足 Beam 模型的要求——这是分布式执行正确性的前提。满足 Beam 模型要求在实际中意味着什么从源码与 Beam 设计可以归纳为以下几点约束以下为 Beam 模型事实具体编码约束可参见各 SDK 的 DoFn 文档可序列化用户代码对象会被分发到各 worker 节点执行因此必须可序列化。这一点在 PTransform.java 的序列化说明中有直接体现PTransform实现Serializable仅是为了方便在apply()中编写匿名 DoFn其writeObject/readObject是空实现不保存任何状态无共享可变状态依赖由于元素可能在不同 worker 上并行处理用户代码不能依赖跨元素的共享状态幂等性与可重试分布式执行可能发生重试用户代码尤其是有副作用的部分应能容忍重复执行。ParDo 实战示例将用户代码应用到 PCollectionParDo是使用用户代码处理元素的通用机制。官方文档给出的 Python 示例展示了完整的最小管道import apache_beam as beam def SomeUserCode(element): # Do something with an element return element with beam.Pipeline() as pipeline: input_collection pipeline | beam.Create([...]) output_collection input_collection | beam.ParDo(SomeUserCode())逐行拆解这个示例with beam.Pipeline() as pipeline:创建管道上下文退出with块时自动执行runpipeline | beam.Create([...])使用Create从内存列表构造输入 PCollection这是无输入 transform 通过管道符施加在Pipeline上的典型用法input_collection | beam.ParDo(SomeUserCode())将ParDo施加到 PCollection 上对每个元素调用SomeUserCode产出新的 PCollection。从 core.py 的ParDo类源码可以看出几个重要的底层细节DoFn 约束ParDo构造时必须传入DoFn实例否则抛出TypeError源码中self.dofn self.fn保留了历史属性名返回值约定DoFn 的process方法必须为每个输入元素返回可迭代对象iterable最简洁的写法是使用yield关键字生成输出若process方法同时混用yield与return会产生不可预期行为SDK 会发出警告侧输入Side Input传给ParDo的位置参数与关键字参数会被逐一检查识别出其中的 PCollection 后作为侧输入处理执行时这些参数位置上会被替换为对应 PCollection 的当前值按 bundle 提供这是实现广播式辅助数据的机制。对于 Java 用户等价的ParDo写法如下Java SDK 中用户代码形式为DoFn子类Pipeline pipeline Pipeline.create(); PCollectionString input pipeline.apply(Create.of(a, b, c)); PCollectionString output input.apply(ParDo.of(new DoFnString, String() { ProcessElement public void processElement(Element String element, OutputReceiverString out) { out.output(element); } }));此外Python SDK 的ParDo还提供了with_exception_handling()方法见 core.py可自动生成一个死信输出把处理失败的坏数据收集起来例如good, bad inputs | beam.Map(maybe_erroring_fn).with_exception_handling()其中good是成功处理的结果 PCollectionbad是(输入, 错误信息)元组的集合还可通过threshold参数设定坏数据比例上限超过则中止整个管道。这对于生产环境的数据质量治理非常实用。源码级原理PTransform 如何被展开expand无论是原生实现还是复合实现每个 PTransform 都通过expand方法定义自身的展开逻辑。Java SDK 中该方法的约定见 PTransform.java非常清晰不应直接调用expand而应通过apply()将 PTransform 施加到输入上复合变换在expand内部组合其他 transform并返回其中某个组合变换的输出非复合原生变换返回一个新的未绑定输出并通过 runner 特定的注册机制注册求值器。Python SDK 的约定一致ptransform.py 中PTransform的类文档明确要求子类必须定义expand()方法典型用法模式为input | CustomTransform(...)expand会以input为参数被调用。同时该类提供了default_label()方法返回类名作为默认标签以及with_input_types()/with_output_types()方法见 ptransform.py用于声明输入/输出类型提示从而让 Beam 的类型系统typehints在编译期或运行期帮助发现类型不匹配问题。从源码结构还可以推断出 PTransform 的其他能力验证钩子Java 的PTransform提供validate(PipelineOptions)与validate(options, inputs, outputs)方法PTransform.java在管道运行前校验 transform 及其输入输出是否完整正确默认空实现子类可按需覆盖资源提示setResourceHints(ResourceHints)方法PTransform.java可为 transform 指定资源需求例如ResourceHints.create().withMinRam(6 GiB)runner 据此进行资源调度。自定义复合变换封装业务逻辑的最佳实践官方文档将用户自定义复合变换列为常见 transform 类型之一并强调复合变换是模块化代码的推荐方式。其核心思想是把一组存在固定顺序的 transform 封装成一个有名字的、可复用的单元。Python 中的最小复合变换如下import apache_beam as beam class WordCount(beam.PTransform): def expand(self, pcoll): return ( pcoll | Split beam.FlatMap(lambda line: line.split()) | Filter beam.Filter(lambda word: word.strip() ! ) | Count beam.combiners.Count.PerElement() ) with beam.Pipeline() as pipeline: lines pipeline | beam.io.ReadFromText(input.txt) counts lines | WordCount()要点复合变换必须继承beam.PTransform并实现expand(self, pcoll)复合变换内部可以使用带命名标签如Split、Count的子步骤让管道图在监控界面中更易读从调用方视角看复合变换与原生变换完全等价Java javadoc 中明确从调用方角度原生实现与复合实现之间没有区别这保证了抽象可以层层嵌套而不泄漏实现细节。Java 中对应的写法是继承PTransformPCollectionString, PCollectionKVString, Long并覆写expand()。仓库的 Java SDK 中大量此类示例可见于 sdks/java/core/src/main/java/org/apache/beam/sdk/transforms 目录Python SDK 中Flatten、Partition、CombinePerKey、GroupByKey等内置变换均定义于 core.py。总结与建议回顾本文要点PTransform 是 Beam 管道的基本构建块接收 PCollection 输入、产出 PCollection 输出凭借多功能性、可组合性、并行执行与可扩展性四大特征支撑统一批流模型变换可分为四类数据源TextIO.Read、Create、处理转换ParDo、GroupByKey、CoGroupByKey、Combine、Count、数据输出TextIO.Write与用户自定义复合变换用户代码须满足 Beam 模型约束可序列化、无共享可变状态、可重试并通过 DoFn 函数对象以逐元素方式应用复合变换是生产级代码组织的首选Java 的expand/ Python 的expand均提供了统一的扩展点还伴随validate、资源提示、类型提示等配套能力。实践建议入门阶段先用CreateParDo跑通最小管道需要复用逻辑时立刻将多步骤封装为命名复合变换接入生产数据前利用with_exception_handling()或侧输入等机制增强健壮性深入学习时可对照 PTransform.java 与 ptransform.py 的源码注释它们本身就是最权威的 Beam 模型说明书。赞分享大数据批处理流处理数据工程【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam4/beam点击查看免费下载相关推荐Apache Beam Python 复合变换实战继承 PTransform 实现 ExtractAndMultiplyNumbersApache Beam Python 复合变换实战继承 PTransform 实现 ExtractAndMultiplyNumbers 本文以 Apache大数据批处理流处理数据工程Apache Beam Kotlin Kata 实战用 ParDo 实现通用并行处理变换Apache Beam Kotlin Kata 实战用 ParDo 实现通用并行处理变换 本指南以 Apache Beam 仓库中 Kotlin 版 Kata大数据批处理流处理数据工程Apache Beam YAML零代码管道指南不用写一行代码定义数据处理作业Apache Beam YAML零代码管道指南不用写一行代码定义数据处理作业 Apache Beam YAML 是 Apache Beam 官方提供的声明式管大数据批处理流处理数据工程上一篇BilibiliDown跨平台的B站视频下载器从单个视频到整个收藏夹批量下载下一篇MTEX 织构分析完整指南免费 Matlab 工具箱从 EBSD 到极图全流程教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
