scala - 如何将 RDD[(Key, Value)] 转换为 Map[Key, RDD[Value]]

Question

我搜索了很长时间的解决方案，但没有得到任何正确的算法。

在 Scala 中使用 Spark RDD，如何转换 RDD[(Key, Value)]变成 Map[key, RDD[Value]] ，知道我不能使用收集或其他可能将数据加载到内存中的方法吗？

事实上，我的最终目标是循环播放 Map[Key, RDD[Value]]按键并调用saveAsNewAPIHadoopFile对于每个 RDD[Value]
例如，如果我得到:

RDD[("A", 1), ("A", 2), ("A", 3), ("B", 4), ("B", 5), ("C", 6)]

我想要 :

Map[("A" -> RDD[1, 2, 3]), ("B" -> RDD[4, 5]), ("C" -> RDD[6])]

我想知道使用 filter 做这件事会不会花费太多。在 RDD[(Key, Value)] 的每个键 A、B、C 上，但我不知道是否多次调用过滤器有不同的键会有效吗？ (当然不是，但也许使用 cache ？)

谢谢

Answer 1

您应该使用这样的代码(Python):

rdd = sc.parallelize( [("A", 1), ("A", 2), ("A", 3), ("B", 4), ("B", 5), ("C", 6)] ).cache()
keys = rdd.keys().distinct().collect()
for key in keys:
    out = rdd.filter(lambda x: x[0] == key).map(lambda (x,y): y)
    out.saveAsNewAPIHadoopFile (...)

一个 RDD 不能是另一个 RDD 的一部分，您无法选择仅收集键并将其相关值转换为单独的 RDD。在我的示例中，您将遍历缓存的 RDD，这没问题并且可以快速运行