大数据学习之Spark根本（增补）

10）mapValues

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{SparkConf, SparkContext}
4. object Demo10MapValues {
5.   def main(args: Array[String]): Unit = {
6.     val conf: SparkConf = new SparkConf()
7.       .setMaster("local")
8.       .setAppName("MapValues算子演示")
10.     val sc: SparkContext = new SparkContext(conf)
12.     val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
14.     val kvRDD1: RDD[(String, Int)] = linesRDD.map(_.split(",")).map {
15.       case Array(_, name:String, age:String, _,_) =>
16.         (name, age.toInt)
17.     }
19.     /**
20.      * mapValues函数也是作用在kv格式的算子上
21.      * 将每个元素的值传递给后面的函数，进行处理得到新的值，键不变，这个处理后的组合重新返回到新的RDD中
22.      * 不改变键，只改变每个键所对应的值（是每个人的年龄加100）
23.      */
24.     kvRDD1.mapValues(_ + 100).foreach(println)
26.   }
27. }

复制代码

11）sortBy

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{SparkConf, SparkContext}
3. object Demo12SortBy {
4.   def main(args: Array[String]): Unit = {
5.     val conf: SparkConf = new SparkConf().setMaster("local").setAppName("MapValues算子演示")
7.     val sc: SparkContext = new SparkContext(conf)
9.     //TODO parallelize：将scala的集合变成spark中的RDD
10.     val rdd1: RDD[Int] = sc.parallelize(List(34, 123, 6, 1, 231, 1, 34, 56, 2))
12.     //TODO 与Scala中不同，若是.sortBy(e)形式则会报错
13.     val rdd2: RDD[Int] = rdd1.sortBy((e: Int) => e)
14.     rdd2.foreach(println)
15.   }
16. }

复制代码

12）mapPartitions

mapPartitions：一次处理处罚一个分区中的数据
它与map的区别在于，map是每次处理处罚一条数据就返回一条数据到下一个rdd
而mapPartitions一次处理处罚一个分区的数据，处理处罚完再返回
末了的处理处罚效果和map的处理处罚效果是一样的
mapPartition可以优化与数据库毗连的次数

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{SparkConf, SparkContext}
4. object Demo11partitionBy {
5.   def main(args: Array[String]): Unit = {
6.     val conf: SparkConf = new SparkConf()
7.       .setMaster("local")
8.       .setAppName("MapValues算子演示")
10.     val sc: SparkContext = new SparkContext(conf)
11. // TODO spark/data/wcs/*目录下有两个文件，相当于两个block块，会生成两个分区
12.     val linesRDD: RDD[String] = sc.textFile("spark/data/wcs/*")
14.     /**
15.      * mapPartitions：一次处理一个分区中的数据
16.      * 它与map的区别在于，map是每次处理一条数据就返回一条数据到下一个rdd
17.      * 而mapPartitions一次处理一个分区的数据，处理完再返回()
18.      * 最后的处理效果和map的处理效果是一样的
19.      * mapPartition可以优化与数据库连接的次数，使用mapPartition时，可以将一个分区的数据视为一个批次，
20.      * 并在该批次内统一处理。这意味着可以在处理整个分区之前建立一次数据库连接，并在处理完整个分区后断开连接。
21.      */
22.     val rdd1: RDD[String] = linesRDD.mapPartitions((itr: Iterator[String]) => {
23.       // 每个分区打印一次
24.         println("=========================================")
25.       itr.map((e: String) => {
26.         e
27.       })
28.     })
30.     val rdd1: RDD[String] = linesRDD.map((itr: String) => {
31.          //  每条数据打印一次
32.       println("=========================================")
33.       itr
34.     })
36. //    linesRDD.map((e:String)=>{
37. //      //...数据库连接
38. //    })
40.     rdd1.foreach(println)
42.   }
43. }

复制代码

2、办法算子

