如何在Python中测试Beam管道（Google Dataflow）？

Question

我正在了解我们应该如何使用Google DataFlow（基于Apache Beam）Python SDK测试管道。

https://beam.apache.org/documentation/pipelines/test-your-pipeline/ https://cloud.google.com/dataflow/pipelines/creating-a-pipeline-beam

上面的链接仅适用于Java。我对Google为什么要指向Java Apache测试感到困惑。

我希望能够查看两个p集合上的CoGroupByKey连接的结果。我来自Python背景，使用Beam / Dataflow几乎没有经验。

真的可以使用任何帮助。我知道这在一定程度上是开放式的。基本上，我需要能够查看管道中的结果，这使我无法查看CoGroupByKey Join的结果。

下面的代码

    #dwsku, product are PCollections coming from BigQuery. Nested Values as 
    #well in Product, but not dwsku
    d1 = {'dwsku': dwsku, 'product': product}
    results = d1 | beam.CoGroupByKey()
    print results

打印内容：

    PCollection[CoGroupByKey/Map(_merge_tagged_vals_under_key).None]

Answer 1

如果您想在计算机上本地测试它，则应该从使用开始DirectRunner，然后您将能够通过打印日志或通过在调试器中停止执行来调试它。

为了在本地查看整个 PCollection，您可以执行以下操作：

d1 = {'dwsku': dwsku, 'product': product}
results = d1 | beam.CoGroupByKey()

def my_debug_function(pcollection_as_list):
    # add a breakpoint in this function or just print
    print pcollection_as_list

debug = (results | beam.combiners.ToList() | beam.Map(my_debug_function))

这里有几件事需要记住：

• ToList()转换可能会分配大量内存
• 使用时DirectRunner您应该使用.wait_until_finish()管道的方法，这样您的脚本就不会在管道完成执行之前结束
• 如果您的管道从 BigQuery 下载数据，则应LIMIT在本地运行时放入查询