如何知道 Spark 使用 Scala 推断出的 RDD 类型是啥

Posted 2023-04-18

【中文标题】如何知道 Spark 使用 Scala 推断出的 RDD 类型是啥

【英文标题】：How to know which is the RDD type inferred by Spark using Scala如何知道 Spark 使用 Scala 推断出的 RDD 类型是什么

【发布时间】：2016-07-20 07:16:51

【问题描述】：

我正在尝试以下示例

val lista = List(("a", 3), ("a", 1), ("b", 7), ("a", 5))
val rdd = sc.parallelize(lista)

然后在shell中我得到以下内容

rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[40] at parallelize at <console>:30

但由于某种原因，我仍然没有想出我能够执行这句话

val resAgg = rdd.aggregateByKey(new HashSet[Int])(_+_, _++_)

在 shell 中获取它

resAgg: org.apache.spark.rdd.RDD[(String, scala.collection.mutable.HashSet[Int])] = ShuffledRDD[41] at aggregateByKey at <console>:32

所以我有一些问题：

1.- 名为 rdd 的 var 的真正 RDD 类型是什么？因为在 shell 中它显示的类型是 org.apache.spark.rdd.RDD[(String, Int)] 但查看 API 时，RDD 类没有方法 aggregateByKey。顺便说一句，JavaPairRDD 类是否有 aggregateByKey 方法

2.- 我如何验证/知道 RDD 的真实类型

3.- ParallelCollectionRDD 出现了什么？我在 github 上查找它，发现是一个私有类，所以我猜是 scala API 上没有出现 is 的原因，但它的用途是什么？

我使用的是 Spark 1.6.2

【参考方案1】：

你看到的是implicit conversion的效果：

rdd 确实有org.apache.spark.rdd.RDD[(String, Int)]的类型

当您尝试调用 aggregateByKey 并且该类型不存在时，编译器会查找某种类型的隐式转换 - 并找到将 this 转换为 PairRDDFunctions 的类型：

implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])
  (implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V] = 
  new PairRDDFunctions(rdd)

然后，PairRDDFunctions.aggregateByKey 被调用。

关于你的最后一个问题：

ParallelCollectionRDD 是什么

RDD 是一个包含许多子类的抽象类，这是其中之一。一般来说，每个子类负责对 RDD 执行的不同操作，例如读/写/洗牌/检查点等。这种特定类型在调用SparkContext.parallelize时使用——意思是，它用于并行化来自驱动程序的集合。实际上，它是私有的，您通常不应该关心您手头实际拥有的 RDD 子类型。

【讨论】：

感谢您的回答，但我仍然对 aggregationByKey 方法调用存有疑问。 RDD 和 PairRDDFunctions 类之间的关系如何？我在关于 PairRDDFunctions 的 API 上有红色这个“通过隐式转换在（键，值）对的 RDD 上可用的额外函数”。但我的意思是我仍然看不到这些类之间是如何建立关系的（除了 Logging Trait 之外）。你介意给我一个线索或参考来尝试理解它。也许这是我还没有红色的scala属性，不是吗？提前致谢

没有直接关系（即它们都没有扩展/实现另一个），请在我刚刚添加的链接中阅读有关隐式转换的更多信息。基本上，隐式转换让你调用rdd.aggregateAndSave(...)，就好像你在调用new PairRDDFunctions(rdd).aggregateAndSave(...)

再次感谢，我要研究scala隐函数，现在明白了。