Spark

1.1 什么是Spark

回顾：Hadoop主要解决，海量数据的存储和海量数据的分析计算。

Spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。

1.2 Hadoop与Spark历史

1.3 Hadoop与Spark框架对比

1.4 Spark内置模块

Spark Core：实现了Spark的基本功能，包含任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core中还包含了对弹性分布式数据集(Resilient Distributed DataSet，简称RDD)的API定义。

Spark SQL：是Spark用来操作结构化数据的程序包。通过Spark SQL，我们可以使用 SQL或者Apache Hive版本的HQL来查询数据。Spark SQL支持多种数据源，比如Hive表、Parquet以及JSON等。

Spark Streaming：是Spark提供的对实时数据进行流式计算的组件。提供了用来操作数据流的API，并且与Spark Core中的 RDD API高度对应。

Spark MLlib：提供常见的机器学习功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据 导入等额外的支持功能。

Spark GraphX：主要用于图形并行计算和图挖掘系统的组件。

集群管理器：Spark设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，Spark支持在各种集群管理器（Cluster Manager）上运行，包括Hadoop YARN、Apache Mesos，以及Spark自带的一个简易调度器，叫作独立调度器。

 Spark得到了众多大数据公司的支持，这些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、腾讯、京东、携程、优酷土豆。当前百度的Spark已应用于大搜索、直达号、百度大数据等业务；阿里利用GraphX构建了大规模的图计算和图挖掘系统，实现了很多生产系统的推荐算法；腾讯Spark集群达到8000台的规模，是当前已知的世界上最大的Spark集群。

1.5 Spark特点

1.6  Master和Worker集群资源管理

Master和Worker是Spark的守护进程、集群资源管理者，即Spark在特定模式(Standalone)下正常运行必须要有的后台常驻进程。

1.7  Driver和Executor任务的管理者

Driver和Executor是临时程序，当有具体任务提交到Spark集群才会开启的程序。

1.8 几种模式对比

1.9 端口号总结

1）Spark查看当前Spark-shell运行任务情况端口号：4040

2）Spark Master内部通信服务端口号：7077    （类比于yarn的8032(RM和NM的内部通信)端口）

3）Spark Standalone模式Master Web端口号：8080（类比于Hadoop YARN任务运行情况查看端口号：8088）

4）Spark历史服务器端口号：18080         （类比于Hadoop历史服务器端口号：19888）

部署Spark集群大体上分为两种模式：单机模式与集群模式

大多数分布式框架都支持单机模式，方便开发者调试框架的运行环境。但是在生产环境中，并不会使用单机模式。因此，后续直接按照集群模式部署Spark集群。

下面详细列举了Spark目前支持的部署模式。

（1）Local模式：在本地部署单个Spark服务

（2）Standalone模式：Spark自带的任务调度模式。（国内常用）

（3）YARN模式：Spark使用Hadoop的YARN组件进行资源与任务调度。（国内最常用）

（4）Mesos模式：Spark使用Mesos平台进行资源与任务的调度。（国内很少用）

2.1 Spark安装

（1）scala环境搭建

解压、改名

[root@kb129 install]# tar -xvf ./scala-2.12.10.tgz -C ../soft/

[root@kb129 soft]# mv ./scala-2.12.10/ scala212

配置环境变量

[root@kb129 soft]# vim /etc/profile

#SCALA_HOME
export SCALA_HOME=/opt/soft/scala212
export PATH=$SCALA_HOME/bin:$PATH

[root@kb129 soft]# source /etc/profile

（2）spark安装部署

解压、改名

[root@kb129 install]# tar -xvf ./spark-3.1.2-bin-hadoop3.2.tgz -C ../soft/

[root@kb129 soft]# mv ./spark-3.1.2-bin-hadoop3.2/ spark312

拷贝配置文件，编辑

[root@kb129 conf]# cp spark-env.sh.template spark-env.sh

[root@kb129 conf]# cp workers.template workers

配置环境变量

[root@kb129 conf]# vim /etc/profile

#SPARK_HOME

export SPARK_HOME=/opt/soft/spark312

export PATH=$SPARK_HOME/bin:$PATH

[root@kb129 conf]# source /etc/profile

编辑配置文件

[root@kb129 conf]# vim ./workers

[root@kb129 conf]# vim ./spark-env.sh

末尾追加

export SCALA_HOME=/opt/soft/scala212

export JAVA_HOME=/opt/soft/jdk180

export SPARK_HOME=/opt/soft/spark312

export HADOOP_HOME=/opt/soft/hadoop313

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

export SPARK_MASTER_IP=192.168.142.129

export SPARK_DRIVER_MEMORY=2G

export SPARK_EXECUTOR_MEMORY=2G

export SPARK_LOCAL_DIRS=/opt/soft/spark312

spark-shell启动

[root@kb129 conf]# spark-shell

1 什么是RDD

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象。

代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

2 RDD五大特性

data 类型为RDD（分布式数据集）

RDD算子

collect收集完装到数组中，数组函数如下

glom

repartition和coalesce的区别

两个都能调整分区数,但repartition的底层依然是调用了coalesce

coalesce的语法: coalesce(num,shuffle=False) 默认不启动shuffle

repartition的语法: repartition(num) 默认启动shuffle

repartition中将shuffle改成了ture,且参数不可修改

因此,repartition常用于增加分区,coalesce常用于减小分区

关键就在于shuffle是否启动

重新分区的根本是通过hash取模后再分区,因此必须通过shuffle

分区数据重新分区时会出现1个分区数据分配到其他多个分区的情况,也就形成了「宽依赖」

减小分区的根本是将1个分区完整归类到另一个分区中,属于1对1的情况,也就形成「窄依赖」

创建RDD的第二种方式mkRDD

Spark wordcount案例

sc.textFile("hdfs://kb129:9000/kb23/tmp/*.txt").flatMap(x=>x.split(" ")).map(x=>(x,1)).reduceByKey(_+_).glom.collect

3 WordCount工作流程及API

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
 
object WordCountDemo {
  def main(args: Array[String]): Unit = {
    //创建SparkConf并设置App名称
    val conf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
 
    //创建Spark context，该对象是提交Spark App的入口
    val sc = SparkContext.getOrCreate(conf)
 
    //创建Spark session
    val spark = SparkSession.builder().config(conf).getOrCreate()
 
    //val rdd = sc.textFile("hdfs://kb129:9000/kb23/tmp/*.txt")
    //val rdd2 = rdd.flatMap(x => x.split(" "))
    //val rdd3 = rdd2.map(x => (x, 1))
    //val result = rdd3.reduceByKey((x, y) => x + y)
 
    //等价于
    val result = sc.textFile("hdfs://kb129:9000/kb23/tmp/*.txt")
      .flatMap(x => x.split(" ")).map(x => (x, 1))
      .reduceByKey((x, y) => x + y)
 
    //控制台输出结果
    result.collect.foreach(println)
 
    //结果保存到hdfs中
    //result.saveAsTextFile("hdfs://kb129:9000/kb23/tmp/wordcount")
  }
}

4 RDD算子练习—求出“张”姓学生总分最高的学生姓名、总分

//1  sortBy方式 求"张"姓 总分最高
sc.parallelize(Array(("zhangsan", 100, 88, 80), ("lisi", 80, 66, 99),
  ("zhangxuefeng", 99, 8, 100), ("zhangwuji", 9, 80, 10)))
  .filter(x => x._1.startsWith("zhang"))
  .map(x => (x._1, x._2 + x._3 + x._4))
  .sortBy(x => - x._2)
  .take(1)
  .foreach(println) //(zhangsan,279)
 
 
