Pyspark：合并嵌套列表中的值

Posted 2023-04-15

【中文标题】Pyspark：合并嵌套列表中的值

【英文标题】：Pyspark: merging values in a nested list

【发布时间】：2016-11-18 07:04:37

【问题描述】：

我有一个具有以下结构的对 RDD： [（键，[（时间字符串，值）]]

例子：

[("key1", [("20161101", 23), ("20161101", 41), ("20161102", 66),...]),
 ("key2", [("20161101", 86), ("20161101", 9), ("20161102", 11),...])
  ...]

我想处理每个键的列表，按时间字符串分组并计算相同时间字符串的所有值的平均值。所以上面的例子会变成：

[("key1", [("20161101", 32), ..]),
 ("key2", [("20161101", 47.5),...])
  ...]

我很难在一个步骤中仅使用 Pyspark 方法找到解决方案，这完全可能还是我需要使用一些中间步骤？

【参考方案1】：

你可以定义一个函数：

from itertools import groupby
import numpy as np

def mapper(xs):
    return [(k, np.mean([v[1] for v in vs])) for k, vs in groupby(sorted(xs), lambda x: x[0])]

还有mapValues

rdd = sc.parallelize([
    ("key1", [("20161101", 23), ("20161101", 41), ("20161102", 66)]),
    ("key2", [("20161101", 86), ("20161101", 9), ("20161102", 11)])
])

rdd.mapValues(mapper)