RDD弹性分布式数据集的基本操作

RDD的中文解释是弹性分布式数据集。
构造的数据集的时候用的是List(链表）或者Array数组类型
/* 使用makeRDD创建RDD */
/* List */
val rdd01 = sc.makeRDD(List(1,2,3,4,5,6))
val r01 = rdd01.map { x => x * x }
println(r01.collect().mkString(","))
/* Array */
val rdd02 = sc.makeRDD(Array(1,2,3,4,5,6))
val r02 = rdd02.filter { x => x < 5}
println(r02.collect().mkString(","))
 
val rdd03 = sc.parallelize(List(1,2,3,4,5,6), 1)
val r03 = rdd03.map { x => x + 1 }
println(r03.collect().mkString(","))
/* Array */
val rdd04 = sc.parallelize(List(1,2,3,4,5,6), 1)
val r04 = rdd04.filter { x => x > 3 }
println(r04.collect().mkString(","))

1 val rdd:RDD[String] = sc.textFile("file:///D:/sparkdata.txt", 1)
2 val r:RDD[String] = rdd.flatMap { x => x.split(",") }
3 println(r.collect().mkString(","))

 1 val rddInt:RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6,2,5,1))
 2 val rddStr:RDD[String] = sc.parallelize(Array("a","b","c","d","b","a"), 1)
 3 val rddFile:RDD[String] = sc.textFile(path, 1)
 4  
 5 val rdd01:RDD[Int] = sc.makeRDD(List(1,3,5,3))
 6 val rdd02:RDD[Int] = sc.makeRDD(List(2,4,5,1))
 7  
 8 /* map操作 */参数是函数，函数应用于RDD每一个元素，返回值是新的RDD
 9 println("======map操作======")
10 println(rddInt.map(x => x + 1).collect().mkString(","))
11 println("======map操作======")
12 /* filter操作 */参数是函数，函数会过滤掉不符合条件的元素，返回值是新的RDD
13 println("======filter操作======")
14 println(rddInt.filter(x => x > 4).collect().mkString(","))
15 println("======filter操作======")
16 /* flatMap操作 */参数是函数，函数应用于RDD每一个元素，将元素数据进行拆分，变成迭代器，返回值是新的RDD
17 println("======flatMap操作======")
18 println(rddFile.flatMap { x => x.split(",") }.first())
19 println("======flatMap操作======")
20 /* distinct去重操作 */没有参数，将RDD里的元素进行去重操作方法转换操作生成一个只包含不同元素的一个新的RDD。开销很大。 
21 println("======distinct去重======")
22 println(rddInt.distinct().collect().mkString(","))
23 println(rddStr.distinct().collect().mkString(","))
24 println("======distinct去重======")
25 /* union操作 */会返回一个包含两个RDD中所有元素的RDD，包含重复数据。
26 println("======union操作======")
27 println(rdd01.union(rdd02).collect().mkString(","))
28 println("======union操作======")
29 /* intersection操作 */只返回两个RDD中都有的元素。可能会去掉所有的重复元素。通过网络混洗来发现共有元素
30 println("======intersection操作======")
31 println(rdd01.intersection(rdd02).collect().mkString(","))
32 println("======intersection操作======")
33 /* subtract操作 */返回只存在第一个RDD中而不存在第二个RDD中的所有的元素组成的RDD。也需要网络混洗
34 println("======subtract操作======")
35 println(rdd01.subtract(rdd02).collect().mkString(","))
36 println("======subtract操作======")
37 /* cartesian操作 */计算两个RDD的笛卡尔积，转化操作会返回所有可能的(a,b)对，其中a是源RDD中的元素，而b则来自于另一个RDD。 
38 println("======cartesian操作======")
39 println(rdd01.cartesian(rdd02).collect().mkString(","))
40 println("======cartesian操作======")

 1 val rddInt:RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6,2,5,1))
 2 val rddStr:RDD[String] = sc.parallelize(Array("a","b","c","d","b","a"), 1)
 3  
 4 /* count操作 */返回RDD所有元素的个数
 5 println("======count操作======")
 6 println(rddInt.count())
 7 println("======count操作======")  
 8 /* countByValue操作 */各元素在RDD中出现次数
 9 println("======countByValue操作======")
10 println(rddInt.countByValue())
11 println("======countByValue操作======")
12 /* reduce操作 */并行整合所有RDD数据，例如求和操作
13 println("======reduce操作======")
14 println(rddInt.reduce((x ,y) => x + y))
15 println("======reduce操作======")
16 /* fold操作 */和reduce功能一样，不过fold带有初始值
17 println("======fold操作======")
18 println(rddInt.fold(0)((x ,y) => x + y))
19 println("======fold操作======")
20 /* aggregate操作 */和reduce功能一样，不过fold带有初始值
21 println("======aggregate操作======")
22 val res:(Int,Int) = rddInt.aggregate((0,0))((x,y) => (x._1 + x._2,y),(x,y) => (x._1 + x._2,y._1 + y._2))
23 println(res._1 + "," + res._2)
24 println("======aggregate操作======")
25 /* foeach操作 */对RDD每个元素都是使用特定函数就是遍历
26 println("======foeach操作======")
27 println(rddStr.foreach { x => println(x) })
28 println("======foeach操作======")

.mapValues(x=>(x,1)).//mapValues是对值的操作,不操作key使数据变成(Tom,（26,1）)