//2  reduce方式 求"张"姓 总分最高
println(
  sc.parallelize(Array(("zhangsan", 100, 88, 80), ("lisi", 80, 66, 99),
  ("zhangxuefeng", 99, 8, 100), ("zhangwuji", 9, 80, 10)))
  .filter(x => x._1.startsWith("zhang"))
  .map(x => (x._1, x._2 + x._3 + x._4))
  .reduce((x, y) => {if (x._2 > y._2) x else y})
) //(zhangsan,279)
 
 
//3  reduceByKey方式 求"张"姓 总分最高
sc.parallelize(Array(("zhangsan", 100, 88, 80), ("lisi", 80, 66, 99),
  ("zhangxuefeng", 99, 8, 100), ("zhangwuji", 9, 80, 10)))
  .filter(x => x._1.startsWith("zhang"))
  .map(x => ("zhang",(x._1, x._2 + x._3 + x._4)))
  .reduceByKey((x, y) => {if (x._2 >y._2) x else y})
  .map(x => x._2)
  .collect().foreach(println)   //(zhangsan,279)
 
//4  max方式 求"张"姓 总分最高
println(
  sc.parallelize(Array(("zhangsan", 100, 88, 80), ("lisi", 80, 66, 99),
    ("zhangxuefeng", 99, 8, 100), ("zhangwuji", 9, 80, 10)))
  .filter(x => x._1.startsWith("zhang"))
  .map(x => (x._2 + x._3 + x._4, x._1))
  .max
  .swap
  ) //(zhangsan,279)
 
//5  groupBy方式 求"张"姓 总分最高
sc.parallelize(Array(("zhangsan", 100, 88, 80), ("lisi", 80, 66, 99),
  ("zhangxuefeng", 99, 8, 100), ("zhangwuji", 9, 80, 10)))
  .filter(x => x._1.startsWith("zhang"))
  .map(x => (x._1, x._2 + x._3 + x._4))
  .groupBy(x => x._1.substring(0, 5))
  .map(x => {
    var name = "";
    var sumscore = 0;
    val itor = x._2.iterator;
    for (e <- itor) {
      if (e._2 > sumscore) {
        name = e._1;
        sumscore = e._2
      }
    };
    (name, sumscore)
  }).collect().foreach(println)   //(zhangsan,279)

2.3 Transformation转换算子（面试开发重点）

RDD整体上分为Value类型、双Value类型和Key-Value类型

2.3.1 Value类型

map()映射

mapPartitions()以分区为单位执行Map

mapPartitionsWithIndex()带分区号

1）函数签名：

      def mapPartitionsWithIndex[U: ClassTag](

      f: (Int, Iterator[T]) => Iterator[U], // Int表示分区编号

      preservesPartitioning: Boolean = false): RDD[U]

2）功能说明：类似于mapPartitions，比mapPartitions多一个整数参数表示分区号

3）需求说明：创建一个RDD，使每个元素跟所在分区号形成一个元组，组成一个新的RDD

flatMap()扁平化

1）函数签名：def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

2）功能说明

与map操作类似，将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。

区别：在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。

3）需求说明：创建一个集合，集合里面存储的还是子集合，把所有子集合中数据取出放入到一个大的集合中。

glom()分区转换数组

1）函数签名：def glom(): RDD[Array[T]]

2）功能说明

    该操作将RDD中每一个分区变成一个数组，并放置在新的RDD中，数组中元素的类型与原分区中元素类型一致

3）需求说明：创建一个2个分区的RDD，并将每个分区的数据放到一个数组，求出每个分区的最大值

groupBy()分组

groupBy会存在shuffle过程

shuffle：将不同的分区数据进行打乱重组的过程

shuffle一定会落盘。可以在local模式下执行程序，通过4040看效果。

filter()过滤

1）函数签名： def filter(f: T => Boolean): RDD[T]

2）功能说明

接收一个返回值为布尔类型的函数作为参数。当某个RDD调用filter方法时，会对该RDD中每一个元素应用f函数，如果返回值类型为true，则该元素会被添加到新的RDD中。

sample()采样

1）函数签名：

       def sample(

            withReplacement: Boolean,

            fraction: Double,

            seed: Long = Utils.random.nextLong): RDD[T]

// withReplacement： true为有放回的抽样，false为无放回的抽样；

// fraction表示：以指定的随机种子随机抽样出数量为fraction的数据；

// seed表示：指定随机数生成器种子。

2）功能说明

从大量的数据中采样

3）需求说明：创建一个RDD（1-10），从中选择放回和不放回抽样

distinct()去重

distinct会存在shuffle过程

coalesce()合并分区

Coalesce算子包括：配置执行Shuffle和配置不执行Shuffle两种方式。

1、不执行Shuffle方式

1）函数签名：

def coalesce(numPartitions: Int, shuffle: Boolean = false,  //默认false不执行shuffle

        partitionCoalescer: Option[PartitionCoalescer] = Option.empty)

        (implicit ord: Ordering[T] = null) : RDD[T]

2）功能说明：缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。

3）需求：4个分区合并为2个分区

2、执行Shuffle方式

//3. 创建一个RDD
val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6), 3)
//3.1 执行shuffle
val coalesceRdd: RDD[Int] = rdd.coalesce(2, true)

输出结果：

(0,1)

(0,4)

(0,5)

(1,2)

(1,3)

(1,6)

3、Shuffle原理

coalesce和repartition区别

1）coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。

2）repartition实际上是调用的coalesce，进行shuffle。源码如下：

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
    coalesce(numPartitions, shuffle = true)
}

3）coalesce一般为缩减分区，如果扩大分区，不使用shuffle是没有意义的，repartition扩大分区执行shuffle。

repartition()重新分区（执行Shuffle）

1）函数签名： def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

2）功能说明

该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的RDD转换为分区数多的RDD，repartition操作都可以完成，因为无论如何都会经shuffle过程。

3）需求说明：创建一个4个分区的RDD，对其重新分区。

sortBy()排序

1）函数签名：

def sortBy[K]( f: (T) => K,

      ascending: Boolean = true, // 默认为正序排列

      numPartitions: Int = this.partitions.length)

      (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

2）功能说明

该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为正序排列。排序后新产生的RDD的分区数与原RDD的分区数一致。

3）需求说明：创建一个RDD，按照数字大小分别实现正序和倒序排序

pipe()调用脚本

1）函数签名： def pipe(command: String): RDD[String]

2）功能说明

管道，针对每个分区，都调用一次shell脚本，返回输出的RDD。

注意：在Worker节点可以访问到的位置脚本需要放

2.3.2 双Value类型交互

intersection()交集

1）函数签名：def intersection(other: RDD[T]): RDD[T]

2）功能说明

对源RDD和参数RDD求交集后返回一个新的RDD

交集：只有3

3）需求说明：创建两个RDD，求两个RDD的交集

union()并集不去重

1）函数签名：def union(other: RDD[T]): RDD[T]

2）功能说明

对源RDD和参数RDD求并集后返回一个新的RDD

并集：1、2、3全包括

3）需求说明：创建两个RDD，求并集

subtract()差集

1）函数签名：def subtract(other: RDD[T]): RDD[T]

2）功能说明

计算差的一种函数，去除两个RDD中相同元素，不同的RDD将保留下来

差集：只有1

3）需求说明：创建两个RDD，求第一个RDD与第二个RDD的差集

zip()拉链

1）函数签名：def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]

2）功能说明

该操作可以将两个RDD中的元素，以键值对的形式进行合并。其中，键值对中的Key为第1个RDD中的元素，Value为第2个RDD中的元素。

将两个RDD组合成Key/Value形式的RDD，这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常。

