如何在python中(单元)测试apache-beam中的流管道？

Question

我写了一些流水线(从Pub/Sub开始)，我想给它添加一些窗口机制。现在我想以适当的方式来测试它，那么如何创建一些“虚拟”流呢？

我的代码：

pipeline_options = PipelineOptions()
pipeline_options.view_as(SetupOptions).save_main_session = True
pipeline_options.view_as(StandardOptions).streaming = True
p = TestPipeline(options=pipeline_options, runner=DirectRunner())
xmls_beam = beam.Create(xmls)
x = p | xmls_beam | beam.FlatMap(process_xmls) | beam.ParDo(FilterTI()) | beam.WindowInto(window.FixedWindows(200)) | beam.GroupByKey()
result = p.run()
result.wait_until_finish()

Answer 1

您可以使用TimestampedValue的PCollection来模拟“虚拟流”。

例如，如果您的输入是：

    l = [window.TimestampedValue('a', 100), window.TimestampedValue('b', 300)]
    pc = p | beam.Create(l) | ...

在你的例子中(宽度为200的固定窗口)，你可以预期输出元素'a‘落入第一个窗口，而'b’落入第二个窗口。