python - Apache Beam 最小值、最大值和平均值
问题描述
在这个链接中,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 个值的数组?
解决方案
这是组合类解决方案,添加了中位数
import numpy as np
class MinMaxMeanFn(beam.CombineFn):
def create_accumulator(self):
# sum, min, max, count, median
return (0.0, 999999999.0, 0.0, 0, [])
def add_input(self, cur_data, input):
(cur_sum, cur_min, cur_max, count, cur_median) = cur_data
if type(input) == list:
cur_count = len(input)
sum_input = sum(input)
min_input = min(input)
max_input = max(input)
else:
sum_input = input
cur_count = 1
return cur_sum + sum_input, min(min_input, cur_min), max(max_input, cur_max), count + cur_count, cur_median + input
def merge_accumulators(self, accumulators):
sums, mins, maxs, counts, medians = zip(*accumulators)
return sum(sums), min(mins), max(maxs), sum(counts), medians
def extract_output(self, cur_data):
(sum, min, max, count, medians) = cur_data
avg = sum / count if count else float('NaN')
med = np.median(medians)
return {
"max": max,
"min": min,
"avg": avg,
"count": count,
"median": med
}
示例用法:
( input |'Format Price' >> beam.ParDo(FormatPriceDoFn())
|'Group Price by ID' >> beam.GroupByKey()
|'Compute price statistic for each ID' >> beam.CombinePerKey(MinMaxMeanFn()))
*我没有测试 CombinePerKey 是否在没有 GroupByKey 的情况下工作,请随意测试。
推荐阅读
- ios - Swift 异步打印顺序?
- dynamics-crm - 我们如何在 Dynamic 365 中使用区分大小写的数据执行高级搜索
- android - 如何在服务中获取资产?
- javascript - 删除javascript中url的第一段和最后一段
- django - Django:一次性在所有应用程序上运行 pylint
- ember.js - 当我刷新页面时,Ember-simple-auth 会话过期
- sql - BigQuery:使用 COUNT 作为 LIMIT
- java - 通过 HTTP 在 Java 上放置
- sql-server - 对索引字段执行 sp_prepexec 查询时性能不佳,为什么?
- sql - SQL - 使用 Where 子句连接两个视图时查询性能变慢