Apache Beam - 使用无界PCollection进行集成测试

Question

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

有关我们管道的详情:

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

目前的测试方法:

• 在运行测试的计算机上运行Google Cloud的Pub/Sub模拟器
• 构建内存中的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)这种集成测试方法是否合理,或者是否有更好的方法可能更直接？这里的任何信息/指导将不胜感激.

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

Answer 1

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

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

当这样做时,可以使用任一Create(其通常将不行使触发)或TestStream(其可以,这取决于你如何构建TestStream)与DirectRunner,以产生输入数据的有限数量,应用该处理PTransform到该PCollection,并使用PAssert上输出PCollection以验证管道是否生成了您期望的输出.

有关测试的更多信息,B​​eam网站在编程指南中有关于这些测试样式的信息,以及关于测试管道的博客文章TestStream.