如何在火花中按多个键分组？答案

【问题标题】：How to group by multiple keys in spark?如何在火花中按多个键分组？
【发布时间】：2015-06-05 00:54:07
【问题描述】：

我有一堆元组，它们是复合键和值的形式。例如，

tfile.collect() = [(('id1','pd1','t1'),5.0), 
     (('id2','pd2','t2'),6.0),
     (('id1','pd1','t2'),7.5),
     (('id1','pd1','t3'),8.1)  ]

我想在这个集合上执行类似 sql 的操作，我可以根据 id[1..n] 或 pd[1..n] 聚合信息。我想使用 vanilla pyspark apis 而不是使用 SQLContext 来实现。在我当前的实现中，我从一堆文件中读取并合并 RDD。

def readfile():
    fr = range(6,23)
    tfile = sc.union([sc.textFile(basepath+str(f)+".txt")
                        .map(lambda view: set_feature(view,f)) 
                        .reduceByKey(lambda a, b: a+b)
                        for f in fr])
    return tfile

我打算创建一个聚合数组作为值。例如，

agg_tfile = [((id1,pd1),[5.0,7.5,8.1])]

其中 5.0,7.5,8.1 表示 [t1,t2,t3] 。我目前正在通过使用字典的香草 python 代码实现相同的目标。它适用于较小的数据集。但我担心这可能不适用于更大的数据集。有没有使用 pyspark apis 实现相同目标的有效方法？

【问题讨论】：

与union相比，使用wholeTextFiles调用（如果PySpark中存在）加载所有文件更有效。
这是一个 Scala aggregate reduction by multiple keys 和一个 Python union reduction of multiple values

标签： python apache-spark pyspark

【解决方案1】：

我的猜测是你想根据多个字段转置数据。

一种简单的方法是将要分组的目标字段连接起来，并使其成为配对 RDD 中的键。例如：

lines = sc.parallelize(['id1,pd1,t1,5.0', 'id2,pd2,t2,6.0', 'id1,pd1,t2,7.5', 'id1,pd1,t3,8.1'])
rdd = lines.map(lambda x: x.split(',')).map(lambda x: (x[0] + ', ' + x[1], x[3])).reduceByKey(lambda a, b: a + ', ' + b)
print rdd.collect()

然后你会得到转置的结果。

[('id1, pd1', '5.0, 7.5, 8.1'), ('id2, pd2', '6.0')]

【讨论】：

这绝对是一个有趣的解决方法。我想出了另一种实现相同目标的方法。但我想，你的方法可能比我的要快得多。我也在分享我自己的解决方案。
PySpark 没有groupByKey吗？
PySpark 有方法groupBykey。但是，该问题倾向于根据两个字段将记录分组在一起，而不是进行聚合，例如SELECT sum(value) FROM data GROUP BY id, pd。所以groupBykey 可能没有帮助。

【解决方案2】：

我将 ((id1,t1),((p1,5.0),(p2,6.0)) 等...分组为我的地图函数。后来，我使用 map_group 为 [p1, p2, . . . ] 并在各自的位置填充值。

def map_group(pgroup):
    x = np.zeros(19)
    x[0] = 1
    value_list = pgroup[1]
    for val in value_list:
        fno = val[0].split('.')[0]
        x[int(fno)-5] = val[1]
    return x

tgbr = tfile.map(lambda d: ((d[0][0],d[0][2]),[(d[0][1],d[1])])) \
                .reduceByKey(lambda p,q:p+q) \
                .map(lambda d: (d[0], map_group(d)))

就计算而言，这确实是一种昂贵的解决方案。但暂时有效。

【讨论】：