如何在python中将动态pcollection作为sideinput传递？答案

【问题标题】：How to pass dynamic pcollection as sideinput in python?如何在python中将动态pcollection作为sideinput传递？
【发布时间】：2020-01-03 02:18:56
【问题描述】：

我有一个用例，我需要将 file_path（作为新元素）与另一个 pcollection（元素字典）附加。

所以我决定将 pcollection 作为下一次转换的输入，并将 file_path（从 pardo 函数生成）作为 sideinput。

由于生成的 file_path 是一个 pCollection，所以我将其用作：

beam.pvalue.AsSingelton(file_path)

但奇怪的是它没有执行这个语句，因此没有输出，甚至没有任何错误。但是当我使用以下方法创建 pcollection 时：

file_path = beam.Create(['some_path_value']) 而不是像上面一样通过 beam.pvalue.AsSingleton(file_path) 完美运行。

但我不想显式创建 pcollection。我需要使用上面的转换输出，因为它是侧输入。

我为我的“file_path”尝试了每种形式，比如将其设为列表、字典、元组。

我也尝试了不同的光束功能。

def collect(pcoll, path):
   print(pcoll, "-----------")
   print(path,"-------------")       
   return path

class Extract(beam.DoFn):    
   def process(self, element, *args, **kwargs):       
       pub_sub_json= json.loads(element)
       gcs_url = "gs://" + pub_sub_json["bucket"] + "/" + pub_sub_json["name"]
       yield gcs_url

file_path = (p |"Read" >> beam.io.ReadFromPubSub(
                           subscription=options.session_subscription,
                           with_attributes=False)
             | "Extract" >> beam.ParDo(Extract()))                     

result = (some_other_pcollection |"display" >> beam.FlatMap(collect, path=beam.pvalue.AsList(file_path)))

期望：它应该进入一个 collect 函数并打印 Extract 方法将作为 file_path 返回的任何内容。

没有错误和警告出现，但它甚至没有进入 collect 方法。当数据不是来自上述转换时，通常会发生这种情况。但是我检查了很多次 file_name 有数据。

但没有输出，它只在我使用 beam.Create(["pass_some_file_path"]) 并将其结果用作侧输入时显示输出

【问题讨论】：

标签： python-3.x google-cloud-platform google-cloud-dataflow apache-beam

【解决方案1】：

Transform 的输出应采用 2 元组 ( (key,value) ) 格式，以便通过 pvalue.AsDict() 将其作为侧面输入接受。它也适用于 value.AsList()。

def collect(pcoll, path):
   print(pcoll, "-----------")
   print(path,"-------------")       
   return path

class Extract(beam.DoFn):    
   def process(self, element, *args, **kwargs):       
       pub_sub_json= json.loads(element)
       gcs_url = "gs://" + pub_sub_json["bucket"] + "/" + pub_sub_json["name"]
       yield pvalue.TaggedOutput("url",("url" :gcs_url))

file_path = (p |"Read" >> beam.io.ReadFromPubSub(
                           subscription=options.session_subscription,
                           with_attributes=False)
             | "Extract" >> beam.ParDo(Extract())) 
file_path_new= (file_path["url"] | "Group by key" >> beam.GroupByKey())                    

result = (some_other_pcollection |"display" >> beam.FlatMap(collect, path=beam.pvalue.AsList(file_path_new)))

你能以更好的方式解决它吗？

【讨论】：