3）需求说明：创建两个RDD，并将两个RDD组合到一起形成一个(k,v)RDD

2.3.3 Key-Value类型

partitionBy()按照K重新分区

1）函数签名：def partitionBy(partitioner: Partitioner): RDD[(K, V)]

2）功能说明

将RDD[K,V]中的K按照指定Partitioner重新进行分区；

如果原有的RDD和新的RDD是一致的话就不进行分区，否则会产生Shuffle过程。

3）需求说明：创建一个3个分区的RDD，对其重新分区

2）自定义分区器

要实现自定义分区器，需要继承org.apache.spark.Partitioner类，并实现下面三个方法。

（1）numPartitions: Int:返回创建出来的分区数。

（2）getPartition(key: Any): Int:返回给定键的分区编号（0到numPartitions-1）。

（3）equals():Java 判断相等性的标准方法。这个方法的实现非常重要，Spark需要用这个方法来检查你的分区器对象是否和其他分区器实例相同，这样Spark才可以判断两个RDD的分区方式是否相同

reduceByKey()按照K聚合V

1）函数签名：

def reduceByKey(func: (V, V) => V): RDD[(K, V)]

def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

2）功能说明：该操作可以将RDD[K,V]中的元素按照相同的K对V进行聚合。其存在多种重载形式，还可以设置新RDD的分区数。

3）需求说明：统计单词出现次数

groupByKey()按照K重新分组

1）函数签名：def groupByKey(): RDD[(K, Iterable[V])]

2）功能说明

groupByKey对每个key进行操作，但只生成一个seq，并不进行聚合。

该操作可以指定分区器或者分区数（默认使用HashPartitioner）

3）需求说明：统计单词出现次数

reduceByKey和groupByKey区别

1）reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[K,V]。

2）groupByKey：按照key进行分组，直接进行shuffle。

3）开发指导：在不影响业务逻辑的前提下，优先选用reduceByKey。求和操作不影响业务逻辑，求平均值影响业务逻辑。

aggregateByKey()按照K处理分区内和分区间逻辑

aggregateByKey存在函数柯里化，有两个参数列表

第一个参数列表，,需要传递一个参数，表示为初始值

主要用于当碰见第一个key的时候，和value进行分区内计算

第二个参数列表需要传递2个参数

第一个参数表示分区内计算规则

第二个参数表示分区间计算规则

foldByKey()分区内和分区间相同的aggregateByKey()

combineByKey()转换结构后分区内和分区间操作

combineByKey :方法需要三个参数

第一个参数表示:将相同key的第一个数据进行结构的转换，实现操作

第二个参数表示:分区内的计算规则

第三个参数表示:分区间的计算规则

1）函数签名：

      def combineByKey[C](

      createCombiner: V => C,

      mergeValue: (C, V) => C,

      mergeCombiners: (C, C) => C): RDD[(K, C)]

（1）createCombiner（转换数据的结构）: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就和之前的某个元素的键相同。如果这是一个新的元素，combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值

（2）mergeValue（分区内）: 如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并

（3）mergeCombiners（分区间）: 由于每个分区都是独立处理的，因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器，就需要使用用户提供的mergeCombiners()方法将各个分区的结果进行合并。

2）功能说明

针对相同K，将V合并成一个集合。

3）需求说明：创建一个pairRDD，根据key计算每种key的平均值。（先计算每个key出现的次数以及可以对应值的总和，再相除得到结果）

sortByKey()按照K进行排序

1）函数签名：

       def sortByKey(

       ascending: Boolean = true, // 默认，升序

       numPartitions: Int = self.partitions.length)  : RDD[(K, V)]

2）功能说明

在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

3）需求说明：创建一个pairRDD，按照key的正序和倒序进行排序

mapValues()只对V进行操作

1）函数签名：def mapValues[U](f: V => U): RDD[(K, U)]

2）功能说明：针对于(K,V)形式的类型只对V进行操作

3）需求说明：创建一个pairRDD，并将value添加字符串"|||"

join()等同于sql里的内连接,关联上的要,关联不上的舍弃

1）函数签名：

def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]

def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]

2）功能说明

在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

3）需求说明：创建两个pairRDD，并将key相同的数据聚合到一个元组。

cogroup()类似于sql的全连接，但是在同一个RDD中对key聚合

1）函数签名：def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]

2）功能说明

在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable,Iterable))类型的RDD

操作两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。

3）需求说明：创建两个pairRDD，并将key相同的数据聚合到一个迭代器。

2.4 Action行动算子

行动算子是触发了整个作业的执行。因为转换算子都是懒加载，并不会立即执行。

2.4.1 reduce()聚合

1）函数签名：def reduce(f: (T, T) => T): T

2）功能说明：f函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据。

2.4.2 collect()以数组的形式返回数据集

1）函数签名：def collect(): Array[T]

2）功能说明：在驱动程序中，以数组Array的形式返回数据集的所有元素。

注意：所有的数据都会被拉取到Driver端，慎用

2.4.3 count()返回RDD中元素个数

1）函数签名：def count(): Long

2）功能说明：返回RDD中元素的个数

2.4.4 first()返回RDD中的第一个元素

1）函数签名：def first(): T

2）功能说明：返回RDD中的第一个元素

2.4.5 take()返回由RDD前n个元素组成的数组

1）函数签名：def take(num: Int): Array[T]

2）功能说明：返回一个由RDD的前n个元素组成的数组

2.4.6 takeOrdered()返回该RDD排序后前n个元素组成的数组

1）函数签名：def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]

2）功能说明：返回该RDD排序后的前n个元素组成的数组

2.4.7 aggregate()案例

2.4.8 fold()案例

2.4.9 countByKey()统计每种key的个数

1）函数签名：def countByKey(): Map[K, Long]

2）功能说明：统计每种key的个数

2.4.10 save相关算子

1）saveAsTextFile(path)保存成Text文件

        （1）函数签名

（2）功能说明：将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本

2）saveAsSequenceFile(path) 保存成Sequencefile文件

        （1）函数签名

（2）功能说明：将数据集中的元素以Hadoop Sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。

注意：只有kv类型RDD有该操作，单值的没有

3）saveAsObjectFile(path) 序列化成对象保存到文件

（1）函数签名

（2）功能说明：用于将RDD中的元素序列化成对象，存储到文件中。

2.4.11 foreach(f)遍历RDD中每一个元素

2.6 RDD依赖关系

2.6.1 查看血缘关系

RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。

2.6.3 窄依赖

窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition使用(一对一or多对一)，窄依赖我们形象的比喻为独生子女。

2.6.4 宽依赖

宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖(只能是一对多)，会引起Shuffle，总结：宽依赖我们形象的比喻为超生。

具有宽依赖的transformations包括：sort、reduceByKey、groupByKey、join和调用rePartition函数的任何操作。

宽依赖对Spark去评估一个transformations有更加重要的影响，比如对性能的影响。

在不影响业务要求的情况下，要尽量避免使用有宽依赖的转换算子，因为有宽依赖,就一定会走shuffle，影响性能

2.6.5 Stage任务划分（面试重点）

1）DAG有向无环图

DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。例如，DAG记录了RDD的转换过程和任务的阶段。

2）任务运行的整体流程

3）RDD任务切分中间分为：Application、Job、Stage和Task

（1）Application：初始化一个SparkContext即生成一个Application；

（2）Job：一个Action算子就会生成一个Job；

（3）Stage：Stage等于宽依赖的个数加1；

（4）Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。

注意：Application->Job->Stage->Task每一层都是1对n的关系。

2.7 RDD持久化

2.7.1 RDD Cache缓存

