如何在Python中将动态pcollection作为sideinput传递？

Question

我有一个用例，在这个用例中，我需要将一个file_path（作为一个新元素）追加到另一个pcollection中，它是一个元素字典。

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

file_path=beam.create(['some_path_value'])并按照上面的方式传递beam.pvalue.assingleton(file_path)，它工作得很好。

但是我不想显式地创建pcollection。我需要使用上面的转换输出作为SideInput。

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方法内部。它主要发生在数据不会来自上述转换的情况下。但是我检查了很多次file_name有数据。

但是没有输出，只有当我使用beam.create([“pass_some_file_path”])并将其结果用作sideinput时才显示输出

Answer 1

转换的输出应该是2-tuple((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)))

你能用更好的方法解决它吗？