如何从PCollection Apache Beam Python创建N个元素的组 [英] How to create groups of N elements from a PCollection Apache Beam Python
问题描述
我正在尝试完成以下操作: 在Beam/Dataflow中批处理PCollection
I am trying to accomplish something like this: Batch PCollection in Beam/Dataflow
以上链接的答案是Java,而我使用的语言是Python.因此,我需要一些帮助来获得类似的构造.
The answer in the above link is in Java, whereas the language I'm working with is Python. Thus, I require some help getting a similar construction.
具体地说,我有这个:
p = beam.Pipeline (options = pipeline_options)
lines = p | 'File reading' >> ReadFromText (known_args.input)
在此之后,由于我的用例需要一组行,因此我需要创建另一个PCollection
,但要用N行的"c1"行作为List
.我无法逐行操作.
After this, I need to create another PCollection
but with a List
of N rows of "lines" since my use case requires a group of rows. I can not operate line by line.
我尝试了ParDo
函数,该函数使用变量将计数与计数器N行相关联,并在groupBy
之后使用Map
.但是这些每1000条记录会重置一次,所以这不是我想要的解决方案.我阅读了链接中的示例,但是我不知道如何在Python中执行类似的操作.
I tried a ParDo
Function using variables for count associating with the counter N rows and after groupBy
using Map
. But these are reset every 1000 records, so it's not the solution I am looking for. I read the example in the link but I do not know how to do something like that in Python.
我尝试将计数器保存在数据存储区中,但是,使用数据存储区读取和写入数据流之间的速度差异非常明显.
I tried saving the counters in Datastore, however, the speed difference between Dataflow reading and writing with Datastore is quite significant.
执行此操作的正确方法是什么?我不知道该怎么办. 问候.
What is the correct way to do this? I don't know how else to approach it. Regards.
推荐答案
假设分组顺序并不重要,您可以将DoFn
分组.
Assume the grouping order is not important, you can just group inside a DoFn
.
class Group(beam.DoFn):
def __init__(self, n):
self._n = n
self._buffer = []
def process(self, element):
self._buffer.append(element)
if len(self._buffer) == self._n:
yield list(self._buffer)
self._buffer = []
def finish_bundle(self):
if len(self._buffer) != 0:
yield list(self._buffer)
self._buffer = []
lines = p | 'File reading' >> ReadFromText(known_args.input)
| 'Group' >> beam.ParDo(Group(known_args.N)
...
这篇关于如何从PCollection Apache Beam Python创建N个元素的组的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!