RDD通过Cache或者Persist方法将前面的计算结果缓存，默认情况下会把数据以序列化的形式缓存在JVM的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的action算子时，该RDD将会被缓存在计算节点的内存中，并供后面重用。

缓存有可能丢失，或者存储于内存的数据由于内存不足而被删除，RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换，丢失的数据会被重算，由于RDD的各个Partition是相对独立的，因此只需要计算丢失的部分即可，并不需要重算全部Partition。

3）自带缓存算子

Spark会自动对一些Shuffle操作的中间数据做持久化操作（比如：reduceByKey）。这样做的目的是为了当一个节点Shuffle失败了避免重新计算整个输入。但是，在实际使用的时候，如果想重用数据，仍然建议调用persist或cache。

2.7.2 RDD CheckPoint检查点

1）检查点：是通过将RDD中间结果写入磁盘。

2）为什么要做检查点？

由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。

3）检查点存储路径：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统

4）检查点数据存储格式为：二进制的文件

5）检查点切断血缘：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。

6）检查点触发时间：对RDD进行Checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍。

2.7.3 缓存和检查点区别

1）Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。

2）Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。

3）建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。

4）如果使用完了缓存，可以通过unpersist()方法释放缓存

广播变量：分布式共享只读变量。

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark Task操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，广播变量用起来会很顺手。在多个Task并行操作中使用同一个变量，但是Spark会为每个Task任务分别发送。

1）使用广播变量步骤：

（1）调用SparkContext.broadcast（广播变量）创建出一个广播对象，任何可序列化的类型都可以这么实现。

（2）通过广播变量.value，访问该对象的值。

（3）广播变量只会被发到各个节点一次，作为只读值处理（修改这个值不会影响到别的节点）。

累加器：分布式共享只写变量。（Executor和Executor之间不能读数据）

累加器用来把Executor端变量信息聚合到Driver端。在Driver中定义的一个变量，在Executor端的每个task都会得到这个变量的一份新的副本，每个task更新这些副本的值后，传回Driver端进行合并计算。

性能优化参考Spark性能优化指南——基础篇 - 美团技术团队

Spark性能优化指南——高级篇 - 美团技术团队

1. 数据压缩：使用压缩格式（如Snappy或Gzip）来减少数据在磁盘和网络传输中的大小，从而减少IO开销。

2. 数据分区：合理地对数据进行分区，以便在执行操作时能够充分利用并行性。可以使用repartition或coalesce方法来重新分区数据。

3. 广播变量：对于较小的数据集，可以将其广播到所有的工作节点上，以减少数据传输开销。

4. 内存管理：通过调整Spark的内存分配参数，如spark.executor.memory和spark.driver.memory，来优化内存使用。

5. 数据持久化：对于需要多次使用的数据集，可以使用persist或cache方法将其缓存到内存中，以避免重复计算。

6. 使用合适的数据结构和算法：根据具体的业务需求，选择合适的数据结构和算法，以提高计算效率。

7. 并行度设置：根据集群的规模和资源情况，适当调整并行度参数，如spark.default.parallelism和spark.sql.shuffle.partitions。

8. 数据倾斜处理：当某些键的数据量远远大于其他键时，可能会导致任务不均衡。可以使用一些技术，如repartition、join优化或使用自定义分区器来处理数据倾斜问题。

9. 使用DataFrame和Dataset：相比于RDD，DataFrame和Dataset提供了更高层次的抽象和优化，可以更好地利用Spark的优化功能。

10. 使用专门的优化工具：Spark提供了一些专门的优化工具，如Spark UI和Spark调优指南，可以帮助识别和解决性能瓶颈。

1.1 什么是Spark SQL

1.2 为什么要有Spark SQL

1.3 Spark SQL原理

1.3.1 什么是DataFrame

1）DataFrame是一种类似RDD的分布式数据集，类似于传统数据库中的二维表格。

2）DataFrame与RDD的主要区别在于，DataFrame带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。

左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解Person类的内部结构。而右侧的DataFrame却提供了详细的结构信息，使得Spark SQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。

3）Spark SQL性能上比RDD要高。因为Spark SQL了解数据内部结构，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在Stage层面进行简单、通用的流水线优化。

1.3.2 什么是DataSet

DataSet是分布式数据集合。

• DataSet是强类型的。比如可以有DataSet[Car]，DataSet[User]。具有类型安全检查
• DataFrame是DataSet的特例，type DataFrame = DataSet[Row] ，Row是一个类型，跟Car、User这些的类型一样，所有的表结构信息都用Row来表示。

1.3.3 RDD、DataFrame和DataSet之间关系

 1）发展历史

RDD（Spark1.0）=》Dataframe（Spark1.3）=》Dataset（Spark1.6）

如果同样的数据都给到这三个数据结构，他们分别计算之后，都会给出相同的结果。不同的是他们的执行效率和执行方式。在后期的Spark版本中，DataSet有可能会逐步取代RDD和DataFrame成为唯一的API接口。

2）三者的共性

（1）RDD、DataFrame、DataSet全都是Spark平台下的分布式弹性数据集，为处理超大型数据提供便利

（2）三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action行动算子如foreach时，三者才会开始遍历运算

（3）三者有许多共同的函数，如filter，排序等

（4）三者都会根据Spark的内存情况自动缓存运算

（5）三者都有分区的概念

1.4 Spark SQL的特点

1）易整合

无缝的整合了SQL查询和Spark编程。

2）统一的数据访问方式

使用相同的方式连接不同的数据源。

3）兼容Hive

在已有的仓库上直接运行SQL或者HQL。

4）标准的数据连接

通过JDBC或者ODBC来连接

2.1 SparkSession新的起始点

在老的版本中，SparkSQL提供两种SQL查询起始点：

• 一个叫SQLContext，用于Spark自己提供的SQL查询；
• 一个叫HiveContext，用于连接Hive的查询。

SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。

SparkSession内部封装了SparkContext，所以计算实际上是由SparkContext完成的。当我们使用spark-shell的时候，Spark框架会自动的创建一个名称叫做Spark的SparkSession，就像我们以前可以自动获取到一个sc来表示SparkContext。

2.2 DataFrame

DataFrame是一种类似于以RDD的分布式数据集，类似于传统数据库中的二维表格。

2.2.1 创建DataFrame

在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：

• 通过Spark的数据源进行创建；
• 从一个存在的RDD进行转换；
• 还可以从Hive Table进行查询返回。

2.2.2 SQL风格语法

SQL语法风格是指我们查询数据的时候使用SQL语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助。

视图：对特定表的数据的查询结果重复使用。View只能查询，不能修改和插入。

select * from t_user where age > 30 的查询结果可以存储在临时表(视图)v_user_age中，方便在后面重复使用。例如：select * from v_user_age

2.2.3 DSL风格语法

DataFrame提供一个特定领域语言（domain-specific language，DSL）去管理结构化的数据，可以在Scala，Java，Python和R中使用DSL，使用DSL语法风格不必去创建临时视图了。

2.3 DataSet

DataSet是具有强类型的数据集合，需要提供对应的类型信息。

2.3.1 创建DataSet（基本类型序列）

使用基本类型的序列创建DataSet

（1）将集合转换为DataSet

scala> val ds = Seq(1,2,3,4,5,6).toDS

ds: org.apache.spark.sql.Dataset[Int] = [value: int]

（2）查看DataSet的值

scala> ds.show

+-----+

|value|

+-----+

|    1|

|    2|

|    3|

|    4|

|    5|

|    6|

+-----+

2.3.2 创建DataSet（样例类序列）

使用样例类序列创建DataSet

（1）创建一个User的样例类

