如何从 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 行行"的 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.

我尝试将计数器保存在 Datastore 中，但是，Dataflow 读取和写入 Datastore 之间的速度差异非常显着.

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屋！