python - 按元组顺序展平 PCollection

Question

我正在尝试使用 Apache Beam 中的 Flatten 函数添加 header 。但是，似乎没有办法根据文档设置顺序:https://beam.apache.org/documentation/sdks/pydoc/2.4.0/apache_beam.transforms.core.html?highlight=flatten#apache_beam.transforms.core.Flatten 。

有时标题位于数据的末尾，而其他则位于顶部。有没有办法设置顺序？想知道我是否遗漏了一些东西。

with beam.Pipeline(options=options) as p:


  header = [
      ('name', 'number'),
  ]
  phones_list = [
      ('amy', '111-222-3333'),
      ('james', '222-333-4444'),
      ('amy', '333-444-5555'),
      ('carl', '444-555-6666'),
  ]

  header = p | 'Header' >> beam.Create(header)
  phones = p | 'CreatePhones' >> beam.Create(phones_list)  

  merged = ((phones,header)
            | 'MergedPColl' >> beam.Flatten())

  output = merged

  output | 'Write' >> beam.io.WriteToText('./_output')

输出1:

('amy', '111-222-3333')
('james', '222-333-4444')
('amy', '333-444-5555')
('carl', '444-555-6666')
('name', 'number')

输出2:

('name', 'number')
('amy', '111-222-3333')
('james', '222-333-4444')
('amy', '333-444-5555')
('carl', '444-555-6666')

Answer 1

Flatten 是一个适用于 PCollection 的转换器。为了使合并工作并行进行，我认为他们不能保证顺序被保留；这与生成的 PCollection 的无序性质是一致的。

但是，如果您的唯一目的是在顶部添加 header ，则可以使用 textio.WriteToText() 的 header 参数。 .

> header (str): String to write at beginning of file as a header. If not :data:None and append_trailing_newlines is set, `\n`` will be added.

phones | 'Write' >> beam.io.WriteToText(
  # Feel free to make your own header format.
  './_output', header="('name', 'number')")

更一般地，为了保留原始输入的序列，我会使用序列号来扩充输入数据。在beam的并行变换(携带每个元素的序列号)之后，您始终可以通过对该序列号进行排序作为后处理步骤(在非并行模式下)来“恢复”原始顺序。