Apache-Beam将序列号添加到PCollection中

问题描述：

我试图构建一个ETL来加载一个Dimension表。我使用Python和DataFlow和BigQuery来分配Apache Bea。Apache-Beam将序列号添加到PCollection中

我需要为pcollection的每个元素分配一个序列号，以便将其加载到BigQuery中，但我找不到任何方法来执行此操作。

我想我需要DataFlow使先前的聚合和连接，以获得我最后的pcollection添加序列号，但在这一刻我需要停止并行处理，并将我的pcollection投到列表（如在Spark中，当你使用.collect（）），然后制作一个简单的循环来分配序列号。这样对吗？

这是管道，我编码：

p | ReadFromAvro(known_args.input) | beam.Map(adapt) | beam.GroupByKey() | beam.Map(adaptGroupBy)

我读过没有办法摆脱pcollection列表： How to get a list of elements out of a PCollection in Google Dataflow and use it in the pipeline to loop Write Transforms?

我怎样才能实现呢？任何帮助？

你能后你有什么到目前为止已经试过，和代码？ –

这是我使用Beam的第一种方法。我会添加我的一段代码，但我没有找到任何方法。 –

你能详细说明为什么你认为你需要添加序号吗？那么你计划在BigQuery中做什么，这将需要这个序列号？ –

答

如果您想要的是获取PCollection中每个元素的列表，则可以使用侧面输入。请记住，这将从您的结果中删除所有并行性，并且您的管道可能会变慢。

如果您仍然想这样做的话：

side_input_coll = beam.pvalue.AsIterable(my_collection) 

(p 
| beam.Create([0]) 
| beam.FlatMap(lambda _, my_seq: [(elem, i) for i, elem in enumerate(my_seq)], 
       my_seq=side_input_coll))

但是不要忘记，为了维护并行性，它可能是最好简单地生成一个随机ID。请记住PCollections本质上是无序的。

要了解更多关于侧面输入，看到Beam Programming Guide on Side Inputs

Apache-Beam将序列号添加到PCollection中

相关推荐