Pyspark RDD 以不同方式聚合不同的值字段

Question

这是一个非常开放的问题，但我有一个这种格式的 RDD。

[('2014-06', ('131313', 5.5, 6.5, 7.5, 10.5 )),
('2014-07', ('246655', 636636.53, .53252, 5252.112, 5242.23)),
('2014-06', ('131232', 1, 2, 4.5, 5.5)),
('2014-07', ('131322464', 536363.6363, 536336.6363, 3563.63636, 9.6464464646464646))]

我想通过键对每个值进行不同的分组和聚合。例如，对于键 '2014-06'，我想获取第一个值字段的计数，即 '131313'，以及键 '2014-06' 的其他字段 5.5, 6.5, 7.5, 10.5 的平均值。

因此，上述简单示例的键 '2014-06' 的结果将是 ('2014-06', (2, 3.25, 5.5, 8))。

对 RDD 执行此操作的最佳方法是什么？ 我不能使用任何 Spark SQL 表达式或函数，只能使用 RDD 函数。

我正在考虑用 mapValues 做一些事情并使用其他函数，但我在制定这个函数时遇到了一些问题。

我知道这个问题是开放式的，所以如果您还有其他问题，请告诉我。

感谢您的宝贵时间。

Answer 1

一种方法是使用 ma​​p() 方法将第一个值转换为 1（用于记录计数），然后使用 reduceByKey() 将每个值与同一把钥匙。最后，使用 ma​​pValues() 计算平均值，除了第一个是计数（保持原样）。

rdd.map(lambda x: (x[0], (1, *x[1][1:]))) \
   .reduceByKey(lambda x,y: tuple([x[i]+y[i] for i in range(len(x))])) \
   .mapValues(lambda x: (x[0], *[ e/x[0] for e in x[1:]])) 

ma​​p() 之后：

[('2014-06', (1, 5.5, 6.5, 7.5, 10.5)),
 ('2014-07', (1, 636636.53, 0.53252, 5252.112, 5242.23)),
 ('2014-06', (1, 1, 2, 4.5, 5.5)),
 ('2014-07', (1, 536363.6363, 536336.6363, 3563.63636, 9.646446464646464))]

reduceByKey() 之后：

[('2014-06', (2, 6.5, 8.5, 12.0, 16.0)),
 ('2014-07',
  (2, 1173000.1663000002, 536337.16882, 8815.74836, 5251.876446464646))]

ma​​pValues() 之后：

[('2014-06', (2, 3.25, 4.25, 6.0, 8.0)),
 ('2014-07',
  (2, 586500.0831500001, 268168.58441, 4407.87418, 2625.938223232323))]

Answer 2

@jxc 解决方案可以满足您的需求，但这是另一种方式。

您可以使用aggregateByKey。该函数接受两个函数seqFunc、combFunc 和一个称为中性零值的累加器值。

zero_value = (0, 0, 0, 0, 0)
d = rdd.aggregateByKey(zero_value, lambda x, y: (1, *y[1:]),
                       lambda x, y: tuple(map(add, x, y))
                       ) \
    .mapValues(lambda v: (v[0], *[i / v[0] for i in v[1:]])) \

第一个 lambda 表达式通过将第一个字符串字段替换为整数 1（计数一次）来转换每个值。 第二个 lambda 表达式通过添加两个列表来合并两个值。

在这个聚合之后，我们只需要将每个值列表的元素除以给出平均值的第一个元素。

输出：

[('2014-06', (2, 3.25, 4.25, 6.0, 8.0)), ('2014-07', (2, 586500.0831500001, 268168.58441, 4407.87418, 2625.938223232323))]