Spark思维导图

Spark 的发展

Spark介绍

Spark内置模块

Spark使用场景

①local(本地模式)：常用于本地开发测试，本地还分为local单线程和local-cluster多线程；master和worker都为本机。运行时直接加断点调试即可。

②standalone(集群模式)：构建一个有Master+Slave构成的Spark集群，spark运行在集群中。简单来说，只用spark，不用yarn，所以不需要启动Hadoop

③on yarn(集群模式)： 运行在 yarn 资源管理器框架之上，由 yarn 负责资源管理，Spark 负责任务调度和计算

④on mesos(集群模式)： 运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark 负责任务调度和计算

⑤on cloud(集群模式)：比如 AWS 的 EC2，使用这个模式能很方便的访问 Amazon的 S3;Spark 支持多种分布式存储系统：HDFS 和 S3

Spark运行流程

Spark基本使用

RDD介绍

RDD属性

RDD特点

RDD的创建

RDD的保存

转换算子

行动算子

小结

在实际开发中我们往往需要自己定义一些对于RDD的操作，那么此时需要注意的是，初始化工作是在Driver中进行的，而实际运行是在Executor端进行的，这就涉及到了跨进程通信，是需要序列化的

一.RDD依赖关系基本介绍

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

二.查看RDD之间的依赖关系

list.toDebugString //查看RDD的所有依赖关系

list.dependencies //查看当前RDD上一层依赖关系

三.窄依赖

窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用，即一个分区的数据原封不动的给另一个分区就叫窄依赖。可以形象的比喻成独生子女 

四.宽依赖

宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition,这就会引起shuffle。简单的说，就有shuffle过程的都叫做宽依赖。可以形象的比喻成超生

五.DAG

DAG(Directed Acyclic Graph)叫做有向无环图，原始的RDD通过一系列转换就形成了DAG，根据DAG之间的依赖关系的不同将DAG划分成不同的stage（阶段）。对窄依赖，partition的转换处理在stage中完成计算。对于宽依赖，由于有shuffle的存在，只能在parent（起源）RDD处理完成后，才能进行接下来的计算，因此宽依赖是划分stage的依据 

六.任务划分（面试重点）

RDD缓存

RDD检查点

本地文件系统

MySQL数据库

Hbase数据库

累加器

广播变量

介绍

特点

两个编程抽象

DataFrame创建

DataSet创建

RDD与DataFrame相互转换

RDD与DataSet相互转换

DataFrame与DataSet相互转换

三者之间的共性

一.添加pom

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.11</artifactId>

<version>${spark.version}</version>

</dependency>

二.代码演示，创建Session，RDD DataFrame DataSet三者进行转换