scala> case class User(name: String, age: Long)

defined class User

（2）将集合转换为DataSet

scala> val caseClassDS = Seq(User("wangyuyan",18)).toDS()

caseClassDS: org.apache.spark.sql.Dataset[User] = [name: string, age: bigint]

（3）查看DataSet的值

scala> caseClassDS.show

+---------+---+

|     name|age|

+---------+---+

|wangyuyan|  18|

+---------+---+

注意：在实际开发的时候，很少会把序列转换成DataSet，更多是通过RDD和DataFrame转换来得到DataSet

2.4 RDD、DataFrame、DataSet相互转换

3.1 加载数据

1）加载数据通用方法

spark.read.load是加载数据的通用方法

3.2 保存数据

1）保存数据通用方法

df.write.save是保存数据的通用方法

3.3 与MySQL交互

（1）pom依赖

  mysql
  mysql-connector-java
  8.0.29

（2）API

import org.apache.spark.sql.SparkSession
 
import java.util.Properties
 
/**
 * 加载外部数据源
 * MySQL
 */
object DataFrameMysql {
  def main(args: Array[String]): Unit = {
    val spark =
      SparkSession.builder()
        .appName("sparkmysql").master("local[*]").getOrCreate()
    val sc = spark.sparkContext
 
    val url = "jdbc:mysql://192.168.142.129:3306/sql50"
    val user = "root"
    val password = "123456"
    val driver = "com.mysql.cj.jdbc.Driver"
    val properties = new Properties()
    properties.setProperty("user", user)
    properties.setProperty("password", password)
    properties.setProperty("driver", driver)
 
    val studentDF = spark.read.jdbc(url, "student", properties)
    val teacherDF = spark.read.jdbc(url, "teacher", properties)
    val scoreDF = spark.read.jdbc(url, "score", properties)
    val courseDF = spark.read.jdbc(url, "course", properties)
    studentDF.printSchema()
    studentDF.show()
   
    spark.close()
  }
}

3.4 与Hive交互

SparkSQL可以采用内嵌Hive，也可以采用外部Hive。企业开发中，通常采用外部Hive。

（1）交互准备

拷贝hive-site.xml至spark的conf目录下

[root@kb129 conf]# cp /opt/soft/hive312/conf/hive-site.xml ./

[root@kb129 conf]# vim ./hive-site.xml   追加配置

        hive.metastore.uris

        thrift://192.168.142.129:9083

    hive.aux.jars.path

file:///opt/soft/hive312/lib/hive-hbase-handler-3.1.2.jar,file:///opt/soft/hive312/lib/zookeeper-3.4.6.jar,file:///opt/soft/hive312/lib/hbase-client-2.3.5.jar,file:///opt/soft/hive312/lib/hbase-common-2.3.5.jar,file:///opt/soft/hive312/lib/hbase-common-2.3.5-tests.jar,file:///opt/soft/hive312/lib/hbase-server-2.3.5.jar,file:///opt/soft/hive312/lib/hbase-common-2.3.5.jar,file:///opt/soft/hive312/lib/hbase-protocol-2.3.5.jar,file:///opt/soft/hive312/lib/htrace-core-3.2.0-incubating.jar

（2）启动hadoop和hive服务，成功连接上hive后，编写API

用例1：数据清洗

1. 读入日志文件并转化为RDD[Row]类型
   1. 按照Tab切割数据
   2. 过滤掉字段数量少于8个的
2. 对数据进行清洗
   1. 按照第一列和第二列对数据进行去重
   2. 过滤掉状态码非200
   3. 过滤掉event_time为空的数据
   4. 将url按照”&”以及”=”切割
3. 保存数据
   1. 将数据写入mysql表中

日志拆分字段：

event_time  url  method  status  sip  user_uip  action_prepend  action_client

（1）创建DataFrame的两种方式

（2）数据清洗

（3）将数据写入MySQL中

1）创建JDBC工具类，配置连接及定义写入数据方式（Append或OverWrite）

2）写入代码

完整代码

（1）JdbcUtils

import org.apache.spark.sql.{DataFrame, SaveMode}
 
import java.util.Properties
 
object JdbcUtils {
  private val url = "jdbc:mysql://kb129:3306/logdemo?createDatabaseIfNotExist=true"
  private val driver = "com.mysql.cj.jdbc.Driver"
  private val user = "root"
  private val password = "123456"
 
  private val properties = new Properties()
  properties.setProperty("user", JdbcUtils.user )
  properties.setProperty("password", JdbcUtils.password)
  properties.setProperty("driver", JdbcUtils.driver)
 
  def dataFrameToMysql(df:DataFrame, table:String, op:Int=1):Unit = {
    if (op == 0) {
      df.write.mode(SaveMode.Append).jdbc(JdbcUtils.url, table, properties)
    }
    else {
      df.write.mode(SaveMode.Overwrite).jdbc(JdbcUtils.url, table, properties)
    }
  }
}

（2）ETLDemo

import nj.zb.kb23.etl.utils.JdbcUtils
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}
 
object ETLDemo {
  case class Log(event_time: String, url: String, method: String, status: String,
                 sip: String, user_uip: String, action_prepend: String, action_client: String)
 
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("etldemo")
      .master("local[*]").getOrCreate()
    val sc = spark.sparkContext
 
    import spark.implicits._
    import org.apache.spark.sql.functions._
 
    /*//样例类方式创建dataframe
    val logDF = sc.textFile("in/test.log")
      .map(x => x.split("\t"))
      .filter(x => x.length >= 8)
      .map(x => Log(x(0), x(1), x(2), x(3), x(4), x(5), x(6), x(7))).toDF()
    logDF.show*/
 
 
    //创建ROW RDD
    val rowRDD = sc.textFile("in/test.log")
      // 按照Tab切割数据
      .map(x => x.split("\t"))
      // 过滤掉字段数量少于8个的
      .filter(x => x.length >= 8)
      .map(x => Row(x(0), x(1), x(2), x(3), x(4), x(5), x(6), x(7)))
    //rowRDD.collect().foreach(println)
 
    //创建schema
    val schema = StructType(
      Array(
        StructField("event_time", StringType),
        StructField("url", StringType),
        StructField("method", StringType),
        StructField("status", StringType),
        StructField("sip", StringType),
        StructField("user_uip", StringType),
        StructField("action_prepend", StringType),
        StructField("action_client", StringType)
      )
    )
    //创建dataframe
    val logDF = spark.createDataFrame(rowRDD, schema)
    //logDF.show()
 
 
    //1 按照第一列和第二列对数据进行去重
    val full_log = logDF.dropDuplicates("event_time", "url")
      //2 过滤掉状态码非200
      .filter(x => x(3) == "200")
      //3 过滤掉event_time为空的数据
      .filter(x => StringUtils.isNotEmpty(x(0).toString))
      //4 将url按照”&”以及”=”切割
      .map(line => {
        val str = line.getAs[String]("url")
        val paramArray = str.split("\\?")
        var paramsMap: Map[String, String] = null
        if (paramArray.length == 2) {
          val tuples = paramArray(1).split("&")
            .map(x => x.split("="))
            .filter(x => x.length == 2)
            .map(x => (x(0), x(1)))
          paramsMap = tuples.toMap
        }
        (
          line.getAs[String]("event_time"),
          paramsMap.getOrElse[String]("userUID", " "),
          paramsMap.getOrElse[String]("userSID", " "),
          paramsMap.getOrElse[String]("actionBegin", " "),
          paramsMap.getOrElse[String]("actionEnd", " "),
          paramsMap.getOrElse[String]("actionType", " "),
          paramsMap.getOrElse[String]("actionName", " "),
          paramsMap.getOrElse[String]("actionValue", " "),
          paramsMap.getOrElse[String]("actionTest", " "),
          paramsMap.getOrElse[String]("ifEquipment", " "),
          line.getAs[String]("method"),
          line.getAs[String]("status"),
          line.getAs[String]("sip"),
          line.getAs[String]("user_uip"),
          line.getAs[String]("action_prepend"),
          line.getAs[String]("action_client")
        )
      }).toDF("event_time", "userUID", "userSID", "actionBegin",
      "actionEnd", "actionType", "actionName", "actionValue", "actionTest", "ifEquipment"
      , "method", "status", "sip", "user_uip", "action_prepend", "action_client")
    //full_log.printSchema()
    //full_log.show(3)
 
