有关我们管道的详情：
>我们使用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)这种集成测试方法是否合理,或者是否有更好的方法可能更直接？这里的任何信息/指导将不胜感激.
如果我能提供任何额外的代码/背景,请告诉我 – 谢谢！
解决方法
对于第三个问题,您的设置似乎合理.但是,如果您希望对管道进行较小规模的测试(需要较少的组件),则可以将所有处理逻辑封装在自定义PTransform中.此PTransform应采用已从输入源完全解析的输入,并生成尚未为输出接收器解析的输出.
当做到这一点,您可以使用创建(这一般不会行使触发)或TestStream(这可能,这取决于你如何构建TestStream)与DirectRunner产生输入数据的有限的量,将此处理PTransform到PCollection,并在输出PCollection上使用PAssert来验证管道是否生成了您期望的输出.
有关测试的更多信息,Beam网站在Programming Guide和blog post中提供了有关使用TestStream测试管道的这些测试样式的信息.