批处理流处理大数据【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam15/beam点击查看免费下载Apache Beam 采用用户代码构建管道图、再由远端 Runner 执行的间接执行模型这使得调试一次失败的远端运行往往代价高昂。本篇指南基于 Apache Beam 官方文档 test-your-pipeline.md 展开系统讲解如何在提交到目标 Runner 之前用 Java / Python SDK 在本地完成函数级、转换级、端到端三个层次的管道测试你将掌握TestPipeline、Create、PAssert/assert_that的完整用法并借助仓库源码理解其底层执行与校验机制最终能够像 Beam 官方测试套件那样为自己的复合转换Composite Transform和完整管道编写可复现的单元测试。为什么必须先本地测试再远端运行Beam 模型的核心特征是间接性你的用户代码并不会直接操作数据而是构建一个PCollection与PTransform构成的管道图pipeline graph再交给某个 Runner如 Dataflow、Flink、Spark在本地或远端集群上执行。当管道在远端失败时定位问题的成本非常高——日志分散、执行时序不可控、难以在失败现场单步调试。因此在把管道提交到目标 Runner 之前对管道代码做本地单元测试通常是发现和修复缺陷最快的方式同时还能复用你熟悉的本地调试工具断点、IDE 调试器、日志等。Beam 官方推荐的两步走策略是先用 DirectRunner本地 Runner做本地测试与开发验证本地测试通过后再使用目标 Runner例如 Flink Runner 本地或远端 Flink 集群做小规模验证。从源码看DirectRunner 专为严格校验 Beam 模型语义而设计direct.md 中列出了它会强制检查的几类行为包括元素不可变性immutability、元素可编码性encodability、元素在任意阶段按任意顺序处理、用户函数DoFn、CombineFn等的可序列化性。它不做执行性能优化且要求全部用户数据驻留内存因此不适用于生产管道但非常适合测试阶段——用 DirectRunner 通过测试能显著提升管道在不同 Runner 间的可移植性与健壮性。Beam 单元测试的三个层次Beam SDK 为管道代码提供了从低到高三个层次的单元测试手段层次测试对象典型手段第一层管道中使用的单个函数如DoFn的处理逻辑直接调用函数断言第二层**整个转换Transform**作为一个测试单元TestPipelineCreatePAssert第三层整条管道端到端静态输入 静态期望输出 PAssert第二、三层是本文重点。为了支持这些测试Beam Java SDK 在org.apache.beam.sdk.testing包源码位于 sdks/java/core/src/main/java/org/apache/beam/sdk/testing中提供了TestPipeline、PAssert、TestStream等一批测试类仓库中 sdks/java/core/src/test/java/org/apache/beam/sdk 下的测试代码可以直接作为参考与模板。转换测试的标准模式测试一个自己编写的转换官方推荐遵循如下五步模式创建一个TestPipeline准备一份静态的、已知结果的测试输入数据用Create转换把输入数据构造成PCollection对输入PCollection应用待测转换保存输出PCollection用PAssert及其子断言类验证输出PCollection是否包含期望的元素。下面逐一讲解模式中的核心组件。TestPipeline专为测试设计的 Pipeline 子类TestPipeline是 Beam Java SDK 与 Python SDK 中专门用于测试转换的类Java 实现见 TestPipeline.java其类声明为public class TestPipeline extends Pipeline implements TestRule第 107 行同时实现了 JUnit 的TestRulePython 实现见 test_pipeline.py其 docstring 明确说明TestPipelineclass is used inside of Beam tests that can be configured to run against pipeline runner。测试中请用TestPipeline取代普通Pipeline创建管道对象。与Pipeline.create不同TestPipeline.create会在内部自动处理PipelineOptions的构造与默认值设置无需你手动传入 Runner 参数。Java 创建方式Pipeline p TestPipeline.create();在真实的 JUnit 测试中标准写法是将其声明为Rule字段。事实上TestPipeline.run()源码中有一处强制校验TestPipeline.java若TestPipeline未以Rule方式声明就调用run()会抛出IllegalStateException提示 Is your TestPipeline declaration missing a Rule annotation?例如Rule public final transient TestPipeline pipeline TestPipeline.create();Rule的作用在于JUnit 会在每个测试方法执行前创建TestPipeline并在测试方法结束后触发其内部的执行强制校验逻辑详见下文源码级原理。Python 创建方式上下文管理器with TestPipeline() as p: ...Python 版的TestPipeline同样接管了PipelineOptions的构造其构造函数会通过_parse_test_option_args解析命令行参数--test-pipeline-options来获得 Runner 选项列表test_pipeline.py并通过run()在blockingTrue默认时阻塞等待执行完成同时断言最终状态必须是DONE或CANCELLED否则抛出 Pipeline execution failed.test_pipeline.py。用 Create 转换构造测试输入Create转换可以把标准内存集合如 Java / Python 的List转换为PCollection是测试中构造已知静态输入的标准手段PCollectionString input p.apply(Create.of(WORDS));input p | beam.Create(WORDS)关于Create更完整的用法如指定 Coder、Create.of与Create.fromIterable等可参考 编程指南 中Creating a PCollection一节。PAssert对 PCollection 内容的断言PAssert是 Beam Java SDK 提供的、针对PCollection内容的断言工具用于验证某个PCollection是否包含一组特定的期望元素。其 Java 实现见 PAssert.java核心特性如下内嵌进管道图执行PAssert的类注释指出Such an assertion can be checked no matter what kind of PipelineRunner is used即无论使用哪种 Runner断言都会随管道一起被执行和校验必须位于run()之前注释明确 the PAssert call must precede the call to Pipeline.run基于 Metrics 汇总成败PAssert内部维护SUCCESS_COUNTER PAssertSuccess与FAILURE_COUNTER PAssertFailure两个计数器PAssert.java在DefaultConcludeFn中遇到失败断言会立即抛出对应的AssertionError。对给定的PCollection可以这样验证其内容元素顺序无关PCollectionString output ...; // Check whether a PCollection contains some elements in any order. PAssert.that(output) .containsInAnyOrder( elem1, elem3, elem2);from apache_beam.testing.util import assert_that from apache_beam.testing.util import equal_to output ... # Check whether a PCollection contains some elements in any order. assert_that( output, equal_to([elem1, elem3, elem2]))Python 侧要点assert_that与equal_to均位于 util.py。assert_that本质是一个PTransformAssertThat它会物化materialize整个PCollection再交给 matcher 校验因此只应用于测试场景其内部通过CoGroupByKey与一个恒等单例值做关联确保即使被断言的PCollection为空matcher 也会被执行。常用的 matcher 包括matcher作用源码位置equal_to(expected)校验期望与实际的元素互为排列顶层顺序无关且支持自定义equals_fn比较util.pycontains_in_any_order(iterable)通过collections.Counter比较两个可迭代对象的计数是否一致util.pyequal_to_per_window(dict)按窗口断言需assert_that开启reify_windowsTrueutil.pyis_empty()/is_not_empty()断言 PCollection 为空 / 非空util.pymatches_all(expected)用 Hamcrest matcher 列表匹配元素util.pyJava 侧要点任何使用PAssert的 Java 代码必须链接 JUnit 与 Hamcrest。若使用 Maven可在项目的pom.xml中加入以下依赖dependency groupIdorg.hamcrest/groupId artifactIdhamcrest/artifactId version2.2/version scopetest/scope /dependency除了PAssert.that(PCollection)之外Java 版PAssert还提供一系列针对不同形态 PCollection 的断言入口均可在 PAssert.java 中查到实现PAssert.thatSingleton(PCollectionT)断言单元素 PCollection 的值如Combine.globally的结果PAssert.thatSingletonIterable(...)断言只含单个Iterable元素的 PCollectionPAssert.thatMap(...)断言KV集合按 key 映射为 Map每 key 至多一个值PAssert.thatMultimap(...)断言KV集合按 key 映射为多值 MapPAssert.thatFlattened(PCollectionList)断言多路 PCollection 被 Flatten 后的整体内容。IterableAssert接口上还提供containsInAnyOrder(T...)、containsInAnyOrder(IterableT)、empty()、notEmpty()、satisfies(SerializableFunction)等多种断言组合覆盖恰好包含这些元素为空非空满足自定义校验函数等常见场景。复合转换测试完整示例CountTest下面是一个完整的复合转换测试被测对象是Count.perElement()转换测试先用Create从ListString构造输入PCollection再对输出做断言。Java 版本public class CountTest { // Our static input data, which will make up the initial PCollection. static final String[] WORDS_ARRAY new String[] { hi, there, hi, hi, sue, bob, hi, sue, , , ZOW, bob, }; static final ListString WORDS Arrays.asList(WORDS_ARRAY); public void testCount() { // Create a test pipeline. Pipeline p TestPipeline.create(); // Create an input PCollection. PCollectionString input p.apply(Create.of(WORDS)); // Apply the Count transform under test. PCollectionKVString, Long output input.apply(Count.StringperElement()); // Assert on the results. PAssert.that(output) .containsInAnyOrder( KV.of(hi, 4L), KV.of(there, 1L), KV.of(sue, 2L), KV.of(bob, 2L), KV.of(, 3L), KV.of(ZOW, 1L)); // Run the pipeline. p.run(); } }Python 版本import unittest import apache_beam as beam from apache_beam.testing.test_pipeline import TestPipeline from apache_beam.testing.util import assert_that from apache_beam.testing.util import equal_to class CountTest(unittest.TestCase): def test_count(self): # Our static input data, which will make up the initial PCollection. WORDS [ hi, there, hi, hi, sue, bob, hi, sue, , , ZOW, bob, ] # Create a test pipeline. with TestPipeline() as p: # Create an input PCollection. input p | beam.Create(WORDS) # Apply the Count transform under test. output input | beam.combiners.Count.PerElement() # Assert on the results. assert_that( output, equal_to([ (hi, 4), (there, 1), (sue, 2), (bob, 2), (, 3), (ZOW, 1)])) # The pipeline will run and verify the results.这个示例并非凭空虚构仓库中真实的 CountTest.javatestCountPerElementBasic方法第 56-70 行使用了完全相同的输入数组与断言结果并额外展示了PAssert.that(output).empty()校验空集合、Count.globally()返回单元素集合等变体可作为扩展参考。端到端测试整条管道TestPipeline与PAssert等测试类同样可以用于整条管道的端到端测试。典型做法如下为管道的每个输入数据源准备一份已知的静态测试输入准备一份与管道最终输出PCollection期望一致的静态输出数据用TestPipeline取代标准的Pipeline.create用Create转换替代管道中的Read转换从静态输入构造一个或多个PCollection依次应用管道自身的各转换用PAssert替代管道中的Write转换验证最终PCollection的内容与静态期望输出一致。WordCount 管道端到端测试下面示例展示如何测试 WordCount 示例管道。WordCount通常从文本文件按行读取输入测试则改为用一个ListString存放若干文本行再通过Create构造初始PCollection。WordCount的最终转换复合转换CountWords产出适合打印的格式化词频PCollectionString测试管道不把该 PCollection 写入输出文件而是用PAssert验证其元素与静态期望字符串数组完全一致。Java 版本public class WordCountTest { // Our static input data, which will comprise the initial PCollection. static final String[] WORDS_ARRAY new String[] { hi there, hi, hi sue bob, hi sue, , bob hi}; static final ListString WORDS Arrays.asList(WORDS_ARRAY); // Our static output data, which is the expected data that the final PCollection must match. static final String[] COUNTS_ARRAY new String[] { hi: 5, there: 1, sue: 2, bob: 2}; // Example test that tests the pipelines transforms. public void testCountWords() throws Exception { Pipeline p TestPipeline.create(); // Create a PCollection from the WORDS static input data. PCollectionString input p.apply(Create.of(WORDS)); // Run ALL the pipelines transforms (in this case, the CountWords composite transform). PCollectionString output input.apply(new CountWords()); // Assert that the output PCollection matches the COUNTS_ARRAY known static output data. PAssert.that(output).containsInAnyOrder(COUNTS_ARRAY); // Run the pipeline. p.run(); } }Python 版本import unittest import apache_beam as beam from apache_beam.testing.test_pipeline import TestPipeline from apache_beam.testing.util import assert_that from apache_beam.testing.util import equal_to class CountWords(beam.PTransform): # CountWords transform omitted for conciseness. # 完整实现可参考仓库中的 wordcount_debugging.py。 pass class WordCountTest(unittest.TestCase): # Our input data, which will make up the initial PCollection. WORDS [ hi, there, hi, hi, sue, bob, hi, sue, , , ZOW, bob, ] # Our output data, which is the expected data that the final PCollection must match. EXPECTED_COUNTS [hi: 5, there: 1, sue: 2, bob: 2] # Example test that tests the pipelines transforms. def test_count_words(self): with TestPipeline() as p: # Create a PCollection from the WORDS static input data. input p | beam.Create(WORDS) # Run ALL the pipelines transforms (in this case, the CountWords composite transform). output input | CountWords() # Assert that the output PCollection matches the EXPECTED_COUNTS data. assert_that(output, equal_to(EXPECTED_COUNTS), labelCheckOutput) # The pipeline will run and verify the results.需要说明的是示例中的CountWords是一个真实的 Beam 复合转换其完整定义可以参考仓库中的 wordcount_debugging.pyCountWords类位于第 112 行该文件还示范了如何在真实管道中结合日志与断言做调试。源码级原理TestPipeline 如何保证测试真的被执行TestPipeline的价值不止于方便它在底层内置了一套执行强制enforcement机制专门防止写出假阳性测试。理解这些机制能帮助你写出更可靠的测试也可以帮助你排查测试意外失败的原因。1. Rule 与执行生命周期Java 版TestPipeline通过实现 JUnit 的TestRuleTestPipeline.java接管测试方法的执行在用户测试代码执行结束后会调用enforcement.get().afterUserCodeFinished()触发后续校验若用户代码抛出的异常未被捕获则跳过校验避免对已经失败的管道再做无意义检查。2. 自动补跑enableAutoRunIfMissing如果测试代码中忘了调用pipeline.run()TestPipeline可以在测试结束时自动补一次运行TestPipeline p TestPipeline.create().enableAutoRunIfMissing(true);对应源码中的PipelineRunEnforcement.afterUserCodeFinished()第 134-138 行当run()从未被调用且自动补跑已开启时会代为执行pipeline.run().waitUntilFinish()。3. 遗弃节点检测Abandoned Node Enforcement更严格的是PipelineAbandonedNodeEnforcementTestPipeline.java它会在管道运行前后各做一次拓扑遍历traverseTopologically并比对节点集合检测两类典型错误并抛出AbandonedNodeException测试缺少pipeline.run()语句管道从未运行此时若未开启自动补跑会抛出PipelineRunMissingException消息为 The pipeline has not been run.管道运行后又追加了 PTransform新增的转换没有被执行。源码注释第 459-472 行明确指出当检测到真实 Runner非CrashingRunner或测试带有Category(NeedsRunner.class)/Category(ValidatesRunner.class)注解时遗弃节点检测会自动启用也可以通过enableAbandonedNodeEnforcement(boolean)手动开关。这套机制确保你断言的东西真的被运行过而非静默漏测。4. PAssert 成功计数校验verifyPAssertsSucceededTestPipeline.run()在执行后还会调用verifyPAssertsSucceeded(pipeline, result)TestPipeline.java它统计管道中注册的PAssert数量并通过MetricsFilter.named(PAssert.class, PAssert.SUCCESS_COUNTER)查询名为PAssertSuccess的计数器断言成功的断言数 期望的断言数。这从框架层面杜绝了PAssert 挂在管道图上但 Runner 从未执行它的漏检场景该逻辑依赖 Runner 的 Metrics 支持。5. 面向 Runner 的测试选项注入Java 版TestPipeline通过系统属性beamTestPipelineOptionsPROPERTY_BEAM_TEST_PIPELINE_OPTIONS第 254-255 行读取一个 JSON 数组形式的PipelineOptions列表例如[ --runnerTestDataflowRunner, --projectmygcpproject, --stagingLocationgs://mygcsbucket/path ]testingPipelineOptions()第 497-528 行在属性为空时直接使用PipelineOptionsFactory.create()的默认选项否则从该 JSON 解析选项同时会把StableUniqueNames设为ERROR并注入默认 FileSystem 选项。注意所需的具体选项集合因 Runner 而异且要求包含 SDK 与测试类的 JAR 都在 classpath 中。Python 侧的等价机制是 pytest 命令行参数--test-pipeline-optionstest_pipeline.py官方 docstring 给出的执行方式如下pytest -m it_validatesrunner \ --test-pipeline-options--runnerDirectRunner \ --job_namemyJobName \ --num_workers1该参数由TestPipeline._parse_test_option_args解析为选项列表并用于构造PipelineOptions当测试被标记为is_integration_testTrue却未提供该参数时会直接SkipTest跳过第 154-160 行避免集成测试被单元测试的执行入口误触发。这些解析逻辑在 test_pipeline_test.py 中有对应的单元测试用例如test_option_args_parsing。6. 测试类别注解与 Runner 分级Java 版TestPipeline的setDeducedEnforcementLevel()第 294-315 行会根据测试方法上的Category注解推断 enforcement 级别Category(NeedsRunner.class)该测试需要真实 Runner执行才能通过Category(ValidatesRunner.class)该测试用于验证 Runner 行为配合TestPipelineOptions可切换到目标 Runner 执行。相关注解定义见 NeedsRunner.java 与 ValidatesRunner.java。若测试被标注为NeedsRunner但 Runner 被设置成了CrashingRunner默认无选项时的占位 Runner会抛出配置错误的IllegalStateException。Python 侧对应的是pytest.mark.it_validatesrunner标记见 test_pipeline.py 第 42-43 行 docstring。补充如何测试无界流式管道本文档中的TestPipelineCreatePAssert组合主要面向有界数据。对于无界管道unbounded pipeline的测试Beam Java SDK 提供TestStream类源码见 TestStream.java可以模拟带有时间戳与 watermark 的流式元素序列。关于用TestStream与PAssert测试无界管道的详细方法可参考 Beam 博客《Testing Unbounded Pipelines in Apache Beam》仓库内版本位于 test-stream.md。小结与最佳实践测试优先于远端调试Beam 的间接执行模型让远端失败难以定位本地单元测试是最快的缺陷发现手段按层次测试先测函数再测转换TestPipelineCreatePAssert最后端到端测试整条管道用Create替Read、用PAssert替Write严格使用RuleJava 中务必以Rule public final transient TestPipeline p TestPipeline.create();声明缺失会直接报错信赖 enforcement 机制不要手动关闭遗弃节点检测enableAbandonedNodeEnforcement(false)它和verifyPAssertsSucceeded一起保证断言真正被执行、结果真正被校验静态数据 确定性断言测试输入与期望输出一律使用静态已知数据断言优先使用containsInAnyOrder/equal_to这类与元素顺序无关的匹配器小规模验证后再上生产 RunnerDirectRunner 通过后再用 Flink 等 Runner 在本地或远端集群做小规模验证最后才提交生产规模作业。至此你已经掌握了从单个函数、单个复合转换到整条管道的完整测试方法论并理解了TestPipeline/PAssert在源码层面的执行强制与校验机制——这些能力可以直接用于为你的 Beam 管道编写可靠、可复现的单元测试。赞分享批处理流处理大数据【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam15/beam点击查看免费下载相关推荐Apache Beam 管道单元测试实战TestPipeline、PAssert 与 Direct Runner 完整指南Apache Beam 管道单元测试实战TestPipeline、PAssert 与 Direct Runner 完整指南 在将 Apache Beam 管道批处理流处理大数据Apache Beam 管道测试实战用 TestPipeline、PAssert 与本地运行器完成单元测试到端到端验证Apache Beam 管道测试实战用 TestPipeline、PAssert 与本地运行器完成单元测试到端到端验证 在 Apache Beam 中用户代大数据批处理流处理数据工程GetQzonehistory把QQ空间十几年的历史说说一次性完整导出来GetQzonehistory把QQ空间十几年的历史说说一次性完整导出来 如果你有一台存了多年照片的老手机想把它备份到硬盘里会找哪款工具对QQ空间来说网页爬虫数据分析上一篇ChatLLM.cpp工具调用功能详解让AI学会使用外部API和工具下一篇5步掌握Momentum Firmware完整构建从源码编译到刷机的终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考