package com.yu.java.spark
import org.apache.hadoop.io.compress.GzipCodec
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Test_RDD1 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[2]").setAppName("wordcount")
val sc = new SparkContext(conf)
var rdd = sc.makeRDD(Array(("A", "1"), ("B", "2"), ("C", "3")), 2)
//println(rdd.first())
rdd.collect().foreach(println)
val rdd2 = rdd.map(lines => {
// val line=lines.split(",")
val filed1 = lines._1
val field2 = lines._2
(filed1, (filed1, field2))
})
rdd2.collect().foreach(println)
var rdd1 = sc.makeRDD(Seq(10, 3, 1, 13, 6))
println(rdd.first())
// 一.Action操作 常用算子
//count :返回rdd中的元素?cái)?shù)量
var rdd3 = sc.makeRDD(Array(("A", "1"), ("B", "2"), ("C", "3")), 2)
println("打印rdd的元素?cái)?shù)量：" + rdd.count())
//reduce(0 :根據(jù)需求對(duì)rdd里的元素進(jìn)行運(yùn)算奶栖，返回結(jié)果
var rdd4 = sc.makeRDD(Array(("A", 2), ("A", 5), ("B", 2), ("C", 3)))
var rdd5 = rdd4.reduce((x, y) => {
(x._1 + y._1, x._2 + y._2)
})
println(rdd5)
//輸出結(jié)果為1到10相加的和
var rdd6 = sc.makeRDD(1 to 10, 2)
rdd6.reduce(_ + _)
println("輸出結(jié)果為1到10相加的和" + rdd6.reduce(_ + _))
//collect()將一個(gè)rdd轉(zhuǎn)換為數(shù)組
var rdd7 = sc.makeRDD(1 to 3, 2)
// println("結(jié)果為："+rdd7.collect().foreach(println) )
rdd7.collect().foreach(println)
//take(n):獲得1到n之間的元素（不排序）
var rdd8 = sc.makeRDD(Seq(10, 3, 1, 13, 6))
rdd8.take(3).foreach(println) //取出前三個(gè)元素畸裳，然后將這三個(gè)元素打印出來(lái)
//top(n):默認(rèn)降序排序，然后返回n個(gè)元素 takOrdered(n):按照與top相反返回n個(gè)元素 先升序排序所有元素，然后取出n個(gè)元素
var rdd9 = sc.makeRDD(Seq(10, 3, 1, 13, 6))
println("top(n):默認(rèn)降序返回3個(gè)元素：")
rdd9.top(3).foreach(println)
//返回結(jié)果為Array(13,10,6)
//
//
println("按照與top相反返回3個(gè)元素：")
rdd9.takeOrdered(3).foreach(println)
println("按照與top相反返回3個(gè)元素：")
for (elem <- rdd9.takeOrdered(3)) {
println(elem)
}
//lookup用于(k,v)類(lèi)型的rdd倔既，制定k值袋励，返回rdd中該k對(duì)應(yīng)的所有v值
var rdd10 = sc.makeRDD(Array(("A", "2"), ("A", "5"), ("B", "2"), ("C", "3")))
println("將rdd10中的所有key為A做對(duì)應(yīng)的所有value值打印輸出：")
rdd10.lookup("A").foreach(println)
println("將rdd10中的所有key為B做對(duì)應(yīng)的所有value值打印輸出：")
for (elem <- rdd10.lookup("B")) {
println(elem)
}
println("將rdd10中的所有key為C所對(duì)應(yīng)的所有value值打印輸出：")
rdd10.lookup("C").foreach(println)
//countByKey:統(tǒng)計(jì)RDD[K,V]中每個(gè)K的數(shù)量
var rdd11 = sc.makeRDD(Array(("A", "1"), ("A", "3213"), ("A", "112"), ("A", "43"), ("A", "43"), ("B", "2"), ("B", "32"), ("C", "3")))
rdd11.countByKey().foreach(println)
//countByValue:統(tǒng)計(jì)RDD[K,V]中每個(gè)v的數(shù)量 如果v重復(fù)出現(xiàn)角虫，就會(huì)打印出重復(fù)出現(xiàn)的次數(shù)
rdd11.countByValue().foreach(println)
//soryBy:根據(jù)給定的排序k函數(shù)將rdd中的元素進(jìn)行排序
var rdd12 = sc.makeRDD(Seq(3, 6, 7, 1, 2, 0), 2)
rdd12.sortBy(x => x).collect().foreach(println) //默認(rèn)升序
rdd12.sortBy(x => x, false).collect().foreach(println) //降序
//saveAsTextFile:以text類(lèi)型保存到制定路徑
var rdd13 = sc.makeRDD(1 to 10, 2)
// rdd13.saveAsTextFile("hdfs://localhost:9000/test111/")//以text形式將文件保存到hdfs
// rdd13.saveAsTextFile("file:///tmp/test/")//將文件保存到本地
// rdd13.saveAsTextFile("hdfs://localhost:9000/test/",classOf[GzipCodec])////指定壓縮格式保存
// rdd13.saveAsObjectFile("hdfs://localhost:9000/test11132") //以saveAsOjbect形式將文件保存到hdfs中
//transformation 轉(zhuǎn)換算子：
//使用flatMap算子：
val data: RDD[String] = sc.textFile("d:\\a.txt") //讀取文件到rdd
val result1 = data.flatMap(word => word.split(","))
result1.collect().foreach(println)
val result333 = data.flatMap(word => word.split(",")).map(x => (x, 1))
result333.collect().foreach(println)
//使用map算子 //-------------------------------
val data123=sc.textFile("d:/a.txt")
println("map算子1:")
var result666= data123.map(line=>line.split(","))
var arr1= result666.take(5)
var arr2= result666.collect()
//打印出arr1這個(gè)數(shù)組里邊的所有元素
for(i<-0 until arr1.length){
println( arr1 (i).toList)
}
//打印出arr2這個(gè)數(shù)組里邊的所有元素
for(i<-0 until arr2.length){
println( arr2(i).toList)
}
//使用map算子1
val a = sc.parallelize(1 to 9, 3)
//定義函數(shù)
def mapDoubleFunc(a: Int): (Int, Int) = {
(a, a * 2)
}
val mapResult = a.map(mapDoubleFunc)
println(mapResult.collect().mkString)
//使用mapPartitions算子 //-------------------------------
def doubleFunc(iter: Iterator[Int]): Iterator[(Int, Int)] = {
var res = List[(Int, Int)]()
while (iter.hasNext) {
val cur = iter.next()
res.::=(cur, cur * 2)
}
res.iterator
}
val result555 = a.mapPartitions(doubleFunc)
println(result555.collect().mkString)
}
}
