在 PySpark 中为 KeyVal RDD 中的每个键收集前 N 个条目答案

【问题标题】：Collecting first N entries for each key in KeyVal RDD in PySpark在 PySpark 中为 KeyVal RDD 中的每个键收集前 N 个条目
【发布时间】：2019-06-17 14:19:39
【问题描述】：

我有一个包含大量 KeyVal 条目的 RDD。同一个键将出现多次，我有兴趣为每个键提取前 N 个条目。作为 Spark 的新手，到目前为止我还无法弄清楚如何做到这一点，因此我们将不胜感激。

输入可能如下所示：

rdd = sc.parallelize([('a',1),('a',2),('b',3),('a',5),('b',4),('b',6)])

例如，每个键的前 2 个条目所需的最终输出：

output = {'a':[1,2], 'b':[3,4]}

如果我一般只对前 n 个条目感兴趣，我当然可以将 take(n) 应用于初始 RDD。我正在寻找的是一种类似于 take(n) 函数的方法，但迭代每个键。如果我可以创建一个仅包含原始 RDD 所需子集的 RDD，那就没问题了。事后收集到字典不是问题。

所以中间输出（RDD 风格）将是：

[('a',1),('a',2),('b',3),('b',4)]

如何在 PySpark 中实现这一点？

编辑：建议的重复问题特别需要使用 reduceByKey 的解决方案，这在这种情况下不是必需的。

【问题讨论】：

PYSPARK : use sort with reduceByKey的可能重复

标签： python apache-spark pyspark

【解决方案1】：

听着，孩子……

nLength = 2
rdd.groupByKey().map(lambda x: (x[0], list(x[1])[:nLength]))

说明：

rdd.groupByKey()

按键对 RDD 进行分组（在我们的例子中：'a' 或 'b'）。结果：
[('a', ResultIterable), ('b', ResultIterable)]

.map(lambda x: (x[0], list(x[1])[:nLength]))

这部分创建了一个元组：在左侧，键（'a' 或 'b'），在右侧，我们从 ResultIterable (x[1]) 创建一个列表，然后切割从 0 到 nLength ([:nLength]) 的列表。

享受吧！

【讨论】：

【解决方案2】：

也许这样简单的事情就可以完成这项工作：

rdd = sc.parallelize([('a',1),('a',2),('b',3),('a',5),('b',4),('b',6)])
n = 2
rdd.groupByKey().map(lambda x : (x[0], list(x[1])[:n])).collect()

输出：

[('b', [3, 4]), ('a', [1, 2])]

【讨论】：

【解决方案3】：

试试这个：

 def slice_list(s,no_of_values):
        return s[0:no_of_values]


rdd.groupByKey().map(lambda x: (x[0],slice_list( list(x[1]),2))).collect()

【讨论】：