Apache Beam 全局计数器

Question

我有一个包含百万条记录的文件，其中一些记录是不良记录（在 ParDo 中处理记录时会知道这些记录）。我想将坏记录连同它们出现在文件中的行号一起写入单独的 PCollection，并将好的记录写入单独的 PCollection。

有没有办法维护一个全局计数器，记录到目前为止跨工作进程读取的行数，以便我可以用它来写出行号？

Answer 1

您可以使用 Apache Beam 指标来保存一个全局监控计数器，您可以从您的机器或跑步者的 UI 中查询该计数器。

如果您想保留所有不良记录的精确集合，以及有关它们的信息（例如行号），那么您需要添加一个可以让您执行此操作的转换。像这样的：

original_records = p | LoadRecords()

class SplitRecords(beam.DoFn):
  BAD_RECORD_TAG = 'BadRecord'

  def process(self, record):
    if self.is_bad(record):
      # Output the record onto the 'special' BadRecord input.
      yield beam.pvalue.TaggedOutput(self.BAD_RECORD_TAG, record)
    else:
      yield record   # Output the record onto the main input

record_collections = (original_records | 
                      beam.ParDo(SplitRecords()).with_outputs(
                          SplitRecords.BAD_RECORD_TAG,
                          main='GoodRecords'))

bad_records = record_collections[SplitRecords.BAD_RECORD_TAG]

good_records = record_collections['GoodRecords']

对于更详细的示例，我建议您查看 Apache Beam 食谱目录，其中包含 example with a multiple-output ParDo