无界PCollection的Apache波束集成测试

Question

我们正在为Apache Beam管道构建一个集成测试，并遇到了一些问题。有关上下文，请参见下文...

有关我们管道的详细信息：

null

null

这是一个简单的集成测试，它将验证我们的管道作为一个整体的行为是预期的。

我们目前面临的问题是，当我们运行管道时，它正在阻塞。我们使用的是Directrunner和pipeline.run()（而不是pipeline.run().waituntilfinish())，但是测试似乎在运行管道后挂起。因为这是一个无界的pcollection（以流模式运行），所以管道不会终止，因此没有到达管道之后的任何代码。

所以，我有几个问题：

让我知道如果我可以提供任何额外的代码/上下文-谢谢！

Answer 1

通过将设置isblockonrun管道选项传递为false，可以使用Directrunner异步运行管道。只要保持对返回的pipelineResult的引用可用，就可以在该结果上调用cancel()来停止管道。

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

完成此操作后，您可以使用create（通常不执行触发器操作）或teststream（根据如何构造teststream)与directrunner来生成有限数量的输入数据，将此处理ptransform应用于该pcollection，并在输出pcollection上使用passert来验证管道是否生成了预期的输出。

有关测试的更多信息，Beam网站在编程指南和一篇关于使用TestStream测试管道的博客文章中有关于这些测试风格的信息。