从这个链接中,Guillem Xercavins编写了一个自定义类来计算最小值和最大值。
class MinMaxFn(beam.CombineFn):
# initialize min and max values (I assumed int type)
def create_accumulator(self):
return (sys.maxint, 0)
# update if current value is a new min or max
def add_input(self, min_max, input):
(current_min, current_max) = min_max
return min(current_min, input), max(current_max, input)
def merge_accumulators(self, accumulators):
return accumulators
def extract_output(self, min_max):
return min_max
我也需要计算平均值,我找到了以下示例代码:
class MeanCombineFn(beam.CombineFn):
def create_accumulator(self):
"""Create a "local" accumulator to track sum and count."""
return (0, 0)
def add_input(self, (sum_, count), input):
"""Process the incoming value."""
return sum_ + input, count + 1
def merge_accumulators(self, accumulators):
"""Merge several accumulators into a single one."""
sums, counts = zip(*accumulators)
return sum(sums), sum(counts)
def extract_output(self, (sum_, count)):
"""Compute the mean average."""
if count == 0:
return float('NaN')
return sum_ / float(count)
有什么办法可以将平均方法合并到MinMax中,以便我只需要一个类来计算最小值、最大值和平均值,并生成一组键和值 - 3个值的数组?
beam.Create([1, 2, 3, 4, 5])
创建了一个管道,那么我该如何在这个PCollection上调用MinMaxMeanFn方法呢? - xennygrimmato