1）foreach

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{SparkConf, SparkContext}
4. object Demo13foreach {
5.   def main(args: Array[String]): Unit = {
6.     val conf: SparkConf = new SparkConf()
7.       .setMaster("local")
8.       .setAppName("MapValues算子演示")
10.     val sc: SparkContext = new SparkContext(conf)
12.     val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
14.     val rdd1: RDD[Array[String]] = linesRDD.map((e: String) => {
15.       e.split(",")
16.     })
18.     val rdd2: RDD[(String, String, String, String, String)] = rdd1.map {
19.       case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
20.         (id, name, age, gender, clazz)
21.     }
23.     /**
24.      * 行动算子，就可以触发一次作业执行，有几次行动算子调用，就会触发几次
25.      *
26.      * rdd是懒加载的性质
27.      */
28.     //    rdd2.foreach(println)
29.     //    println("====================================")
30.     //    rdd2.foreach(println)
32.     println("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$") // 一定会打印，属于Scala而不属于spark作业中的语句
33.     val rdd3: RDD[(String, String, String, String, String)] = rdd2.map((t5: (String, String, String, String, String)) => {
34.       println("===============================")
35.       t5
36.     })
37.     println("#############################")
39.     rdd3.foreach(println)
41.     while (true) {
43.     }
45.   }
46. }

复制代码

2）collect

1. import org.apache.spark.{SparkConf, SparkContext}
2. import org.apache.spark.rdd.RDD
4. object Demo14collect {
5.   def main(args: Array[String]): Unit = {
6.     val conf: SparkConf = new SparkConf()
7.       .setMaster("local")
8.       .setAppName("MapValues算子演示")
10.     val sc: SparkContext = new SparkContext(conf)
12.     val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
14.     val rdd1: RDD[Array[String]] = linesRDD.map((e: String) => {
15.       e.split(",")
16.     })
18.     val rdd2: RDD[Student] = rdd1.map {
19.       case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
20.         Student(id.toInt, name, age.toInt, gender, clazz)
21.     }
23.     //collect将rdd转成合适的scala中的数据结构
24.     val stuArray: Array[Student] = rdd2.collect()
25.     //foreach是scala中的foreach，不会产生作业执行的
26.     stuArray.foreach(println)
27.     while (true){
29.     }
31.   }
32. }
34. case class Student(id: Int, name: String, age: Int, gender: String, clazz: String)

复制代码

3、算子案例

1. import org.apache.spark.{SparkConf, SparkContext}
2. import org.apache.spark.rdd.RDD
4. object Demo15StudentTest1 {
5.   def main(args: Array[String]): Unit = {
6.     //求年级总分前10的学生各科分数的详细信息
7.     val conf: SparkConf = new SparkConf()
8.       .setMaster("local")
9.       .setAppName("MapValues算子演示")
11.     val sc: SparkContext = new SparkContext(conf)
13.     /**
14.      * 文件中数据格式为：
15.      * 1500100001,1000001,98
16.      * 1500100001,1000002,5
17.      * 1500100001,1000003,137
18.      * 1500100001,1000004,29
19.      */
20.     val linesRDD: RDD[String] = sc.textFile("spark/data/score.txt")
22.     val idWithScoreRDD: RDD[(String, String, Int)] = linesRDD.map((line: String) => {
23.       line.split(",") match {
24.         case Array(id: String, subject_id: String, score: String) =>
25.           (id, subject_id, score.toInt)
26.       }
27.     })
29.     val array1: Array[String] = idWithScoreRDD
30.       .map((t3: (String, String, Int)) => (t3._1, t3._3))
31.       // 求出每个人的总分，为下面取总分前十名做铺垫
32.       .reduceByKey(_ + _)
33.       .sortBy((kv: (String, Int)) => -kv._2)
34.       .take(10)
35.       .map(_._1)
37.     idWithScoreRDD.filter((t3: (String, String, Int)) => {
38.       val bool: Boolean = array1.contains(t3._1)
39.       if(bool){
40.         println("存在")
41.       }
42.       bool
43.     }).foreach((t3: (String, String, Int)) => {
44.       println("==========================")
45.       println(t3)
46.     })
48.   }
49. }

复制代码

Spark中的缓存

checkpoint和cache的区别？
cache是将一个复杂的RDD做缓存，未来实验的时间，只是这个rdd会从缓存中取
checkpoint是永世将rdd数据恒久化，未来实验的时间，直接从查抄点的rdd以后实验
1、cache

