java – Apache Beam – 无界PCollection的集成测试

发布时间：2020-12-14 05:57:30 所属栏目：Java 来源：网络整理

导读：我们正在为Apache Beam管道构建集成测试,并且遇到了一些问题.有关背景信息,请参见有关我们管道的详情：我们使用PubsubIO作为我们的数据源(无界PCollection) 中间变换包括自定义CombineFn和非常简单的窗口/触发策略我们的最终转换是JdbcIO,使用org.neo4j.j

我们正在为Apache Beam管道构建集成测试,并且遇到了一些问题.有关背景信息,请参见

有关我们管道的详情：

>我们使用PubsubIO作为我们的数据源(无界PCollection)
>中间变换包括自定义CombineFn和非常简单的窗口/触发策略
>我们的最终转换是JdbcIO,使用org.neo4j.jdbc.Driver写入Neo4j

目前的测试方法：

>在运行测试的计算机上运行Google Cloud的发布/订阅模拟器
>构建内存中的Neo4j数据库,并将其URI传递给我们的管道选项
>通过调用OurPipeline.main运行管道(TestPipeline.convertToArgs(options)
>使用Google Cloud的Java Pub / Sub客户端库将消息发布到测试主题(使用Pub / Sub仿真器),PubsubIO将从中读取
>数据应该流经管道并最终命中我们的Neo4j内存实例
>在Neo4j中对这些数据的存在做出简单的断言

这是一个简单的集成测试,它将验证我们的整个管道是否按预期运行.

我们目前面临的问题是,当我们运行我们的管道时,它会阻塞.我们正在使用DirectRunner和pipeline.run()(而不是pipeline.run().waitUntilFinish()),但测试似乎在运行管道后挂起.因为这是一个无限制的PCollection(以流模式运行),所以管道不会终止,因此不会到达任何代码.

所以,我有几个问题：

1)有没有办法运行管道然后稍后手动停止？

2)有没有办法异步运行管道？理想情况下,它会启动管道(然后将继续轮询Pub / Sub以获取数据),然后转到负责发布到Pub / Sub的代码.

3)这种集成测试方法是否合理,或者是否有更好的方法可能更直接？这里的任何信息/指导将不胜感激.

如果我能提供任何额外的代码/背景,请告诉我 – 谢谢！

解决方法

您可以通过将isBlockOnRun管道选项设置为false来使用DirectRunner异步运行管道.只要保留对可用返回的PipelineResult的引用,对该结果调用cancel()就应该停止管道.

对于第三个问题,您的设置似乎合理.但是,如果您希望对管道进行较小规模的测试(需要较少的组件),则可以将所有处理逻辑封装在自定义PTransform中.此PTransform应采用已从输入源完全解析的输入,并生成尚未为输出接收器解析的输出.

当做到这一点,您可以使用创建(这一般不会行使触发)或TestStream(这可能,这取决于你如何构建TestStream)与DirectRunner产生输入数据的有限的量,将此处理PTransform到PCollection,并在输出PCollection上使用PAssert来验证管道是否生成了您期望的输出.

有关测试的更多信息,B??eam网站在Programming Guide和blog post中提供了有关使用TestStream测试管道的这些测试样式的信息.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!