大数据学习之Spark根本

Spark根本

简述

1、spark作业实行的特点：

   （1）只有遇到举措算子的时间，整个spark作业才会被触发实行
  （2）遇到频频，实行频频
  2、RDD: 弹性分布式数据集

   弹性：数据量小大由之
  RDD类似于容器，但是自己存储的不是数据，是盘算逻辑
  当遇到举措算子的时间，整个spark作业才会被触发实行，从第一个RDD开始实行，数据才开始产生存动
  数据在RDD之间只是运动关系，不会存储
  运动的数据量可以很大，也可以很小，以是称为弹性
  分布式：
  spark本质上它是必要从HDFS中读取数据的，HDFS是分布式，数据block块未来大概会在差别的datanode上
  RDD中运动的数据，大概会来自差别的datanode中的block块数据
  数据集：
  盘算运动过程中，可以短暂地将RDD当作一个容器，容器中有数据，默认环境下在内存中不会举行存储
  背面会有办法将一个RDD的数据存储到磁盘中
  RDD的5大特性：（口试必问！）

1、RDD是由一系列分区构成

   1）读文件时的minPartitions参数只能决定最小分区数，实际读取文件后的RDD分区数，由数据内容自己以及集群的分布来共同决定的
  2）若设置minPartitions的数量比block块数量还少的话，实际上以block块数量来决定分区数
  3）产生shuffle的算子调用时，可以传入numPartitions（比方：groupby()），可以真正改变RDD的分区数，设置多少，终极RDD就有多少分区
  4）文件会以block块的情势存储在HDFS上，若文件未到达128M默认值的话也会被一个block块存储。
  一开始RDD中的分区数由读取数据的block块数量决定的。
  后一个RDD中的分区数据，除KV函数以外，对应的是前一个RDD中的分区数据所举行逻辑处理惩罚后的结果。默认环境下，若后续分区不做处理惩罚的话，后续全部的RDD的分区数取决于第一个RDD。
  终极RDD中有几个分区，未来在HDFS中就会看到几个结果文件（HDFS -> RDD -> HDFS）
  2、算子是作用在每一个分区上的（每一个分区都会处理惩罚）

3、RDD与RDD之间存在一些依赖关系

   1）窄依赖 前一个RDD中的某一个分区数据只会到后一个RDD中的某唯一分区中 一对一（也大概前多个分区到后一个分区中）的关系
  2）宽依赖 前一个RDD中的某一个分区数据会进入到后一个RDD中的差别分区中 一对多的关系 也可以通过查察是否产生shuffle来判断
  3）整个spark作业会被宽依赖的个数分别多少个stage, Num(stage) = Num(宽依赖) + 1
  4）当遇到产生shuffle的算子的时间，涉及到从前一个RDD写数据到磁盘中，从磁盘中读取数据到后一个RDD的征象，
  注意：第一次触发实行的时间，磁盘是没有数据的，以是会从第一个RDD产生开始实行
  当重复触发类似的实行的时间，对于同一个DAG有向无环图而言，会直接从shuffle之后的RDD开始实行（省略从前一个RDD写数据到磁盘中的过程），可以直接从磁盘读取数据。
  5）**在一个阶段中，RDD有几个分区，**就会有几个并行task使命
  4、kv算子只能作用在kv的RDD上

5、spark会提供最优的使命盘算方式，只移动盘算，不移动数据。

   Spark的筹划原则之一是数据当地化（Data Locality），即只管让盘算使命在数据地点的节点上实行，从而镌汰数据的网络传输开销。
  Spark实例：wordcount