实验过后缓存就都没了

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.storage.StorageLevel
3. import org.apache.spark.{SparkConf, SparkContext}
5. object Demo16Cache {
6.   def main(args: Array[String]): Unit = {
7.     val conf: SparkConf = new SparkConf()
8.       .setMaster("local")
9.       .setAppName("缓存演示")
10.     val sc: SparkContext = new SparkContext(conf)
11.     //===================================================================
13.     val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
14.     val studentsRDD: RDD[Student2] = linesRDD.map(_.split(","))
15.       .map {
16.         case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
17.           Student2(id, name, age.toInt, gender, clazz)
18.       }
20.     /**
21.      * 缓存：
22.      * 在缓存的第一次执行时，就已经从缓存中放入数据并且下面取的时候就从缓存中取数据了
23.      * 缓存的目的是为了spark core作业执行的时候，缩短rdd的执行链，能够更快的得到结果
24.      * RDD在缓存之后，若是再次执行这个RDD有向无环图时，即可从所缓存的RDD开始执行，缩短执行速度
25.      * 缓存的实现方式：
26.      *  1、需要缓存的rdd调用cache函数
27.      *  2、persist(StorageLevel.MEMORY_ONLY) 修改缓存级别
28.      */
29. //    studentsRDD.cache() // 默认将rdd缓存到内存中，缓存级别为memory_only
30.     // 将rdd缓存到磁盘中
31.     studentsRDD.persist(StorageLevel.MEMORY_AND_DISK)
33.     //需求1：求每个班级的人数
34.     val rdd1: RDD[(String, Iterable[Student2])] = studentsRDD.groupBy(_.clazz)
35.     val resRDD1: RDD[(String, Int)] = rdd1.map((kv: (String, Iterable[Student2])) =>
36.       (kv._1, kv._2.size))
37.     resRDD1.foreach(println)
39.     //需求2：求每个年龄的人数
40.     val rdd2: RDD[(Int, Iterable[Student2])] = studentsRDD.groupBy(_.age)
41.     val resRDD2: RDD[(Int, Int)] = rdd2.map((kv: (Int, Iterable[Student2])) =>
42.       (kv._1, kv._2.size))
43.     resRDD2.foreach(println)
45.     while (true){
47.     }
48.   }
49. }
51. case class Student2(id:String,name:String,age:Int,gender:String,clazz:String)

复制代码

缓存前

缓存后

2、checkpoint

永世将实验过程中RDD中运动的数据存储到磁盘（hdfs）中
checkpoint
须要设置checkpoint的路径，同一设置的
checkpoint也相称于一个办法算子，触发作业实验（触发一个新的Job作业的实验）
第二次DAG有向无环图实验的时间（再存储完之后的又一次调用办法算子后），直接从末了一个有查抄点的rdd开始向下实验

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{SparkConf, SparkContext}
4. object Demo17Checkpoint {
5.   def main(args: Array[String]): Unit = {
6.     val conf: SparkConf = new SparkConf()
7.       .setMaster("local")
8.       .setAppName("缓存演示")
9.     val sc: SparkContext = new SparkContext(conf)
10.     //TODO 设置检查点的存储路径
11.     sc.setCheckpointDir("spark/data/checkpoint1")
12.     //===================================================================
14.     val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
15.     val studentsRDD: RDD[Student2] = linesRDD.map(_.split(","))
16.       .map {
17.         case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
18.           Student2(id, name, age.toInt, gender, clazz)
19.       }
20.       
21.     // TODO checkpoint
22.     studentsRDD.checkpoint()
24.     //需求1：求每个班级的人数
25.     val rdd1: RDD[(String, Iterable[Student2])] = studentsRDD.groupBy(_.clazz)
26.     val resRDD1: RDD[(String, Int)] = rdd1.map((kv: (String, Iterable[Student2])) => (kv._1, kv._2.size))
27.     resRDD1.foreach(println)
29.     //需求2：求每个年龄的人数
30.     val rdd2: RDD[(Int, Iterable[Student2])] = studentsRDD.groupBy(_.age)
31.     val resRDD2: RDD[(Int, Int)] = rdd2.map((kv: (Int, Iterable[Student2])) => (kv._1, kv._2.size))
32.     resRDD2.foreach(println)
34.     while (true) {
36.     }
37.   }
38. }

复制代码

运行界面信息

Spark在linux中的摆设

设置

1、上传解压，设置环境变量 设置bin目次
解压
tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C …/
重定名
mv spark-3.1.3-bin-hadoop3.2 spark-3.1.3
设置环境变量
vim /etc/profile
更改用户组：
chown -R root:root spark-3.1.3
2、修改设置文件 conf
cp spark-env.sh.template spark-env.sh
增长设置：
export SPARK_MASTER_IP=master
export SPARK_MASTER_PORT=7077
export SPARK_WORKER_CORES=2
export SPARK_WORKER_INSTANCES=1
export SPARK_WORKER_MEMORY=2g
export JAVA_HOME=/usr/local/soft/jdk1.8.0_171
master相称于RM  worker相称于NM
增长从节点设置
cp workers.template workers
vim workers
更改localhost为：
node1
node2
3、复制到别的节点
scp -r spark-3.1.3 node1:pwd
scp -r spark-3.1.3 node2:pwd
4、在主节点实验启动下令
启动集群，在master中实验
./sbin/start-all.sh
http://master:8080/  访问spark ui
standalone client模式