    //将数据写入mysql表中
    JdbcUtils.dataFrameToMysql(full_log, "full_log")
 
    spark.close()
  }
}

Json使用参考https://www.cnblogs.com/tomato0906/articles/7291178.html

练习1—解析Json字符串及转DataFrame转JSON

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
 
object JsonStu {
  // 定义样例类
  case class DeviceData(id: Int, device: String)
 
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("retention").master("local[*]").getOrCreate()
    val sc = spark.sparkContext
 
    import spark.implicits._
    import org.apache.spark.sql.functions._
 
    //创建schema结构
    val jsonSchema = new StructType()
      .add("battery_level", LongType)
      .add("c02_level",LongType)
      .add("cca3",StringType)
      .add("cn", StringType)
      .add("device_id",LongType)
      .add("device_type", StringType)
      .add("signal", LongType)
      .add("ip",StringType)
      .add("temp", LongType)
      .add("timestamp", TimestampType)
 
    val eventsFromJSONDS = Seq(
      (0, """{"device_id": 0, "device_type": "sensor-ipad", "ip": "68.161.225.1", "cca3": "USA", "cn": "United States", "temp": 25, "signal": 23, "battery_level": 8, "c02_level": 917, "timestamp" :1475600496 }"""),
      (1, """{"device_id": 1, "device_type": "sensor-igauge", "ip": "213.161.254.1", "cca3": "NOR", "cn": "Norway", "temp": 30, "signal": 18, "battery_level": 6, "c02_level": 1413, "timestamp" :1475600498 }"""),
      (2, """{"device_id": 2, "device_type": "sensor-ipad", "ip": "88.36.5.1", "cca3": "ITA", "cn": "Italy", "temp": 18, "signal": 25, "battery_level": 5, "c02_level": 1372, "timestamp" :1475600500 }"""),
      (3, """{"device_id": 3, "device_type": "sensor-inest", "ip": "66.39.173.154", "cca3": "USA", "cn": "United States", "temp": 47, "signal": 12, "battery_level": 1, "c02_level": 1447, "timestamp" :1475600502 }"""),
      (4, """{"device_id": 4, "device_type": "sensor-ipad", "ip": "203.82.41.9", "cca3": "PHL", "cn": "Philippines", "temp": 29, "signal": 11, "battery_level": 0, "c02_level": 983, "timestamp" :1475600504 }"""),
      (5, """{"device_id": 5, "device_type": "sensor-istick", "ip": "204.116.105.67", "cca3": "USA", "cn": "United States", "temp": 50, "signal": 16, "battery_level": 8, "c02_level": 1574, "timestamp" :1475600506 }"""),
      (6, """{"device_id": 6, "device_type": "sensor-ipad", "ip": "220.173.179.1", "cca3": "CHN", "cn": "China", "temp": 21, "signal": 18, "battery_level": 9, "c02_level": 1249, "timestamp" :1475600508 }"""),
      (7, """{"device_id": 7, "device_type": "sensor-ipad", "ip": "118.23.68.227", "cca3": "JPN", "cn": "Japan", "temp": 27, "signal": 15, "battery_level": 0, "c02_level": 1531, "timestamp" :1475600512 }"""),
      (8, """ {"device_id": 8, "device_type": "sensor-inest", "ip": "208.109.163.218", "cca3": "USA", "cn": "United States", "temp": 40, "signal": 16, "battery_level": 9, "c02_level": 1208, "timestamp" :1475600514 }"""),
      (9, """{"device_id": 9, "device_type": "sensor-ipad", "ip": "88.213.191.34", "cca3": "ITA", "cn": "Italy", "temp": 19, "signal": 11, "battery_level": 0, "c02_level": 1171, "timestamp" :1475600516 }"""))
      .toDF("id", "json")
 
    //eventsFromJSONDS.printSchema()
    //eventsFromJSONDS.show(3)
 
    //直接解析
    val df = eventsFromJSONDS.select(
      get_json_object($"json", "$.device_id").as("device_id"),
      get_json_object($"json", "$.device_type").as("device_type"),
      get_json_object($"json", "$.ip").as("ip"),
      get_json_object($"json", "$.cca3").as("cca3"),
      get_json_object($"json", "$.cn").as("cn"),
      get_json_object($"json", "$.temp").as("temp"),
      get_json_object($"json", "$.signal").as("signal"),
      get_json_object($"json", "$.battery_level").as("battery_level"),
      get_json_object($"json", "$.c02_level").as("c02_level"),
      get_json_object($"json", "$.timestamp").as("timestamp")
    )
    df.printSchema()
    df.show(3)
 
    //使用样例类格式创建DF
    val eventsDS = Seq(
      (0, """{"device_id": 0, "device_type": "sensor-ipad", "ip": "68.161.225.1", "cca3": "USA", "cn": "United States", "temp": 25, "signal": 23, "battery_level": 8, "c02_level": 917, "timestamp" :1475600496 }"""),
      (1, """{"device_id": 1, "device_type": "sensor-igauge", "ip": "213.161.254.1", "cca3": "NOR", "cn": "Norway", "temp": 30, "signal": 18, "battery_level": 6, "c02_level": 1413, "timestamp" :1475600498 }"""),
      (2, """{"device_id": 2, "device_type": "sensor-ipad", "ip": "88.36.5.1", "cca3": "ITA", "cn": "Italy", "temp": 18, "signal": 25, "battery_level": 5, "c02_level": 1372, "timestamp" :1475600500 }"""),
      (3, """{"device_id": 3, "device_type": "sensor-inest", "ip": "66.39.173.154", "cca3": "USA", "cn": "United States", "temp": 47, "signal": 12, "battery_level": 1, "c02_level": 1447, "timestamp" :1475600502 }"""),
      (4, """{"device_id": 4, "device_type": "sensor-ipad", "ip": "203.82.41.9", "cca3": "PHL", "cn": "Philippines", "temp": 29, "signal": 11, "battery_level": 0, "c02_level": 983, "timestamp" :1475600504 }"""),
      (5, """{"device_id": 5, "device_type": "sensor-istick", "ip": "204.116.105.67", "cca3": "USA", "cn": "United States", "temp": 50, "signal": 16, "battery_level": 8, "c02_level": 1574, "timestamp" :1475600506 }"""),
      (6, """{"device_id": 6, "device_type": "sensor-ipad", "ip": "220.173.179.1", "cca3": "CHN", "cn": "China", "temp": 21, "signal": 18, "battery_level": 9, "c02_level": 1249, "timestamp" :1475600508 }"""),
      (7, """{"device_id": 7, "device_type": "sensor-ipad", "ip": "118.23.68.227", "cca3": "JPN", "cn": "Japan", "temp": 27, "signal": 15, "battery_level": 0, "c02_level": 1531, "timestamp" :1475600512 }"""),
      (8, """ {"device_id": 8, "device_type": "sensor-inest", "ip": "208.109.163.218", "cca3": "USA", "cn": "United States", "temp": 40, "signal": 16, "battery_level": 9, "c02_level": 1208, "timestamp" :1475600514 }"""),
      (9, """{"device_id": 9, "device_type": "sensor-ipad", "ip": "88.213.191.34", "cca3": "ITA", "cn": "Italy", "temp": 19, "signal": 11, "battery_level": 0, "c02_level": 1171, "timestamp" :1475600516 }"""))
      .toDF("id", "device").as[DeviceData]
 
