sc.parallelize 和 sc.markRDD
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]]())
}
makeRDD()，有两种重构方法
/** 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)
}
/**
* 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，为每个集合对象创建一个最佳分区
测试使用
object MyTask2 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("rdd maker").setMaster("local")
val sc = new SparkContext(conf)
val list = List(("A",List("a1","a2","a3")),("B",List("b1","b2","b3"),("C",List("c1","c2","c3"))))
val rddmaker = sc.makeRDD(list)
val rddP = sc.parallelize(list)
println("rddmaker partitions size:",rddmaker.partitions.size)
println("rddP partitions size:",rddP.partitions.size)
}
}
//(rddmaker partitions size:,1)
//(rddP partitions size:,1)
distinct
代码
object MyTask3 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("task3"))
println("rdd partitions size is ",rdd.partitions.size)
val rdd = sc.parallelize(List("a","b","c","b","b","a"))
rdd.distinct().collect().foreach(print(_))
}
}
//(rdd partitions size is ,1)
//abc
union
代码
object MyTask4 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task4"))
val rddLeft = sc.parallelize(List("2","3","4","5"))
val rddRight = sc.parallelize(List("1","3","5","7"))
val rddUnion = rddLeft.union(rddRight)
rddUnion.collect().foreach(item => print(item + ","))
}
}
//2,3,4,5,1,3,5,7,
intersection 求交集
object MyTask5 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task4"))
val rddLeft = sc.parallelize(List("2","3","4","5"))
val rddRight = sc.parallelize(List("1","3","5","7"))
val rddIntersec = rddLeft.intersection(rddRight)
rddIntersec.collect().foreach(item => print(item + ","))
}
}
//5,3,
subtract 把Rdd中的与另一个Rdd相同的元素去掉
代码
object MyTask6 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task6"))
val rddLeft = sc.parallelize(List("2","3","4","5"))
val rddRight = sc.parallelize(List("1","3","5","7"))
val rddSubtract = rddLeft.subtract(rddRight)
rddSubtract.collect().foreach(item => print(item + ","))
}
}
//2,4,
cartesian 笛卡尔积
代码
object MyTask7 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task7"))
val rddLeft = sc.parallelize(List("2","3","4","5"))
val rddRight = sc.parallelize(List("1","3","5","7"))
val rddCartesian = rddLeft.cartesian(rddRight)
rddCartesian.collect().foreach(item => print(item + ","))
}
}
//(2,1),(2,3),(2,5),(2,7),(3,1),(3,3),(3,5),(3,7),(4,1),(4,3),(4,5),(4,7),(5,1),(5,3),(5,5),(5,7)
countByValue 求出value出现的次数
代码
object MyTask8 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task8"))
val rddLeft = sc.parallelize(List("2","3","4","5"))
val rddRight = sc.parallelize(List("1","3","5","7"))
val rddUnion = rddLeft.union(rddRight)
val rddCountByValue:scala.collection.Map[String, scala.Long] = rddUnion.countByValue
rddCountByValue.foreach(item => println(item._1 + "," + item._2))
}
}
/*
4,1
5,2
1,1
2,1
7,1
3,2
*/
reduce 并行计算出函数
代码
object MyTask9 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task9"))
val rdd = sc.parallelize(1 to 11)
val result = rdd.reduce((x,y) => x+y)
println(result)
}
}
//66
fold
代码
object MyTask10 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task10"))
val rdd = sc.parallelize(1 to 11,2)
val result = rdd.fold(10)(_+_)
println(result)
}
}
//96
解释，与reduce类似，只是多了一个初始值。
aggregate
函数签名
def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
解释：
aggregate先对每个分区的元素做聚集，然后对所有分区的结果做聚集，聚集过程中，使用的是给定的聚集函数以及初始值”zero value”。这个函数能返回一个与原始RDD不同的类型U，因此，需要一个合并RDD类型T到结果类型U的函数，还需要一个合并类型U的函数。这两个函数都可以修改和返回他们的第一个参数，而不是重新新建一个U类型的参数以避免重新分配内存。
参数zeroValue：seqOp运算符的每个分区的累积结果的初始值以及combOp运算符的不同分区的组合结果的初始值 - 这通常将是初始元素（例如“Nil”表的列表 连接或“0”表示求和）
参数seqOp： 每个分区累积结果的聚集函数。
参数combOp： 一个关联运算符用于组合不同分区的结果
代码
object MyTask11 {
def main(args: Array[String]): Unit = {
val sc = new SparkContext(new SparkConf().setMaster("local[1]").setAppName("task11"))
val rdd = sc.parallelize(1 to 4,3)
val result = rdd.aggregate((0,0,0))(
(acc,number) => (acc._1+number,acc._1,acc._3+1),
(x,y) => (x._1 + y._1,x._2 + y._2,x._3+y._3)
)
println(result)
}
}
//(10,3,4)