1、standalone client模式   日记在本地输出，一样寻常用于上线前测试(bin/下实验)
须要进入到spark-examples_2.11-2.4.5.jar 包所在的目次下实验
cd /usr/local/soft/spark-2.4.5/examples/jars
提交spark任务
spark-submit --class org.apache.spark.examples.SparkPi --master spark://master:7077 --executor-memory 512m --total-executor-cores 1 ./spark-examples_2.12-3.1.3.jar 10
2、standalone cluster模式   上线利用，不会在本地打印日记
在该模式下运行时，必须包管node1、node2的/usr/local/soft/spark-3.1.3/examples/jars下由所要运行的jar包。由于不确定是哪个节点在运行任务。

1. scp spark-1.0.jar node1:/usr/local/soft/spark-3.1.3/examples/jars

复制代码

实验模板：
spark-submit --class org.apache.spark.examples.SparkPi --master spark://master:7077 --executor-memory 512M --total-executor-cores 1 --deploy-mode cluster ./spark-examples_2.12-3.1.3.jar 100
spark-shell ： spark 提供的一个交互式的下令行，可以直接写代码
spark-shell master spark://master:7077
整合yarn

yarn中之前没有设置yarn.application.classpath，须要对其举行设置：
样例：
在Linux中查找value：

1. hadoop classpath

复制代码

1. <property>
2. <name>yarn.application.classpath</name>
3. <value>/usr/local/soft/hadoop-3.1.3/etc/hadoop:/usr/local/soft/hadoop-3.1.3/share/hadoop/common/    lib/*:/usr/local/soft/hadoop-3.1.3/share/hadoop/common/*:/usr/local/soft/hadoop-3.1.3/share/hadoop/hd    fs:/usr/local/soft/hadoop-3.1.3/share/hadoop/hdfs/lib/*:/usr/local/soft/hadoop-3.1.3/share/hadoop/hdf    s/*:/usr/local/soft/hadoop-3.1.3/share/hadoop/mapreduce/lib/*:/usr/local/soft/hadoop-3.1.3/share/hado    op/mapreduce/*:/usr/local/soft/hadoop-3.1.3/share/hadoop/yarn:/usr/local/soft/hadoop-3.1.3/share/hado    op/yarn/lib/*:/usr/local/soft/hadoop-3.1.3/share/hadoop/yarn/*</value>
4. </property>

复制代码

提交到了yarn上，可以在master:9870网页查察
在公司一样寻常不实用standalone模式，由于公司一样寻常已经有yarn  不须要搞两个资源管理框架
制止spark集群
在spark sbin目次下实验  ./stop-all.sh
spark整合yarn只须要在一个节点整合, 可以删除node1 和node2中全部的spark 文件
1、增长hadoop 设置文件所在

1. conf目录下
2. vim spark-env.sh
3. 增加
4. export HADOOP_CONF_DIR=/usr/local/soft/hadoop-3.1.3/etc/hadoop

复制代码

2、往yarn提交任务须要增长两个设置  yarn-site.xml(/usr/local/soft/hadoop-3.1.3/etc/hadoop/yarn-site.xml)（Hadoop设置时已设置）
先关闭yarn
stop-all.sh
cd /usr/local/soft/hadoop-3.1.3/etc/hadoop
vim yarn-site.xml
增长设置

yarn.nodemanager.pmem-check-enabled
false

yarn.nodemanager.vmem-check-enabledfalse4、同步到其他节点，重启yarn（Hadoop设置时已设置）
scp -r yarn-site.xml node1:pwd
scp -r yarn-site.xml node2:pwd
启动yarn
start-all.sh
cd /usr/local/soft/spark-3.1.3/examples/jars
3.spark        on yarn client模式   日记在本地输出，一班用于上线前测试
spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode client spark-examples_2.12-3.1.3.jar 100
4.spark on yarn cluster模式   上线利用，不会再本地打印日记   镌汰io
spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode cluster spark-examples_2.12-3.1.3.jar 100
获取yarn步伐实验日记  实验乐成之后才气获取到
yarn logs -applicationId application_1560967444524_0003
hdfs webui
http://node1:50070
yarn ui
http://node1:8088