两种方式
sc.textFile() 方法来加载文件数据，并将文件数据转换为RDD
val rdd1 = sc.textFile("file:///home/hzp/Documents/input.txt")
ps: 1.如果文件只存在master节点上，那么可能会报 fileNotFoundException，需要把数据文件传给 worker 节点
https://blog.csdn.net/hzp666/article/details/117112227
2.因为spark-shell 的惰性机制，所以即使这里文件路径写错了，也不会立马报错，而是等遇到action 算子才会去真正执行，才会报错。
另外，当输入路径写的是 文件夹/* 时，系统会把当前文件夹中所有文件都加载进来
eg: sc.textFile("file:///D:/doc/spark/input/dir1/*")
3.当将计算结果保存到本地文件时，
使用saveAsTextFile() 来保存
avgRDD.saveAsTextFile("D:\\doc\\spark\\out\\result1")或
avgRDD.saveAsTextFile("D:/doc/spark/out/result1")
保存效果：
ps: 1. saveAsTextFile() 里边写的是 文件保存的路径，不是指定文件名，即使说写的路径为 avgRDD.saveAsTextFile("D:/doc/spark/out/result1.txt") 指定文件名，那么也是创建一个 名为result1.txt的文件夹。
2.文件夹内有2种文件success文件和part-00000文件， success代表本次代码执行成功；
part-00000是数据文件，这里只有一个文件因为程序没设置分区数，数据文件数和分区数对应。
3.当读取本地文件时，只用写到文件夹名称，系统会自动把文件夹内的所有part 文件都读取
val rdd2 = sc.textFile("hdfs:/user/root/source.txt")
scala> rdd2.flatMap(_.split(" ")).filter(s => s.contains("hadoop")).saveAsTextFile("hadoopWord.txt")
scala>
.saveAsTextFile("writeback")
会保存到 hdfs文件中，默认保存在 /usr/对应用户下
比如登录用户名为hadoop，那么将保存在/user/hadoop/writeback下
hdfs 查看：http://master:50070/explorer.html#/
https://blog.csdn.net/hzp666/article/details/117752252
parallelize 方法eg:
//create spark conf
val conf = new SparkConf().setAppName("sparkTemp").setMaster("local")
//create sc
val sc = new SparkContext(conf)
//create a list for make RDD
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三")
//create rdd
val rdd1 = sc.parallelize(list1)
//rdd1.foreach(println(_))
//flatmap
val flatMapedRDD2 = rdd1.flatMap(s=>s.split(" "))
//flatMapedRDD2.foreach(println(_))
val filteredRDD3 = flatMapedRDD2.filter(s => s.contains("z"))
val filteredRDD4 = flatMapedRDD2.filter(s => s.startsWith("z"))
filteredRDD3.foreach(println(_))
filteredRDD4.foreach(println("444",_))输出：
对RDD的操作总的分为2种，transformation 和 action
两个的区别，
transformation 只是记录要对rdd进行的操作动作，而不会真正的执行，只有遇到action算子 才会开始真正的从头到尾的计算。
常见的算子
过滤RDD中的元素
filter是一个高阶函数，参数中传递函数
eg1 过滤取到z 开头的元素:
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三")
//create rdd
val rdd1 = sc.parallelize(list1)
//filter
val filteredRDD = rdd1.filter(s => s.startsWith("z"))
//输出
filteredRDD.foreach(println(_))输出：
eg2 取出包含w 字符的元素:
//create a list for make RDD
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三")
//create rdd
val rdd1 = sc.parallelize(list1)
//rdd1.foreach(println(_))
val filteredRDD2 = rdd1.filter(s=>s.contains("w"))
filteredRDD2.foreach(println(_))
输出：
对RDD中每个元素进行操作
eg1:
val arr1 = Array(1,2,3,4,5)
val rdd2 = sc.parallelize(arr1)
val mapedRDD2 = rdd2.map(s => s+10)
mapedRDD2.foreach(println(_))输出：
eg2:
//create a list for make RDD
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三")
//create rdd
val rdd1 = sc.parallelize(list1)//map
val mapedRDD2 = rdd1.map(s => s+"-add")
mapedRDD2.foreach(println(_))输出：
eg3:
切分压平
eg1
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三")
val rdd1 = sc.parallelize(list1)
//flatmap
val flatMapedRDD2 = rdd1.flatMap(s=>s.split(" "))
flatMapedRDD2.foreach(println(_))
输出：
eg2：
应用于(key, value)的数据集，返回（key, Interable）形式的数据集
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三","mz 马志","ww 王武")
val flatMappedRDD = rdd1.flatMap(_.split(" "))
val mapedRDD = flatMappedRDD.map(s => (s,1))
val rdd3 = mapedRDD.groupByKey()
rdd3.foreach(println(_))
输出：
应用于(key, value)的数据集，返回（key, value），其中 返回的value 是根据传入 reduceByKey的函数方法，计算得到的
ps:1.相当于是把 groupByKey返回的 （key,Interable）中的 Interable 多了一步操作
2，reduceByKey中传入的方法 其实是对 Interable中的元素的操作，eg: mapedRDD.reduceByKey((a,b)=> a+b) 其实是对 每个key中的Interable中的元素 组内求和
reduceByKey 就相当于 groupByKey().map(t => (t._1, t._2.sum))
eg
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三","mz 马志","ww 王武")
val flatMappedRDD = rdd1.flatMap(_.split(" "))
val mapedRDD = flatMappedRDD.map(s => (s,1))
val rdd3 = mapedRDD.reduceByKey((a,b)=> a+b)rdd3.foreach(println(_))
输出：
eg2：
动作类型算子Action和Transformation转换算子的区别，就是基于惰性机制， 转换并不会真的执行，Action动作算子相当于点火装置，引燃整个计算。
计算当前RDD的元素个数
val arr1 = Array(1,2,3,4,5)
val rdd2 = sc.parallelize(arr1)
//filter elements >3
val more2RDD = rdd2.filter(_ > 3)
println(more2RDD.count())输出：
RDD把分布在各个节点的数据都回收回来，返回array 数组
eg:
val arr1 = Array(1,2,3,4,5)
println(rdd2.collect().mkString("Array(", ", ", ")"))输出：
取到RDD中第一个元素
eg:
val arr1 = Array(1,2,3,4,5)
//create rdd
val rdd2 = sc.parallelize(arr1)val rdd2 = sc.parallelize(arr1)
println(rdd2.first())输出：
获取RDD的前几个元素
eg;
val arr1 = Array(1,2,3,4,5)
//create rdd
val rdd2 = sc.parallelize(arr1)
println(rdd2.top(3).mkString("Array(", ", ", ")"))输出：
高阶函数， 传入一个函数，来聚合RDD中的元素
eg:
val arr1 = Array(1,2,3,4,5)
//create rdd
val rdd2 = sc.parallelize(arr1)val result = rdd2.reduce((a,b)=> a+b)
println(result)输出：
将RDD的计算结果保存到内存中去
有两种方法
rdd2.persist(MEMORY_ONLY) //仅保存在内存，当内存不够时，之前的老数据会被新数据替换掉
rdd2.persist(MEMORY_AND_DISK) //优先保存在内存中，内存不够时，写到磁盘
rdd2.cache() //等同于 rdd2.persist(MEMORY_ONLY)
取消内存持久化：
unpersist() 方法可以取消内存中的 数据持久化
RDD中的数据被分区存储在多台机器上。
主要有两点，增加并行计算能力 和 减少网络传输通信开销
RDD中的数据被分布在多台机器，多台机器可以并行计算，增加计算效率。
首先，先讲下数据分区 和 分块的区别：
分块：只是把数据按照一定的大小进行打包，不考虑数据的分类，只是分为一定数据大小的block 块，存储在不同机器上。
分区：是把数据按照一定的规则，进行分类存储在不同的机器上。
那么分区是怎么减少通信开销的，
如果数据不进行分区，那么在做数据关联join时 会出现一个情况，就是要把散布在各个机器上的 同类型数据进行 归类聚合，然后再去关联。
比如，有两张表，一个学生表，一个考场表
| 学生表 | |
| student_name | class_name |
| 张三 | 一班 |
| 李四 | 一班 |
| 王五 | 二班 |
| 麻子 | 三班 |
| 马六 | 一班 |
| 考场表 | |
| class_name | exam_name |
| 一班 | A考场 |
| 二班 | B考场 |
| 三班 | C考场 |
那如果想要把学生表和考场表关联，得到每个学生的考场
| student_name | class_name | exam_name |
| 张三 | 一班 | A考场 |
| 李四 | 一班 | A考场 |
| 王五 | 二班 | B考场 |
| 麻子 | 三班 | C考场 |
那么如果数据没有分区存储，那就需要把散落在各个机器上的一班学生数据 都拉到机器A上， 二班的学生数据都拉到机器B上。。。。，就会产生大量的网络传输通信开销。
如果是数据分区存储的，比如按照班级分区， 一班的学生数据都存在A机器上， 二班的学生数据都存在B机器。。。那么在做join关联时 就可以直接把一班的数据放到A机器上关联计算，二班数据都在B机器上计算， 不需要数据的传输。
尽量让分区的数量，等于CPU的数量，来实现并行计算。
可以通过设置spark.dafault.parallelise 来设置分区数，
不同模式下的默认分区数
1.local模式 默认是机器最大CPU数，如果设置了启动时候 设置了 local[N] 那么会启动N个 分区数
2.Mesos默认分区数是 8
3.standlone 和 yarn ：会把集群中总CPU数 和 2进行比较，取最大值，来作为分区数。
eg:
val rddT = sc.textFile("file:///D:/doc/spark/input/dir1/input.txt",2)
rddT.foreach(println(_))或
val list1 = List("mz 麻子","ww 王五","ls 李四","zs 张三","mz 马志","ww 王武")
//create rdd
val rdd1 = sc.parallelize(list1,2)
界面会有 分区提示
rddT.partitions.length4.2 重分区
repartition 方法
eg:
val rddT = sc.textFile("file:///D:/doc/spark/input/dir1/input.txt",3)
rddT.foreach(println(_))
println("before",rddT.partitions.length)
val rddt2 = rddT.repartition(1) //注意 重分区后，返回值是一个新RDD
rddt2.foreach(println("repartition: --",_))
println("repartition 's num",rddt2.partitions.length)输出：
spark有自带的分区 哈希分区、区域分区等
自定义分区需要做的几步
getPartition() 这个方法是 真正计算 数据存储在哪个分区的地方， 计算后返回对应数据 应存储的分区编号
eg:
class Mypartitioner(numParts:Int) extends Partitioner {
override def numPartitions = numParts
override def getPartition(key: Any): Int = {
//judge the number is ODD or EVEN
return key.toString.toInt % 2
}
}
//to create a rdd
val list = 1 to 10
val rdd = sc.parallelize(list,2)
// map((_,1)) because partitionBy only support (key,value) sytle
//map(_._1) because to remove the added value
rdd.map((_,1)).partitionBy(new Mypartitioner(2)).map(_._1).saveAsTextFile("D:/doc/spark/out/t3")
ps：1. 写完MyPartition 方法 extents Partition后，在MyPartition 上alt + enter， 选择override 选择全部重写即可
2. 在编写 getPartition 返回分区编号时，注意分区编号默认是从0 开始的
输出：
两个分区
第一个分区
第二个
自定义分区 eg2:
根据首字母来区分，分区
class MyPartition(partitionNum:Int) extends Partitioner{
override def numPartitions: Int = partitionNum
override def getPartition(key: Any): Int = {
val firstStr = key.toString.substring(0, 1)
// if the word startwith h then save to partition 0, else save to partition 1
firstStr match {
case "h" =>{
return 0
}
case _ =>{
return 1
}
}
}
}
val rddT = sc.textFile("D:/doc/spark/input/dir1/input.txt",2)
val rddfix = rddT.flatMap(s => s.split(" ")).map((_,1)).partitionBy(new MyPartition(2)).map(_._1)
rddfix.saveAsTextFile("D:/doc/spark/out/t5")输出：
第一个分区：
第二个分区：
RDD写个wordCount
//读取数据生成RDD
val sourceText = sc.textFile("D:/doc/spark/input/dir1/input.txt")
//flatMap
val flatMapedRDD = sourceText.flatMap(_.split(" "))
//转换为 键值对， 组内字频求和 ， 保存
flatMapedRDD.map((_,1)).reduceByKey((a,b) => a+b).saveAsTextFile("D:/doc/spark/out/t6")
把普通RDD通过map((_,1))转换
reduceByKey 和 groupByKey之前已经讲过了
是把键值对RDD中的所有key都放到一个list中返回
是返回值的list
按照key进行排序，默认参数是true升序， sortByKey(false) 降序
如果想根据 value值进行排序，eg： (a,10) (b, 20) (c,5)
想要根据 value进行排序 得到 （b, 20） (a, 10) (c, 5)
sortBy(_._2,false) 这种写法可以实现
eg:
val sourceText = sc.textFile("D:/doc/spark/input/dir1/input.txt")
val flatMapedRDD = sourceText.flatMap(_.split(" "))
val reduceedRDD = flatMapedRDD.map((_,1)).reduceByKey((a,b) => a+b)
//sortByKey
val sortedRDD = reduceedRDD.sortByKey()
sortedRDD.foreach(println(_))
//sortBy
val sortedByRDD = reduceedRDD.sortBy(_._2, false)
sortedByRDD.foreach(println("sortBy-----",_))输出：
对键值对RDD中的value进行 操作，不影响key，
mapValues(_+1) 等同于.map(item => (item._1, item._2 +1 ))
eg:
val sourceText = sc.textFile("D:/doc/spark/input/dir1/input.txt", 1)
val mapedRDD = sourceText.flatMap(_.split(" ")).map((_, 1))
val mapValueRDD = mapedRDD.mapValues(s=> s+1)
mapValueRDD.foreach(println(_))
val mapedRDD2 = mapedRDD.map(item => (item._1,item._2+1))
mapedRDD2.foreach(println("maped2----",_))输出：
eg:
val sourceText = sc.textFile("D:/doc/spark/input/dir1/input.txt", 1)
val mapedRDD = sourceText.flatMap(_.split(" ")).map((_, 1))
val mapValueRDD = mapedRDD.mapValues(s=> s+1)
mapValueRDD.foreach(println(_))
//reverse map
val reversedRDD = mapedRDD2.map(item => (item._2, item._1))
reversedRDD.foreach(println("reverse ----",_))输出：
把两个RDD 根据 key 进行关联，相同的key 对应的value 合并
eg:
val RDD1 = sc.parallelize(Array(("张三", 3), ("李四", 4), ("王五", 5), ("麻子六", 6), ("李四", "山西人")))
val RDD2 = sc.parallelize(Array(("张三", "广东人"), ("李四", "男")))
val RDD3 = RDD2.join(RDD1)
RDD3.foreach(println(_))输出:
有一堆数据 ("sparkBOOK", 5), ("hadoopBOOK", 4), ("hadoopBOOK", 2), ("hudiBOOK", 3), ("sparkBOOK", 8)，
("sparkBOOK", 5) 代表一次卖出5本书，
要计算出 每个技术栈数据平均每次卖几本书
eg:
val sourceRDD = sc.parallelize(Array(("spark", 5), ("hadoop", 4), ("hadoop", 2), ("hudi", 3), ("spark", 8)))
//to create rdd like this ("spark",(5,1)) ("hadoop", (4,1)) ...
// simple just for denominator
val addOneRDD = sourceRDD.map((item => (item._1, (item._2, 1))))
//to sum Numerator and denominator
val reducedRDD = addOneRDD.reduceByKey((a, b) => (a._1 + b._1, a._2 + b._2))
//numerator / denominator = avg
val avgRDD = reducedRDD.map(item => (item._1, item._2._1 / item._2._2))
avgRDD.foreach(println(_))输出：
