如何将 Iterable 转换为 RDD答案

【问题标题】：How to convert an Iterable to an RDD如何将 Iterable 转换为 RDD
【发布时间】：2016-05-13 11:26:42
【问题描述】：

更具体地说，如何将 scala.Iterable 转换为 org.apache.spark.rdd.RDD ？

我的 RDD 为 (String, Iterable[(String, Integer)]) 我希望将其转换为 (String, RDD[String, Integer]) 的 RDD，以便我可以将 reduceByKey 函数应用于内部 RDD.

例如我有一个 RDD，其中键是人名的 2 个字母前缀，值是成对的人名和他们在事件中花费的时间的列表

我的 RDD 是：

("To", List(("Tom",50),("Tod","30"),("Tom",70),("Tod","25"),("Tod",15)) ("Ja", List(("Jack",50),("James","30"),("Jane",70),("James","25"),("Jasper",15))

我需要将列表转换为 RDD，以便我可以使用累积每个人花费的总小时数。应用 reduceByKey 并使结果为 ("To", RDD(("Tom",120),("Tod","70")) ("Ja", RDD(("Jack",120),("James","55"),("Jane",15))

但我找不到任何这样的转换函数。我怎样才能做到这一点？

提前致谢。

【问题讨论】：

我经历了stackoverflow.com/questions/33284507/… 但 sparkContext.parallelize 是如何尝试创建多个 sparkContext 并因此导致异常。我想在我的应用程序中保留 sparkContext 的单个实例。
你为什么需要那个？
不可能。你不能嵌套 RDD。
那么，有没有其他方法可以从 Iterables 实现这个解决方案（无需转换为 RDD）？
flatMap 将您的 RDD 转换为 ((prefix, name), hours)，reduceByKey 以获取每个用户的总数，map 以获取 (prefix , (name, total))，aggegrateByKey（或只是groupBy）获取(prefix, List((name1, total), (name2, total),...)。容易。

标签： scala hadoop apache-spark aggregate-functions rdd

【解决方案1】：

您可以通过使用flatMap 和reduceByKey 来实现此目的。像这样的：

rdd.flatMap{case(key, list) => list.map(item => ((key,item._1), item._2))}
   .reduceByKey(_+_)
   .map{case((key,name),hours) => (key, List((name, hours)))}
   .reduceByKey(_++_)

【讨论】：

它就像一个魅力。非常感谢。由于我是 scala 的新手，我可以理解这个解决方案，但不是很清楚。我的一些问题包括：“案例”在这里做什么？最后的“++”运算符是否实际上将元素添加到相同前缀的列表中？最后一个是：我如何对结果进行排序？
case 只是普通的模式匹配，++ 附加两个列表，你可以使用sortBy 对结果进行排序:)
@Puneet Chuarasia 编辑了我的答案，使它实际上是错误的（！！）。实际上需要“（名称，小时）” ARE 周围的括号，因为它需要映射到元组列表，而不是 Any 类型的列表。我已将答案改回原来的正确答案。
对此表示歉意并感谢您的描述。 :)
一般来说，编辑别人的答案是一种非常糟糕的风格（如果你改变了他们的答案之一，SO 上的许多高技能的人会大发雷霆！）。如果您觉得需要添加或澄清某些内容，则应将其放在评论中，并让撰写答案的人决定是否需要更改答案。