Spark提供了两种创建RDD的方式:读取外部数据集,以及在驱动器程序中对一个集合进行并行化。
在驱动器程序中对一个集合进行并行化的方式有两种:parallelize()和makeRDD()。
1、parallelize()
def parallelize[T: ClassTag](
seq: Seq[T],
numSlices: Int = defaultParallelism): RDD[T] = withScope {
assertNotStopped()
new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
}
2、makeRDD()
有两种重构方法,如下:
2.1、方法一:
/** Distribute a local Scala collection to form an RDD.
*
* This method is identical to `parallelize`.
*/
def makeRDD[T: ClassTag](
seq: Seq[T],
numSlices: Int = defaultParallelism): RDD[T] = withScope {
parallelize(seq, numSlices)
}
可以发现,该重构方法的实现就是调用parallelize()方法
2.2、方法二:
/**
* Distribute a local Scala collection to form an RDD, with one or more
* location preferences (hostnames of Spark nodes) for each object.
* Create a new partition for each collection item.
*/
def makeRDD[T: ClassTag](seq: Seq[(T, Seq[String])]): RDD[T] = withScope {
assertNotStopped()
val indexToPrefs = seq.zipWithIndex.map(t => (t._2, t._1._2)).toMap
new ParallelCollectionRDD[T](this, seq.map(_._1), math.max(seq.size, 1), indexToPrefs)
}
注释的意思为:分配一个本地Scala集合形成一个RDD,为每个集合对象创建一个最佳分区。
给出如下例子,可以更清晰的看到它们之间的区别:
首先定义集合对象:
val seq = List(("American Person", List("Tom", "Jim")), ("China Person", List("LiLei", "HanMeiMei")), ("Color Type", List("Red", "Blue")))
使用parallelize()创建RDD:
val rdd1 = sc.parallelize(seq)
查询rdd1的分区数:
rdd1.partitions.size
使用makeRDD()创建RDD
val rdd2 = sc.makeRDD(seq)
查看rdd2的分区数
rdd2.partitions.size
当调用parallelize()方法的时候,不指定分区数的时候,使用系统给出的分区数,而调用makeRDD()方法的时候,会为每个集合对象创建最佳分区,而这对后续的调用优化很有帮助。
版权声明:本文为beautiful_huang原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。