def main(args: Array[String]): Unit = {

//创建配置对象

val conf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01")

//构建sparkSQL环境对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

//RDD,DataFrame,DataSet三者进行转换之前
//需要引入隐式转换规则
//这里的spark含义不是包名的意思，而是Session对象的名称
import spark.implicits._

//创建rdd
val rdd:RDD[(Int,String,Int)] = spark.sparkContext.makeRDD(List((1,"张三",20),(2,"李四",30),(1,"王五",22)))

//将 rdd 转换成 DF
val df:DataFrame = rdd.toDF("id","name","age")

//将 DF 转换成 DS
//前提：需要创建类型规程，即创建一个样例类
val ds:Dataset[User] = df.as[User]

//将 DS 转换成 DF
val df1 = ds.toDF()

//将 DF 转换成 RDD
val rdd1:RDD[Row] = df1.rdd

//打印rdd1
rdd1.foreach(
//打印姓名
row => {
println(row.getString(1))
println(row)
}
)

UDF

UDAF

文件的读取与保存

MySQL读取与存储

Spark操作hive数据库

一.基本介绍

1.Spark Straming用于流式数据处理。Spark Straming支持多种数据源输入，例如kafka，flume，twitter，zeroMQ和简单的TCP套接字等等。数据输入后可以用spark的高度抽象原语，如map，reduce，join，window进行操作运算。而结果也可以保存再很多地方，如HDFS，数据库等。

2.和spark基于RDD的概念很相似，spark streaming是用离散化流（discretized stream）作为抽象表示，叫做DStream。DStream是随着时间推移而收到的数据的序列。每个时间收到的数据都作为RDD存在，而DStream是由这些RDD组成的序列，因此得名“离散化

二.Spark Steaming特点

1.易用
2.容错
3.易整合到spark体系

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

二.Linux安装netcat工具

yum install -y nc
nc -lk 9999 //向9999端口发送消息

三.代码演示

//使用SparkSteaming完成wordcount
//构建sparkConf对象
val conf = new SparkConf().setMaster("local[*]").setAppName("wordcount")

//构建实时数据分析环境对象
//采集周期：以指定的时间为周期采集实时数据
val streamingContext:StreamingContext = new StreamingContext(conf, Seconds(3))

//从指定端口采集数据
val socketLineDStream:ReceiverInputDStream[String] = streamingContext.socketTextStream("192.168.220.25",9999)

//将采集的数据进行分解
val wordDStream:DStream[String] = socketLineDStream.flatMap(line => line.split(" "))

//将数据结构转换为k-v类型
val mapDStream:DStream[(String,Int)] = wordDStream.map(a => a -> 1)

//统计每个单词的数量
val mapToSumDStream:DStream[(String,Int)] = mapDStream.reduceByKey(_+_)

//将结果打印
mapToSumDStream.print()

//启动采集器
streamingContext.start()

//Driver等待采集器执行
streamingContext.awaitTermination()

一.DStream

Spark Streaming原生支持一些不同的数据源。一些和核心的数据源已经被打包到Spark Streaming的maven工件中，而其他的一些数据源则可以通过Spark-Streaming-kafka等附加工件获取。每个接收器都以spark执行器程序中一个长期运行的任务形式运行，因此会占据分配给应用的CPU核心。此外，我们还需要有可用的CPU核心来处理数据。这意味着如果要运行多个接收器，就必须至少有和接收器数目相同的核心数，还要加上用来完成计算所需要的核心数。例如：如果我们想要在流计算应用中运行10个接收器，那么至少为应用分配11个CPU核心，所以如果在本地运行，不要使用local[1]

二.文件数据源

文件数据源：能够读取所有HDFS API兼容的文件系统文件，通过fileStream方式进行读取，Spark Streaming将会监控dataDriectory目录不断移动进来的文件，记住目前不支持嵌套目录。

注意： ①文件需要有相同的数据格式

②文件进入dataDriectory的方式需要通过移动或者重命名来实现

③一旦文件移动进目录，则不能再修改，即便修改了也不会读取数据

三.代码演示

val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming02_FileDataSource")

//构建实时数据分析环境对象
//采集周期：以指定的时间为周期采集实时数据
val streamingContext:StreamingContext = new StreamingContext(conf, Seconds(5))

//从指定文件夹采集数据
val fileDStream:DStream[String] = streamingContext.textFileStream("D:\\Spark\\test")

//将采集的数据进行分解
val wordDStream:DStream[String] = fileDStream.flatMap(line => line.split(" "))

//将数据结构转换为k-v类型
val mapDStream:DStream[(String,Int)] = wordDStream.map(a => a -> 1)

//统计每个单词的数量
val mapToSumDStream:DStream[(String,Int)] = mapDStream.reduceByKey(_+_)

//将结果打印
mapToSumDStream.print()题

四.RDD队列用法及说明
测试过程中，可以通过使用ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列的RDD，都会作为一个DStream处理

五.需求
循环创建几个RDD，将RDD放入到队列中。通过SparkStreaming创建DStream，计算WordCount

一.用法和说明

在工程中需要引入Maven 工件spark-streaming-kafka_2.10 来使用它。包内提供的KafkaUtils 对象可以在StreamingContext 和JavaStreamingContext 中以你的Kafka 消息创建出DStream。由于KafkaUtils 可以订阅多个主题，因此它创建出的DStream 由成对的主题和消息组成。要创建出一个流数据，需要使用StreamingContext 实例、一个由逗号隔开的ZooKeeper 主机列表字符串、消费者组的名字(唯一名字)，以及一个从主题到针对这个主题的接收器线程数的映射表来调用createStream() 方法 

二.添加依赖

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
<version>2.4.3</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
<groupId>org.apache.kafka</groupId
<artifactId>kafka-clients</artifactId>
<version>2.3.0</version>
</dependency>

三.代码演示

package sparkStreaming

import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka._

object test01 {
def main(args: Array[String]): Unit = {

//待接收的topic名称
val topics = "name1,spark"

//broker地址
val brokers = "192.168.220.25:9092"

//创建SparkConf对象
val conf = new SparkConf().setMaster("local[*]").setAppName("kafka test")

//创建ssc对象
val ssc = new StreamingContext(conf,Seconds(3))

//分割待接收topic
val topicsSet = topics.split(",").toSet

//连接kafka的配置文件
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)

//根据配置文件获取Kafka消息
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSet
)

//读取Kafka消息
val lines = messages.map(_._2)


val words = lines.flatMap(_.split(" ")).map(x=>(x,1))

words.reduceByKey(_ + _).print()

ssc.start()

ssc.awaitTermination()

}

}

一.DStream转换基本介绍

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

二.无状态转化操作

无状态转化就是把简单的RDD转化操作应用到每个批次上，也就是转化DStream中的每一个RDD。简单的说就是每个批次的数据没有关联

三.有状态转化操作 （重点）

一.Window Operations（窗口操作）介绍

Window Operations可以设置窗口的大小和滑动的间隔来动态的获取当前Streaming的用于状态

//设置窗口函数

//窗口大小应该为采集周期的整数倍，窗口滑动的步长也应该为采集周期的整数倍

val window:DStream[(String,String)] = kafkaDStream.window(Seconds(9),Seconds(3))