    val frame = eventsDS.select(from_json($"device", jsonSchema) as "devices")
    frame.printSchema()
    frame.show(3,false)
 
    //找出DF中的所有Struct 包含的所有子列
    frame.select($"devices.*").show(3)
    //找出DF中的Struct 指定的cn子列
    frame.select($"devices.device_id",$"devices.cn").show(3)
 
 
    //DataFrame转Json字符串
    val frame2 =
      frame.select(to_json(
        struct($"devices.device_id", $"devices.battery_level")).as("jsonstr")
      )
    frame2.show(false)
 
    spark.close()
  }
}

练习2----

解析如下json

{"name":"njzb","roomInfo":{"area":2000,"employee":98,"roomNum":30},"students":[{"classes":"kb01","stuId":"1","stuName":"zhangsan","teacher":"xingxing"},{"classes":"kb02","stuId":"2","stuName":"lisi","teacher":"gree"},{"classes":"kb03","stuId":"3","stuName":"wangwu","teacher":"gree"},{"classes":"kb05","stuId":"4","stuName":"zhaoliu","teacher":"xingxing"}],"teachers":[{"name":"gree","skill":"bigdata&java","yearNum":7},{"name":"xingxing","skill":"bigdata&java","yearNum":3}]}

import nj.zb.kb23.etl.utils.JdbcUtils
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
 
object JsonStu4 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("retention").master("local[*]").getOrCreate()
    val sc = spark.sparkContext
 
    import spark.implicits._
    import org.apache.spark.sql.functions._
 
    val rdd = sc.textFile("in/school.txt")
    //rdd.collect().foreach(println)
 
    //"name", "roomInfo", "students", "teachers"
    val sc1DF = rdd.toDF("school")
    //sc1DF.printSchema()
    //sc1DF.show(false)
 
    val sc2DF = sc1DF.select(
      get_json_object($"school", "$.name").as("name"),
      get_json_object($"school", "$.roomInfo").as("roomInfo"),
      get_json_object($"school", "$.students").as("students"),
      get_json_object($"school", "$.teachers").as("teachers")
    )
    //sc2DF.printSchema()
    //sc2DF.show(false)
 
    val sc3DF = sc2DF.select(
      $"name",
      get_json_object($"roomInfo", "$.area").as("area"),
      get_json_object($"roomInfo", "$.employee").as("employee"),
      get_json_object($"roomInfo", "$.roomNum").as("roomNum"),
      $"students", $"teachers"
    )
    //sc3DF.printSchema()
    //sc3DF.show(false)
 
    val stuSchema = StructType(
      Array(
        StructField("classes", StringType),
        StructField("stuId", StringType),
        StructField("stuName", StringType),
        StructField("teacher", StringType)
      )
    )
    val teaSchema = StructType(
      Array(
        StructField("name", StringType),
        StructField("skill", StringType),
        StructField("yearNum", StringType)
      )
    )
 
    val sc4StuDF = sc3DF.select(
      $"name", $"area", $"employee", $"roomNum",
      from_json($"students", ArrayType(stuSchema)).as("students"))
    //sc4StuDF.printSchema()
    //sc4StuDF.show(false)
   
    val sc4TeaDF = sc3DF.select(
      $"name", $"area", $"employee", $"roomNum",
      from_json($"teachers", ArrayType(teaSchema)).as("teachers"))
    //sc4TeaDF.printSchema()
    //sc4TeaDF.show(false)
   
    val sc5StuDF = sc4StuDF
      .withColumn("students", explode($"students"))
    //sc5StuDF.printSchema()
    //sc5StuDF.show(false)
 
    val sc5TeaDF = sc4TeaDF
      .withColumn("teachers", explode($"teachers"))
    //sc5TeaDF.printSchema()
    //sc5TeaDF.show(false)
 
    val sc6StuDF = sc5StuDF
      .withColumn("classes", $"students.classes")
      .withColumn("stuId", $"students.stuId")
      .withColumn("stuName", $"students.stuName")
      .withColumn("teacher", $"students.teacher")
      .drop("students")
    //sc6StuDF.show(false)
 
    val sc6TeaDF = sc5TeaDF
      .withColumn("teacherName", $"teachers.name")
      .withColumn("skill", $"teachers.skill")
      .withColumn("yearNum", $"teachers.yearNum")
      .drop("name")
      .drop("area")
      .drop("employee")
      .drop("roomNum")
      .drop("teachers")
    //sc6TeaDF.show(false)
 
    val schoolDF =
      sc6StuDF
        .join(sc6TeaDF, $"teacher" === $"teacherName")
        .drop("teacherName")
    schoolDF.show()
 
    //使用工具类将DF导入到Mysql中
    JdbcUtils.dataFrameToMysql(schoolDF,"school")
  }
}

第1章 SparkStreaming概述

1.1 Spark Streaming是什么

1.2 Spark Streaming架构原理

1.2.1 什么是DStream

第2章 DStream入门

2.1 WordCount案例实操

需求：使用netcat工具向9999端口不断的发送数据，通过SparkStreaming读取端口数据并统计不同单词出现的次数

监听端口信息及wordcount

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
 
object SparkStreamDemo {
  def main(args: Array[String]): Unit = {
    //1.初始化Spark配置信息
    val conf = new SparkConf().setMaster("local[2]").setAppName("spark_stream")
    //2.初始化SparkStreamingContext
    val streamingContext = new StreamingContext(conf, Seconds(3))
 
    //3.通过监控端口创建DStream，读进来的数据为一行行
    val socketLineStream = streamingContext.socketTextStream("kb129", 7777)
 
    //3.1 编写wordcount计算
    val wordCountStream = socketLineStream
      .flatMap(x => x.split("\\s+"))
      .map((_, 1))
      .reduceByKey(_ + _)
 
    //3.2 打印
    //socketLineStream.print()
    wordCountStream.print()
 
    //4 启动SparkStreamingContext
    streamingContext.start()
    // 将主线程阻塞，主线程不退出
    streamingContext.awaitTermination()
   
  }
}

Wordcount

2.2 WordCount解析

DStream是Spark Streaming的基础抽象，代表持续性的数据流和经过各种Spark算子操作后的结果数据流。

在内部实现上，每一批次的数据封装成一个RDD，一系列连续的RDD组成了DStream。对这些RDD的转换是由Spark引擎来计算。

说明：DStream中批次与批次之间计算相互独立。如果批次设置时间小于计算时间会出现计算任务叠加情况，需要多分配资源。通常情况，批次设置时间要大于计算时间。

第3章 DStream创建

3.1 RDD队列

3.1.1 用法及说明

测试方法：

（1）使用ssc.queueStream（queueOfRDDs）来创建DStream

（2）将每一个推送到这个队列中的RDD，都会作为DStream的一个批次处理。

3.3 Kafka数据源（面试、开发重点）

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
 
object SparkStreamingKafka {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("sparkStreamingKafka")
    val streamingContext = new StreamingContext(conf, Seconds(5))
 
//有状态数据操作时需要设置检查点路径
    streamingContext.checkpoint("checkpoint")
 