1. object WordCount2 {
2.    
3.    
4.   def main(args: Array[String]): Unit = {
5.    
6.    
7.     //创建spark配置文件对象
8.     val conf: SparkConf = new SparkConf()
9.     //设置运行模式
10.     //如果是本地local模式运行的话，需要设置setMaster
11.     //将来如果是集群进行，将这句话注释即可
12.     conf.setMaster("local")
13.     //设置spark作业的名字
14.     conf.setAppName("wordcount")
16.     //创建spark core上下文环境对象
17.     val sc: SparkContext = new SparkContext(conf)
18.     //===================================================================================
20.     //读取文件,每次读取一行
21.     //RDD是spark core中的核心数据结构，将来运行的时候，数据会在RDD之间流动，默认基于内存计算
22.     val linesRDD: RDD[String] = sc.textFile("spark/data/wcs/*")
23.     //    println(s"linesRDD的分区数：${linesRDD.getNumPartitions}")
25.     //一行数据根据分隔符分割
26.     val wordRDD: RDD[String] = linesRDD.flatMap(_.split("\\|"))
27.     //    println(s"wordRDD的分区数：${wordRDD.getNumPartitions}")
29.     //将每一个单词组成(word,1)
30.     val kvRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
31.         println(s"kvRDD的分区数：${kvRDD.getNumPartitions}")
33.     //根据键进行分组，并设置分区数为 5
34.     val kvRDD2: RDD[(String, Iterable[(String, Int)])] = kvRDD.groupBy(_._1,numPartitions = 5)
35.         println(s"kvRDD2的分区数：${kvRDD2.getNumPartitions}")
37.     val resRDD: RDD[(String, Int)] = kvRDD2.map((e: (String, Iterable[(String, Int)])) => (e._1, e._2.size))
38.         println(s"resRDD的分区数：${resRDD.getNumPartitions}")
40.     //打印
41.     resRDD2.foreach(println)
43.     //指定的是所要写入数据的文件夹的路径
44.     //spark如果是local本地运行的话，会将本地文件系统看作一个hdfs文件系统
45.     resRDD.saveAsTextFile("spark/data/outdata1")
47.   }
48. }

复制代码

Spark中RDD调用的函数，称之为算子

   算子分为两类：
  1、转换算子（RDD -> RDD，处理惩罚逻辑）
  2、举措算子（触发作业的实行）
  1、转换算子

1）Map

1. import org.apache.spark.rdd.RDD
2. import org.apache.spark.{
3.    
4.    SparkConf, SparkContext}
6. object Demo1Map {
7.    
8.    
9.   def main(args: Array[String]): Unit = {
10.    
11.    
12.     val conf = new SparkConf()
13.     conf.setMaster("local")
14.     conf.setAppName("map算子演示")
15.     val sc: SparkContext = new SparkContext(conf)
17.     val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
18.     //map操作算子：将rdd中的数据依次取出，传递给后面函数逻辑，将计算后的数据返回到新的rdd中
19.     //将rdd中的数据依次取出，处理完的数据返回下一个rdd直接继续执行后续的逻辑
20.     val rdd2: RDD[(String,String,String,String,String)] = lineRDD.map((line: String) => {
21.    
22.    
23.       println("==============处理后的数据========================")
24.       val array1: Array[String] = line.split(",")
25.       (array1(0),array1(1),array1(2),array1(3),array1(4))
26.     })
28.     //foreach是一个行动算子，遇到行动算子，触发作业执行
29.     /**
30.      * 转换操作（转换算子中定义了操作逻辑）仅仅是定义了数据应该如何被转换，而不会立即执行。
31.      * 只有当需要计算一个结果时（即调用行动算子时），才会执行。
32.      * 打印结果：
33.      * ==============处理后的数据========================
34.      * (1500100001,施笑槐,22,女,文科六班)
35.      * ==============处理后的数据========================
36.      * (1500100002,吕金鹏,24,男,文科六班)
37.      *每次调用行动算子（foreach）打印一条数据，都会是整个RDD重新执行一次（所有RDD的执行关系是一个有向无环图）
38.      */
39.     rdd2.foreach(println)
41.   }
43. }

复制代码

2）filter

1. import org.apache.spark.{
2.    
3.    SparkConf, SparkContext}
4. import org.apache.spark.rdd.RDD

复制代码