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中的元素數量
var rdd3 = sc.makeRDD(Array(("A", "1"), ("B", "2"), ("C", "3")), 2)
println("打印rdd的元素数量：" + rdd.count())
//reduce(0 :根据需求对rdd里的元素进行运算，返回结果
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)
//输出结果为1到10相加的和
var rdd6 = sc.makeRDD(1 to 10, 2)
rdd6.reduce(_ + _)
println("输出结果为1到10相加的和" + rdd6.reduce(_ + _))
//collect()将一个rdd转换为数组
var rdd7 = sc.makeRDD(1 to 3, 2)
// println("结果为："+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) //取出前三个元素，然后将这三个元素打印出来
//top(n):默认降序排序，然后返回n个元素 takOrdered(n):按照与top相反返回n个元素 先升序排序所有元素，然后取出n个元素
var rdd9 = sc.makeRDD(Seq(10, 3, 1, 13, 6))
println("top(n):默认降序返回3个元素：")
rdd9.top(3).foreach(println)
//返回结果为Array(13,10,6)
//
//
println("按照与top相反返回3个元素：")
rdd9.takeOrdered(3).foreach(println)
println("按照与top相反返回3个元素：")
for (elem <- rdd9.takeOrdered(3)) {
println(elem)
}
//lookup用于(k,v)类型的rdd，制定k值，返回rdd中该k对应的所有v值
var rdd10 = sc.makeRDD(Array(("A", "2"), ("A", "5"), ("B", "2"), ("C", "3")))
println("将rdd10中的所有key为A做对应的所有value值打印输出：")
rdd10.lookup("A").foreach(println)
println("将rdd10中的所有key为B做对应的所有value值打印输出：")
for (elem <- rdd10.lookup("B")) {
println(elem)
}
println("将rdd10中的所有key为C所对应的所有value值打印输出：")
rdd10.lookup("C").foreach(println)
//countByKey:统计RDD[K,V]中每个K的数量
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:统计RDD[K,V]中每个v的数量 如果v重复出现，就会打印出重复出现的次数
rdd11.countByValue().foreach(println)
//soryBy:根据给定的排序k函数将rdd中的元素进行排序
var rdd12 = sc.makeRDD(Seq(3, 6, 7, 1, 2, 0), 2)
rdd12.sortBy(x => x).collect().foreach(println) //默认升序
rdd12.sortBy(x => x, false).collect().foreach(println) //降序
//saveAsTextFile:以text类型保存到制定路径
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 转换算子：
//使用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这个数组里边的所有元素
for(i<-0 until arr1.length){
println( arr1 (i).toList)
}
//打印出arr2这个数组里边的所有元素
for(i<-0 until arr2.length){
println( arr2(i).toList)
}
//使用map算子1
val a = sc.parallelize(1 to 9, 3)
//定义函数
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)
}
}
