Apache管道中Python的CombineFn

Question

我一直在用Python中的Apache进行实验，以编写数据处理管道。

我的数据模拟了Google主题中的IoT传感器数据，该主题如下所示：

{"id": 1, "temperature": 12.34}
{"id": 2, "temperature": 76.54}

有从0到99的ID。将JSON读入Python并不是问题。

我创建了一个自定义CombineFn来由CombinePerKey进行处理。我希望我的累加器的输出将是计算，按PCollection中字典中相应的PCollection字段分组。

但是，当调用add_input方法时，它只接收字符串 temperature，而不是整个字典。我也没有找到任何引用来告诉CombinePerKey哪个键(在我的例子中是id字段)，我希望它对数据进行分组。

也许我也误解了CombinePerKey和CombineFn的概念。我希望能在这方面提供任何帮助或暗示。也许有人有一个用基于ID的分组处理JSON批的示例？我必须把字典翻译成别的东西吗？

Answer 1

您需要调整您的CombineFn或(我建议的)保持CombineFn尽可能通用，并相应地映射CombinePerKey的输入。我在下面使用这个公有梁的例子做了两个案例的简短例子。

特异CombineFn

import apache_beam as beam

class SpecificAverageFn(beam.CombineFn):
  def create_accumulator(self):
    sum = 0.0
    count = 0
    accumulator = sum, count
    return accumulator

  def add_input(self, accumulator, input):
    sum, count = accumulator
    extracted_input = input['temperature'] # <- this is a dict, you need to create custom code here
    return sum + extracted_input, count + 1

  def merge_accumulators(self, accumulators):
    # accumulators = [(sum1, count1), (sum2, count2), (sum3, count3), ...]
    sums, counts = zip(*accumulators)
    # sums = [sum1, sum2, sum3, ...]
    # counts = [count1, count2, count3, ...]
    return sum(sums), sum(counts)

  def extract_output(self, accumulator):
    sum, count = accumulator
    if count == 0:
      return float('NaN')
    return sum / count

with beam.Pipeline() as pipeline:
  (
    pipeline
    | "mock input" >> beam.Create([
     {'id': 1, 'temperature': 2},
     {'id': 2, 'temperature': 3},
     {'id': 2, 'temperature': 2}
    ])
    | "add key" >> beam.Map(lambda x: (x['id'], x))
    | beam.CombinePerKey(SpecificAverageFn())
    | beam.Map(print)
  )

通用Combinefn

import apache_beam as beam

class GenericAverageFn(beam.CombineFn):
  # everything as SpecificAverageFn, except add_input:
  def add_input(self, accumulator, input):
    sum, count = accumulator
    return sum + input, count + 1


with beam.Pipeline() as pipeline:
  iot_data = (
    pipeline
    | "mock input" >> beam.Create([
     {'id': 1, 'temperature': 2},
     {'id': 2, 'temperature': 3},
     {'id': 2, 'temperature': 2}
    ])
    | "add key" >> beam.Map(lambda x: (x['id'], x))
  )

  # repeat below for other values
  (
    iot_data
    | "extract temp" >> beam.Map(lambda x: (x[0], x[1]['temperature']) 
    | beam.CombinePerKey(AverageFn())
    | beam.Map(print)
  )

两种方法都回来了

(1, 2.0)
(2, 2.5)