以下提供的代码,我正在尝试按每月基础计算浮点列的平均值。
rdd = sc.parallelize(
[['JAN', 'NY', 3.0],
['JAN', 'PA', 1.0],
['JAN', 'NJ', 2.0],
['JAN', 'CT', 4.0],
['FEB', 'PA', 1.0],
['FEB', 'NJ', 1.0],
['FEB', 'NY', 2.0],
['FEB', 'VT', 1.0],
['MAR', 'NJ', 2.0],
['MAR', 'NY', 1.0],
['MAR', 'VT', 2.0],
['MAR', 'PA', 3.0]])
def avg_map(row):
return (row[0], (row[2], 1))
def avg_reduce_func(value1, value2):
return (value1[0], (value1[1][0] + value2[1][0], value1[1][1] + value2[1][1]))
dataset_rdd.map(avg_map_func).reduceByKey(avg_reduce_func).collect()
从宏观角度来看,我首先尝试使用map函数创建以下形式的RDD:
[('JAN', (3.0, 1)),
('JAN', (1.0, 1)),
('JAN', (2.0, 1)),
('JAN', (4.0, 1)),
('FEB', (1.0, 1)),
('FEB', (1.0, 1)),
('FEB', (2.0, 1)),
('FEB', (1.0, 1)),
('MAR', (2.0, 1)),
('MAR', (1.0, 1)),
('MAR', (2.0, 1)),
('MAR', (3.0, 1))]
然后,我想使用 reduceByKey 函数按键添加1和浮点数,创建一个新的 RDD,其中每个月将包含一行,表示元组中的浮点数总计和表示行数的整数。例如,Jan 行看起来像这样:
('Jan',(10.0,4))
然而,我似乎无法正确地索引到元组,并最终在 reduceByKey 函数中出现运行时错误。
问题1:为什么我无法在 avg_reduce_func 中对元组进行索引? 问题2:如何重新编写此代码以按每月基础计算浮点数列的平均值?