Spark

Spark core

一、 spark是一个分布式同通用计算框架，可用于离线大数据处理、实时流计算、交互式计算、图计算等，集成了SparkSQL、SparkStream、MLLib等库，提供了丰富的API。

spark中针对弹性分布数据集RDD进行计算，RDD可缓存在内存或磁盘中，程序运行结束时删除。可通过sc.testFile()、其他RDD通过转化因子转化，DF/DS转化等方式构建成RDD。RDD会有多个分区，如果从HDFS上读取的数据，如果不指定分区，默认分区与block块一一对应，如果不指定最小分区数，默认为2。

spark提供两种算子对RDD进行计算，Transformation 和 Action，Transformation 为惰性执行，只记录关系，当执行Action算子时，才会触发Transformatin的一系列执行操作，生成新的RDD。一个Action对应一个job。

spark与mapReduce的相比，具有优势：

1.计算速度快，spark采用了缓存机制，中间结果不用写入HDFS再读取，减少了IO消耗。

2.spark采用DAG引擎、以及执行任务时，excutor中采用多线程方式处理，提升了任务执行速度。container中不能多线程吗？

2.mr只有map 和 reduce两种语义，实现复杂的逻辑不够灵活，spark提供了丰富的Transformation 和 action 算子，实现较为方便

二、spark的程序运行结构

1. Driver：Driver中封装了SparkContext，一个程序对应一个driver，main函数运行在driver中，driver负责程序的解析。sparkContext加载程序运行的环境，及创建内部的DAGscheduler和TaskSchedule。

TaskSchedule将任务调度到WordNode的Executor中，一和节点可以启动多个Executor中，每个Executor中可以以多线程方式运行多个Task。

spark程序内部执行流程：

1.根据Transformation算子，查询RDD间的依赖关系，构建逻辑查询计划

2.构建物理查询计划：由DAGSchedule 构建DAG图，根据算子是否为宽依赖进行Stage划分，一个Stage中生成一组Task，组成一个TaskSet，task可并行执行，提交给TaskSchedule

3.TaskSchedule将任务调度到Executor，以及负责任务失败的重新调度，执行较慢的任务启动备份执行等。

三、Spark-shuffle

划分Stage时，类似MR的Shuffle，sparkShuffle分为 shuffleWrite 和 shuffleRead阶段，

shuffleWrite：将Task中间结果写入磁盘

shuffleRead：从磁盘拉取数据到内存中进行并行计算

shuffleWrite 方式有：

基于Hash：将每一个mapTask的结果对ReduceTaskNum取余，将结果写入本地小文件，例，一个Executor中有2个MapTask，3个ReduceTask，则每个MapTask会生成3个小文件，分别对应3个ReduceTask，在写入小文件时，会先将数据写到buffer缓冲区，再写入小文件。基于Hash的方式会生成大量小文件，增加IO消耗，且消耗缓冲区。

基于Hash的优化版：如果一个Excutor中有多个MapTask，余数相同的会写入相同的小文件中，例，一个Executor中有2个MapTask，3个ReduceTask，则该Executor中会生成3个小文件，减少了小文件个数，但未减少缓冲区消耗。

基于Sort：

shuffleRead 方式：

基于Hash的优化版、基于Sort 的shuffleWrite 采用相同的shuffleRead实现，将拉取的数据写入到HashMap中，如果需要对key进行排序，则排序后写入HashMap中（HashMap本身是无序的，用的是linkHashMap？），HasHMap写满后，溢出一个小文件到磁盘中，最后将多个小文件进行归并排序合成大文件，进行处理输出。

四、Spark运行模型

1.本地local方式运行 spark-shell

2.独立集群模式

3.结合其他调度系统，例如Spark-on-YARN

参见 https://blog.csdn.net/swing2008/article/details/60869183

五、常用算子

reduceBykey、groupByKey等

SparkSQL

一、sparkSql：基于spark的SQL引擎，提供更加丰富的数据源及API：DF、DS；可通过SQL、DF、DS来对数据进行处理。

使用SQL操作Spark组件，进行数据分析。

与spark core相比：

1. 提供数据集类型DF（schema +RDD）、DS 处理结构化数据，比直接操作RDD简单方便

2. 提供更多灵活API，例如读写多种数据源的数据，json等，可以通过API直接解析

3.内置优化器catalyst自动优化程序（优化逻辑查询计划等），而且效率较高，若直接操作RDD，还需要自己优化程序。

相比直接利用Spark SQL处理数据，利用 spark DF/DS 处理数据，可以实现复杂的语义逻辑，例如实现算法等，用SQL处理较复杂的逻辑实现很困难。

RDD的算子，DF、DS基本都支持，RDD可处理 结构化/非结构化 数据，DF/DS 只能处理结构化数据。三者可相互转化。

spark SQL: 适用于交互式查询，实时查询，数据量不太大，响应时间快的操作。

Hive：适用于大规模的数据查询分析，对响应时间要求较低的操作。（底层是mr）

广泛应用场景：

部署 spark thrift server， 接受BI系统的SQL请求，转化为spark 引擎处理，读取hive中的数据。

二、程序流程

三、Spark Streaming

将流式计算转化为一批很小的，确定的批处理作业。以秒（分钟等）为单位将作业切分成很小的离散的作业（以时间为单位切分数据流）。  低延迟，秒级别。

特有算子：

window

mapWithStatus：类似Spark自己维护的分布式HashMap（不需要再借助外部存储，节约与外部存储交互的开销），状态信息以KeyValue形式存储状态信息。实时操作中的状态信息。适合保存短时间内的状态信息用于分析。