    //定义Kafka参数：kafka集群地址、消费者组名称、key序列化、value序列化
    val kafkaParams = Map(
      (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "kb129:9092"),
      (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer"),
      (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer"),
      (ConsumerConfig.GROUP_ID_CONFIG -> "sparkStreamGroup1")
    )
 
    //读取Kafka数据创建DStream
    val kafkaStream: InputDStream[ConsumerRecord[String, String]]
    = KafkaUtils.createDirectStream(
      streamingContext,
      LocationStrategies.PreferConsistent,//优先位置
      ConsumerStrategies.Subscribe(Set("sparkstreamstu"), kafkaParams)// 消费策略：（订阅多个主题，配置参数）
    )
 
   /* val dsStream = kafkaStream
      .map(x => (x.topic(), x.offset(), x.partition(), x.value()))
    dsStream.print*/
 
    //每个周期内（5秒）收集到的单词进行wordCount（单个周期）
    /*val wordCountStream = kafkaStream
      .flatMap(_.value().toString.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
    wordCountStream.print()*/
 
    //每个周期收集到的单词进行wordCount（累加）
    val wordCountStream =
      kafkaStream
      .flatMap(_.value().split(" "))
      .map((_, 1))
      .updateStateByKey(
        //定义更新状态方法，参数seq为当前批次单词次数，buffer缓冲区为以往批次单词次数
        (seq:Seq[Int], buffer:Option[Int]) => {
          println("进入updateStateByKey函数中")
          println("Seq value" + seq.toList.toString)
          println("buffer value"+ buffer.getOrElse(0).toString)
          //缓冲区内的value求和，开始时缓冲区无数据，给默认值0
          val sum = buffer.getOrElse(0) + seq.sum
          Option(sum)
        }
      )
    wordCountStream.print()
 
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}

第4章 DStream转换

DStream上的操作与RDD的类似，分为转换和输出两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及各种Window相关的原语。

4.1 无状态转化操作

无状态转化操作：就是把RDD转化操作应用到DStream每个批次上，每个批次相互独立，自己算自己的。

4.1.1 常规无状态转化操作

DStream的部分无状态转化操作列在了下表中，都是DStream自己的API。

注意，针对键值对的DStream转化操作，要添加import StreamingContext._才能在Scala中使用，比如reduceByKey()。

需要记住的是，尽管这些函数看起来像作用在整个流上一样，但事实上每个DStream在内部是由许多RDD批次组成，且无状态转化操作是分别应用到每个RDD(一个批次的数据)上的。

4.1.2 Transform

需求：通过Transform可以将DStream每一批次的数据直接转换为RDD的算子操作。

4.2 有状态转化操作

有状态转化操作：计算当前批次RDD时，需要用到历史RDD的数据。

4.2.1 UpdateStateByKey

updateStateByKey()用于键值对形式的DStream，可以记录历史批次状态。例如可以实现累加WordCount。

updateStateByKey()参数中需要传递一个函数，在函数内部可以根据需求对新数据和历史状态进行整合处理，返回一个新的DStream。

注意：使用updateStateByKey需要对检查点目录进行配置，会使用检查点来保存状态。

• checkpoint小文件过多
• checkpoint记录最后一次时间戳，再次启动的时候会把间隔时间的周期再执行一次

//每个周期收集到的单词进行wordCount（累加）
val wordCountStream =
  kafkaStream
  .flatMap(_.value().split(" "))
  .map((_, 1))
  .updateStateByKey(
    //定义更新状态方法，参数seq为当前批次单词次数，buffer缓冲区为以往批次单词次数
    (seq:Seq[Int], buffer:Option[Int]) => {
      println("进入updateStateByKey函数中")
      println("Seq value" + seq.toList.toString)
      println("buffer value"+ buffer.getOrElse(0).toString)
      //缓冲区内的value求和，开始时缓冲区无数据，给默认值0
      val sum = buffer.getOrElse(0) + seq.sum
      Option(sum)
    }
  )

4）原理说明

4.2.2 WindowOperations

Window Operations可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Streaming的允许状态。所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。

• 窗口时长：计算内容的时间范围；
• 滑动步长：隔多久触发一次计算。

注意：这两者都必须为采集批次大小的整数倍。

如下图所示WordCount案例：窗口大小为批次的2倍，滑动步等于批次大小。

4.2.3 Window

1）基本语法：window(windowLength, slideInterval): 基于对源DStream窗口的批次进行计算返回一个新的DStream。

2）需求：统计WordCount：3秒一个批次，窗口12秒，滑步6秒。

4.2.4 reduceByKeyAndWindow

1）基本语法：

reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])：当在一个(K,V)对的DStream上调用此函数，会返回一个新(K,V)对的DStream，此处通过对滑动窗口中批次数据使用reduce函数来整合每个key的value值。

4.2.5 reduceByKeyAndWindow（反向Reduce）

1）基本语法：

reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks]): 这个函数是上述函数的变化版本，每个窗口的reduce值都是通过用前一个窗的reduce值来递增计算。通过reduce进入到滑动窗口数据并“反向reduce”离开窗口的旧数据来实现这个操作。一个例子是随着窗口滑动对keys的“加”“减”计数。通过前边介绍可以想到，这个函数只适用于“可逆的reduce函数”，也就是这些reduce函数有相应的“反reduce”函数（以参数invFunc形式传入）。如前述函数，reduce任务的数量通过可选参数来配置。

4.2.6 Window的其他操作

（1）countByWindow(windowLength, slideInterval): 返回一个滑动窗口计数流中的元素个数；

（2）reduceByWindow(func, windowLength, slideInterval): 通过使用自定义函数整合滑动区间流元素来创建一个新的离散化数据流；

第5章 DStream输出

DStream通常将数据输出到，外部数据库或屏幕上。

DStream与RDD中的惰性求值类似，如果一个DStream及其派生出的DStream都没有被执行输出操作，那么这些DStream就都不会被求值。如果StreamingContext中没有设定输出操作，整个Context就都不会启动。

1）输出操作API如下：

• saveAsTextFiles(prefix, [suffix])：以text文件形式存储这个DStream的内容。每一批次的存储文件名基于参数中的prefix和suffix。“prefix-Time_IN_MS[.suffix]”。
• saveAsObjectFiles(prefix, [suffix])：以Java对象序列化的方式将DStream中的数据保存为 SequenceFiles 。每一批次的存储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]"。
• saveAsHadoopFiles(prefix, [suffix])：将Stream中的数据保存为 Hadoop files。每一批次的存储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]"。

注意：以上操作都是每一批次写出一次，会产生大量小文件，在生产环境，很少使用。

• print()：在运行流程序的驱动结点上打印DStream中每一批次数据的最开始10个元素。这用于开发和调试。
• foreachRDD(func)：这是最通用的输出操作，即将函数func用于产生DStream的每一个RDD。其中参数传入的函数func应该实现将每一个RDD中数据推送到外部系统，如将RDD存入文件或者写入数据库。

在企业开发中通常采用foreachRDD()，它用来对DStream中的RDD进行任意计算。这和transform()有些类似，都可以让我们访问任意RDD。在foreachRDD()中，可以重用我们在Spark中实现的所有行动操作(action算子)。比如，常见的用例之一是把数据写到如MySQL的外部数据库中。

3）注意

（1）连接不能写在Driver层面（序列化）

（2）如果写在foreach则每个RDD中的每一条数据都创建，得不偿失；

（3）增加foreachPartition，在分区创建（获取）。

第6章 优雅关闭

流式任务需要7*24小时执行，但是有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死，所以配置优雅的关闭就显得至关重要了。

关闭方式：使用外部文件系统来控制内部程序关闭。