4

我有一个从 AVRO 文件中解析记录的管道。

我需要将传入的记录拆分为 500 个项目的块，以便调用同时接受多个输入的 API。

有没有办法用 Python SDK 做到这一点？

4

1 回答 1

3

我假设您的意思是批处理用例。你有几个选择：

如果您的 PCollection 足够大，并且您对捆绑包的大小有一定的灵活性，则可以GroupByKey在将随机/循环顺序的键分配给您的元素后使用转换。例如：

my_collection = p | ReadRecordsFromAvro()

element_bundles = (my_collection 
                     # Choose a number of keys that works for you (I chose 50 here)
                   | 'AddKeys' >> beam.Map(lambda x: (randint(0, 50), x))
                   | 'MakeBundles' >> beam.GroupByKey()
                   | 'DropKeys' >> beam.Map(lambda (k, bundle): bundle)
                   | beam.ParDo(ProcessBundlesDoFn()))

ProcessBundlesDoFn像这样的东西在哪里：

class ProcessBundlesDoFn(beam.DoFn):
  def process(self, bundle):
    while bundle.has_next():
      # Fetch in batches of 500 until you're done
      result = fetch_n_elements(bundle, 500)
      yield result

如果您需要拥有正好 500 个元素的所有捆绑包，那么您可能需要：

  1. 计算 PCollection 中的元素数
  2. 将该计数作为单例侧输入传递给'AddKeys'ParDo，以准确确定您需要的键数。

希望有帮助。

于 2017-07-31T17:33:42.337 回答