如何通过侧面输入将两个 Pcollections(various sizes/data) 与一个公共 "key"(Street) 合并？

Question

我有两个 PCollections：一个从 Pub/Sub 中提取信息，另一个从 CSV 文件中提取数据。在每个管道中进行一些不同的转换之后，我想将两者合并到它们共享的公共密钥 "STREET" 上。我将第二个 PCollection 作为辅助输入。但是，我在尝试运行.

时遇到错误

我尝试使用 CoGroupByKey，但我一直收到有关 Pcollection 中数据类型差异的错误。我尝试重构输出，并通过 __setattr__ 设置 PCollection 的属性以强制类型相等，但无论如何它都报告了 "mixed values"。经过进一步研究，似乎最好使用侧输入，尤其是当元素之间的数据大小存在差异时。即使有侧面输入，我仍然无法克服当前错误：

from_runner_api raise ValueError('No producer for %s' % id)
ValueError: No producer for ref_PCollection_PCollection_6

我的应用逻辑如下：

def merge_accidents(element, pcoll):
    print(element)
    print(pcoll)
    "some code that will append to existing data"

accident_pl = beam.Pipeline()
accident_data = (accident_pl |
                        'Read' >> beam.io.ReadFromText('/modified_Excel_Crashes_Chicago.csv')
                        | 'Map Accidents' >> beam.ParDo(AccidentstoDict())
                        | 'Count Accidents' >> Count.PerKey())

chi_traf_pl = beam.Pipeline(options=pipeline_options)
chi_traffic = (chi_traf_pl | 'ReadPubSub' >> beam.io.ReadFromPubSub(subscription=subscription_name, with_attributes=True)
                           | 'GeoEnrich&Trim' >> beam.Map(loc_trim_enhance)
                           | 'TimeDelayEnrich' >> beam.Map(timedelay)
                           | 'TrafficRatingEnrich' >> beam.Map(traffic_rating)
                           | 'MergeAccidents' >> beam.Map(merge_accidents, pcoll=AsDict(accident_data))
                           | 'Temp Write'>> beam.io.WriteToText('testtime', file_name_suffix='.txt'))

accident_pl.run()
chi_result = chi_traf_pl.run()
chi_result.wait_until_finish()```

**Pcoll 1:**
[{'segment_id': '1', 'street': 'Western Ave', 'direction': 'EB', 'length': '0.5', 'cur_traffic': '24', 'county': 'Cook County', 'neighborhood': 'West Elsdon', 'zip_code': '60629', 'evnt_timestamp': '2019-04-01 20:50:20.0', 'traffic_rating': 'Heavy', 'time_delay': '0.15'}]
**Pcoll 2:**
('MILWAUKEE AVE', 1)
('CENTRAL AVE', 2)
('WESTERN AVE', 6)

**Expected:**
[{'segment_id': '1', 'street': 'Western Ave', 'direction': 'EB', 'length': '0.5', 'cur_traffic': '24', 'county': 'Cook County', 'neighborhood': 'West Elsdon', 'zip_code': '60629', 'evnt_timestamp': '2019-04-01 20:50:20.0', 'traffic_rating': 'Heavy', 'time_delay': '0.15', 'accident_count': '6'}]

**Actual Results:**
"from_runner_api raise ValueError('No producer for %s' % id)ValueError: No producer for ref_PCollection_PCollection_6

Answer 1

所以我想出了问题。在查看 pipeline.py 和侧输入的单元测试源后，我意识到有一个针对创建的 Pipeline 对象的检查。

我是新手，所以我最初认为您需要创建两个单独的 Pipeline 对象（流式与批处理），以便我可以将不同的选项传递给两者； i.e.streaming：没错。话虽如此，我认为没有必要。

将它们合并为如下所示的单个对象后，错误消失了，我能够接受函数的边输入：

'''

pipeline = beam.Pipeline(options=pipeline_options)
accident_data = (pipeline
                 | 'Read' >> beam.io.ReadFromText('modified_Excel_Crashes_Chicago.csv')
                 | 'Map Accidents' >> beam.ParDo(AccidentstoDict())
                 | 'Count Accidents' >> Count.PerKey())

chi_traffic = (pipeline
               | 'ReadPubSub' >> beam.io.ReadFromPubSub(subscription=subscription_name, with_attributes=True)
               | 'GeoEnrich&Trim' >> beam.Map(loc_trim_enhance)
               | 'TimeDelayEnrich' >> beam.Map(timedelay)
               | 'TrafficRatingEnrich' >> beam.Map(traffic_rating)
               | 'MergeAccidents' >> beam.Map(merge_accidents, pcoll=pvalue.AsDict(accident_data))
               | 'Temp Write' >> beam.io.WriteToText('testtime',
                                                     file_name_suffix='.txt'))

chi_result = pipeline.run()
chi_result.wait_until_finish()

'''

如何通过侧面输入将两个 Pcollections(various sizes/data) 与一个公共 "key"(Street) 合并？

How can I merge two Pcollections(various sizes/data) with a common "key"(Street) via a side input?

python

merge

pipeline

google-cloud-dataflow

apache-beam