如何通过Apache Beam（Python）中的键以流模式在静态查找表上连接PCollection

Question

我正在以字典的形式将（无界）数据从谷歌云Pubsub流进一个PCollection。当流数据进来时，我想通过在一个静态（有界）查找表上按键连接它来丰富它。这个表足够小，可以保存在内存中。

我已经使用beam.io.ReadFromText函数从csv中读取了有界查找表，并将值解析到字典中。然后，我创建了一个pardo函数，该函数将我的无界pcollection和查找字典作为侧输入。在pardo中，它使用生成器“联接”查找表的正确行，并充实输入元素。

以下是主要部分…

# Get bounded lookup table
lookup_dict = (pcoll | 'Read PS Table' >> beam.io.ReadFromText(...) 
| 'Split CSV to Dict' >> beam.ParDo(SplitCSVtoDict()))

# Use lookup table as side input in ParDo func to enrich unbounded pcoll
# I found that it only worked on my local machine when decorating it with AsList
enriched = pcoll | 'join pcoll on lkup' >> beam.ParDo(JoinLkupData(), data=beam.pvalue.AsList(lookup_dict)

class JoinLkupData(beam.DoFn):
    def process(self, element, lookup_data):
        # I used a generator here
        lkup = next((row for row in lookup_data if row[<JOIN_FIELD>]) == element[<JOIN_FIELD>]), None)

        if lkup:
           # If there is a join, add new fields to the pcoll
           element['field1'] = lkup['field1']
           element['field2'] = lkup['field2']
        yield element

在本地使用DirectRunner运行时，我能够获得正确的结果，但在DataFlow运行器上运行时，我收到以下错误：

apache_beam.runners.Dataflow.dataflow_runner.dataflowRuntimeException：数据流管道失败。状态：失败，错误：工作流失败。原因：预期自定义源的拆分数为非零。

这篇文章：“在Dataflow runner上拆分pcollections时出错”，让我想到这个错误的原因与多个工作人员在拆分工作时无法访问相同的查找表有关。

Answer 1

将来，如果可以的话，请分享Beam的版本和堆栈跟踪。

在这种情况下，错误消息不是很好是一个已知的问题。在撰写本文时，Python流数据流仅限于Pubsub用于读写，BigQuery用于写。在管道中使用文本源会导致此错误。