一、Spark简介
1.1 Spark是什么
Spark是一个通用的并行计算框架，由UCBerkeley的AMP实验室开发。Spark基于map reduce 算法模式实现的分布式计算，拥有Hadoop MapReduce所具有的优点；但不同于Hadoop MapReduce的是Job中间输出和结果可以保存在内存中，从而不再需要读写HDFS，节省了磁盘IO耗时，号称性能比Hadoop快100倍。
1.2 Spark官网
http://spark.apache.org/
1.3 Spark架构及生态
（1）Spark Core：包含Spark的基本功能；尤其是定义RDD的API、操作以及这两者上的动作。其他Spark的库都是构建在RDD和Spark Core之上的
（2）Spark SQL：提供通过Apache Hive的SQL变体Hive查询语言（HiveQL）与Spark进行交互的API。每个数据库表被当做一个RDD，Spark SQL查询被转换为Spark操作。
（3）Spark Streaming：对实时数据流进行处理和控制。Spark Streaming允许程序能够像普通RDD一样处理实时数据
（4）MLlib：一个常用机器学习算法库，算法被实现为对RDD的Spark操作。这个库包含可扩展的学习算法，比如分类、回归等需要对大量数据集进行迭代的操作。
（5）GraphX：控制图、并行图操作和计算的一组算法和工具的集合。GraphX扩展了RDD API，包含控制图、创建子图、访问路径上所有顶点的操作
1.4 Spark架构的组成图
（1）Cluster Manager：在standalone模式中即为Master主节点，控制整个集群，监控worker。在YARN模式中为资源管理器
（2）Worker节点：从节点，负责控制计算节点，启动Executor或者Driver。
（3）Driver： 运行Application 的main()函数
（4）Executor：执行器，是为某个Application运行在worker node上的一个进程
1.5 Spark三种集群模式
1.Standalone独立集群
2.Mesos, apache mesos
3.Yarn, hadoop yarn
1.6 Spark与hadoop比较
（1）Hadoop有两个核心模块，分布式存储模块HDFS和分布式计算模块Mapreduce
（2）spark本身并没有提供分布式文件系统，因此spark的分析大多依赖于Hadoop的分布式文件系统HDFS
（3）Hadoop的Mapreduce与spark都可以进行数据计算，而相比于Mapreduce，spark的速度更快并且提供的功能更加丰富
1.7 Spark运行流程
spark运行流程图如下：
（1）构建Spark Application的运行环境，启动SparkContext
（2）SparkContext向资源管理器（可以是Standalone，Mesos，Yarn）申请运行Executor资源，并启动StandaloneExecutorbackend，
（3）Executor向SparkContext申请Task
（4）SparkContext将应用程序分发给Executor
（5）SparkContext构建成DAG图，将DAG图分解成Stage、将Taskset发送给Task Scheduler，最后由Task Scheduler将Task发送给Executor运行
（6）Task在Executor上运行，运行完释放所有资源
1.8 Spark运行特点
（1）每个Application获取专属的executor进程，该进程在Application期间一直驻留，并以多线程方式运行Task。这种Application隔离机制是有优势的，无论是从调度角度看（每个Driver调度他自己的任务），还是从运行角度看（来自不同Application的Task运行在不同JVM中），当然这样意味着Spark Application不能跨应用程序共享数据，除非将数据写入外部存储系统
（2）Spark与资源管理器无关，只要能够获取executor进程，并能保持相互通信就可以了
（3）提交SparkContext的Client应该靠近Worker节点（运行Executor的节点），最好是在同一个Rack里，因为Spark Application运行过程中SparkContext和Executor之间有大量的信息交换
（4）Task采用了数据本地性和推测执行的优化机制
1.9 Spark运行模式
（1）Spark的运行模式多种多样，灵活多变，部署在单机上时，既可以用本地模式运行，也可以用伪分布模式运行，而当以分布式集群的方式部署时，也有众多的运行模式可供选择，这取决于集群的实际情况，底层的资源调度即可以依赖外部资源调度框架，也可以使用Spark内建的Standalone模式。
（2）对于外部资源调度框架的支持，目前的实现包括相对稳定的Mesos模式，以及hadoop YARN模式
（3）本地模式：常用于本地开发测试，本地还分别 local 和 local cluster
1.10 Spark特点
（1）运行速度快 => Spark拥有DAG执行引擎，支持在内存中对数据进行迭代计算。官方提供的数据表明，如果数据由磁盘读取，速度是Hadoop MapReduce的10倍以上，如果数据从内存中读取，速度可以高达100多倍。
（2）适用场景广泛 => 大数据分析统计，实时数据处理，图计算及机器学习
（3）易用性 => 编写简单，支持80种以上的高级算子，支持多种语言，数据源丰富，可部署在多种集群中
（4）容错性高。Spark引进了弹性分布式数据集RDD (Resilient Distributed Dataset) 的抽象，它是分布在一组节点中的只读对象集合，这些集合是弹性的，如果数据集一部分丢失，则可以根据“血统”（即充许基于数据衍生过程）对它们进行重建。另外在RDD计算时可以通过CheckPoint来实现容错，而CheckPoint有两种方式：CheckPoint Data，和Logging The Updates，用户可以控制采用哪种方式来实现容错。
1.11 Spark的适用场景
（1）目前大数据处理场景有以下几个类型：
（2）复杂的批量处理（Batch Data Processing），偏重点在于处理海量数据的能力，至于处理速度可忍受，通常的时间可能是在数十分钟到数小时；
（3）基于历史数据的交互式查询（Interactive Query），通常的时间在数十秒到数十分钟之间
（4）基于实时数据流的数据处理（Streaming Data Processing），通常在数百毫秒到数秒之间
1.12 Spark基本概念
Application =>Spark的应用程序，包含一个Driver program和若干Executor
SparkContext => Spark应用程序的入口，负责调度各个运算资源，协调各个Worker Node上的Executor
Driver Program =>运行Application的main()函数并且创建SparkContext
Executor =>是为Application运行在Worker node上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上。每个Application都会申请各自的Executor来处理任务
Cluster Manager =>在集群上获取资源的外部服务 (例如：Standalone、Mesos、Yarn)
Worker Node =>集群中任何可以运行Application代码的节点，运行一个或多个Executor进程
Task =>运行在Executor上的工作单元
Job => SparkContext提交的具体Action操作，常和Action对应
Stage =>每个Job会被拆分很多组task，每组任务被称为Stage，也称TaskSet
RDD =>是Resilient distributed datasets的简称，中文为弹性分布式数据集;是Spark最核心的模块和类
DAGScheduler =>根据Job构建基于Stage的DAG，并提交Stage给TaskScheduler
TaskScheduler =>将Taskset提交给Worker node集群运行并返回结果
Transformations =>是Spark API的一种类型，Transformation返回值还是一个RDD，所有的Transformation采用的都是懒策略，如果只是将Transformation提交是不会执行计算的
Action =>是Spark API的一种类型，Action返回值不是一个RDD，而是一个scala集合；计算只有在Action被提交的时候计算才被触发。
二、Spark Core简介
2.1 Spark Core之RDD
2.1.1 RDD是什么
RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。
2.1.2 RDD的属性
（1）partitions(分区):一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
（2）partitioner(分区方法):一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。
（3）dependencies(依赖关系):RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。
（4）compute(获取分区迭代列表)：一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
（5）preferedLocations(优先分配节点列表) ：一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。
2.1.3 RDD的创建方式
（1）通过读取文件生成的
由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等
scala> val file = sc.textFile("/spark/hello.txt")
（2）通过并行化的方式创建RDD
由一个已经存在的Scala集合创建。
scala> val array = Array(1,2,3,4,5)
array: Array[Int] = Array(1, 2, 3, 4, 5)
scala> val rdd = sc.parallelize(array)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[27] at parallelize at <console>:26
（3）其他方式
读取数据库等等其他的操作。也可以生成RDD。RDD可以通过其他的RDD转换而来的
2.1.4 RDD算子操作之Transformation
主要做的是就是将一个已有的RDD生成另外一个RDD。Transformation具有lazy特性(延迟加载)。Transformation算子的代码不会真正被执行。只有当我们的程序里面遇到一个action算子的时候，代码才会真正的被执行。这种设计让Spark更加有效率地运行。
常用的Transformation：
2.1.4 RDD算子操作之Action
触发代码的运行，我们一段spark代码里面至少需要有一个action操作。
常用的Action:
2.1.5 RDD依赖关系的本质内幕
由于RDD是粗粒度的操作数据集，每个Transformation操作都会生成一个新的RDD，所以RDD之间就会形成类似流水线的前后依赖关系；RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。如图所示显示了RDD之间的依赖关系。
从图中可知：
窄依赖：是指每个父RDD的一个Partition最多被子RDD的一个Partition所使用，例如map、filter、union等操作都会产生窄依赖；（独生子女）
宽依赖：是指一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖；（超生）
需要特别说明的是对join操作有两种情况：
（1）图中左半部分join：如果两个RDD在进行join操作时，一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join，那么这种类型的join操作就是窄依赖，例如图1中左半部分的join操作(join with inputs co-partitioned)；
（2）图中右半部分join：其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition进行join的转换，这就涉及到了shuffle，因此这种类型的join操作也是宽依赖。
总结：
在这里我们是从父RDD的partition被使用的个数来定义窄依赖和宽依赖，因此可以用一句话概括下：如果父RDD的一个Partition被子RDD的一个Partition所使用就是窄依赖，否则的话就是宽依赖。因为是确定的partition数量的依赖关系，所以RDD之间的依赖关系就是窄依赖；由此我们可以得出一个推论：即窄依赖不仅包含一对一的窄依赖，还包含一对固定个数的窄依赖。
一对固定个数的窄依赖的理解：即子RDD的partition对父RDD依赖的Partition的数量不会随着RDD数据规模的改变而改变；换句话说，无论是有100T的数据量还是1P的数据量，在窄依赖中，子RDD所依赖的父RDD的partition的个数是确定的，而宽依赖是shuffle级别的，数据量越大，那么子RDD所依赖的父RDD的个数就越多，从而子RDD所依赖的父RDD的partition的个数也会变得越来越多。
2.1.6 RDD依赖关系下的数据流视图
在spark中，会根据RDD之间的依赖关系将DAG图（有向无环图）划分为不同的阶段，对于窄依赖，由于partition依赖关系的确定性，partition的转换处理就可以在同一个线程里完成，窄依赖就被spark划分到同一个stage中，而对于宽依赖，只能等父RDD shuffle处理完成后，下一个stage才能开始接下来的计算。
因此spark划分stage的整体思路是：从后往前推，遇到宽依赖就断开，划分为一个stage；遇到窄依赖就将这个RDD加入该stage中。因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。
在spark中，Task的类型分为2种：ShuffleMapTask和ResultTask；
简单来说，DAG的最后一个阶段会为每个结果的partition生成一个ResultTask，即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的！而其余所有阶段都会生成ShuffleMapTask；之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中；也就是说上图中的stage1和stage2相当于mapreduce中的Mapper,而ResultTask所代表的stage3就相当于mapreduce中的reducer。
在之前动手操作了一个wordcount程序，因此可知，Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不过区别在于：Hadoop中的MapReduce天生就是排序的；而reduceByKey只是根据Key进行reduce，但spark除了这两个算子还有其他的算子；因此从这个意义上来说，Spark比Hadoop的计算算子更为丰富。
2.2 Spark Core之Job
spark中对RDD的操作类型分为transformation和action，其中transformation是一种延迟执行的操作，并不会立即执行而是返回一个含有依赖关系的RDD，例如map、filter、sortBy、flatMap等操作，当调用action操作时，** spark会通过DAGScheduler构建一个job的执行拓扑,包括多个stage和task,所有的stage和task构成了这个action触发的job,最后提交到集群中执行。
2.3 Spark Core之Stage
Job是由stage构成的,spark中的stage只有两种,shuffleMapStage和resultStage。Job划分Stage的依据是shuffle与否(即依赖的类型),当DagScheduler进行DAG执行图构建时,遇到一个shuffle dependency会生成一个shuffle map stage,调用链最后一个shuffle reduce端将生成为result stage 也叫 final stage.
handleJobSubmitted->submitStage(finalStage)->递归调用submitStage,如果当前提交的stage没有parent stage则直接提交taskSet,否则将当前stage加入waiting stage列表,每当触发某些事件时(MapStageSubmitted、TaskCompletion..)都会进行一次 waiting stages 的提交。
2.4 Spark Core之Task
task是逻辑的具体执行单元，stage由task构成,当提交stage时,DAGScheduler会根据当前stage的类型序列化出不同类型的task并进行broadcast,如果是shuffleMapStage则序列化出ShuffleMapTask,如果是resultStage则序列化出ResultTask,其中task的数量和当前stage所依赖的RDD的partition的数量是一致的，Task作用的数据单元是partition,即每个task只处理一个partition的数据。
2.5 Spark Core之shuffle
shuffle是影响spark性能的核心所在，和mapreduce中的shuffle概念类似,在spark2.0中shuffleManager只有sortShuffleManager,并且在满足一定条件下可以使用Serialized sorting 即在tungsten中对其进行的优化
2.6 Spark Core之Tungsten
该项目主要是为了让spark计算模型能更好的利用硬件性能，主要包含三部分：
1、内存管理与二进制处理
2、cache-aware
3、代码生成
三、Spark Streaming简介
3.1 Spark Streaming概述
Spark Streaming是Spark核心API的一个扩展，可以实现高吞吐量的、具备容错机制的实时流数据的处理。支持从多种数据源获取数据，包括Kafk、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets，从数据源获取数据之后，可以使用诸如map、reduce、join和window等高级函数进行复杂算法的处理。最后还可以将处理结果存储到文件系统，数据库和现场仪表盘。在“One Stack rule them all”的基础上，还可以使用Spark的其他子框架，如集群学习、图计算等，对流数据进行处理。
Spark Streaming处理的数据流图：
3.2 Spark Streaming架构
SparkStreaming是一个对实时数据流进行高通量、容错处理的流式处理系统，可以对多种数据源（如Kdfka、Flume、Twitter、Zero和TCP 套接字）进行类似Map、Reduce和Join等复杂操作，并将结果保存到外部文件系统、数据库或应用到实时仪表盘。
l计算流程：Spark Streaming是将流式计算分解成一系列短小的批处理作业。这里的批处理引擎是Spark Core，也就是把Spark Streaming的输入数据按照batch size（如1秒）分成一段一段的数据（Discretized Stream），每一段数据都转换成Spark中的RDD（Resilient Distributed Dataset），然后将Spark Streaming中对DStream的Transformation操作变为针对Spark中对RDD的Transformation操作，将RDD经过操作变成中间结果保存在内存中。整个流式计算根据业务的需求可以对中间的结果进行叠加或者存储到外部设备。下图显示了Spark Streaming的整个流程。
l容错性：对于流式计算来说，容错性至关重要。首先我们要明确一下Spark中RDD的容错机制。每一个RDD都是一个不可变的分布式可重算的数据集，其记录着确定性的操作继承关系（lineage），所以只要输入数据是可容错的，那么任意一个RDD的分区（Partition）出错或不可用，都是可以利用原始输入数据通过转换操作而重新算出的。
对于Spark Streaming来说，其RDD的传承关系如下图所示，图中的每一个椭圆形表示一个RDD，椭圆形中的每个圆形代表一个RDD中的一个Partition，图中的每一列的多个RDD表示一个DStream（图中有三个DStream），而每一行最后一个RDD则表示每一个Batch Size所产生的中间结果RDD。我们可以看到图中的每一个RDD都是通过lineage相连接的，由于Spark Streaming输入数据可以来自于磁盘，例如HDFS（多份拷贝）或是来自于网络的数据流（Spark Streaming会将网络输入数据的每一个数据流拷贝两份到其他的机器）都能保证容错性，所以RDD中任意的Partition出错，都可以并行地在其他机器上将缺失的Partition计算出来。这个容错恢复方式比连续计算模型（如Storm）的效率更高。
l实时性：对于实时性的讨论，会牵涉到流式处理框架的应用场景。Spark Streaming将流式计算分解成多个Spark Job，对于每一段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程。对于目前版本的Spark Streaming而言，其最小的Batch Size的选取在0.5~2秒钟之间（Storm目前最小的延迟是100ms左右），所以Spark Streaming能够满足除对实时性要求非常高（如高频实时交易）之外的所有流式准实时计算场景。
l扩展性与吞吐量：Spark目前在EC2上已能够线性扩展到100个节点（每个节点4Core），可以以数秒的延迟处理6GB/s的数据量（60M records/s），其吞吐量也比流行的Storm高2～5倍，图4是Berkeley利用WordCount和Grep两个用例所做的测试，在Grep这个测试中，Spark Streaming中的每个节点的吞吐量是670k records/s，而Storm是115k records/s。
3.3 Spark Streaming之StreamingContext
有两种创建StreamingContext的方式：
appName，是用来在Spark UI上显示的应用名称。master，是一个Spark、Mesos或者Yarn 集群的URL，或者是local[*]。
batch interval可以根据你的应用程序的延迟要求以及可用的集群资源情况来设置。
一个StreamingContext定义之后，必须做以下几件事情：
1、通过创建输入DStream来创建输入数据源。
2、通过对DStream定义transformation和output算子操作，来定义实时计算逻辑。
3、调用StreamingContext的start()方法，来开始实时处理数据。
4、调用StreamingContext的awaitTermination()方法，来等待应用程序的终止。可以使用CTRL+C手动停止，或者就是让它持续不断的运行进行计算。
5、也可以通过调用StreamingContext的stop()方法，来停止应用程序。
StreamingContext需要注意的要点：
1、只要一个StreamingContext启动之后，就不能再往其中添加任何计算逻辑了。比如执行start()方法之后，还给某个DStream执行一个算子。
2、一个StreamingContext停止之后，是肯定不能够重启的。调用stop()之后，不能再调用start()
3、一个JVM同时只能有一个StreamingContext启动。在你的应用程序中，不能创建两个StreamingContext。
4、调用stop()方法时，会同时停止内部的SparkContext，如果不希望如此，还希望后面继续使用SparkContext创建其他类型的Context，比如SQLContext，那么就用stop(false)。
5、一个SparkContext可以创建多个StreamingContext，只要上一个先用stop(false)停止，再创建下一个即可。
3.4 Spark Streaming之DStream
DStream（Discretized Stream）作为Spark Streaming的基础抽象，它代表持续性的数据流。这些数据流既可以通过外部输入源赖获取，也可以通过现有的Dstream的transformation操作来获得。在内部实现上，DStream由一组时间序列上连续的RDD来表示。每个RDD都包含了自己特定时间间隔内的数据流。如图7-3所示。
对DStream中数据的各种操作也是映射到内部的RDD上来进行的，如图7-4所示，对Dtream的操作可以通过RDD的transformation生成新的DStream。这里的执行引擎是Spark。
3.5 Spark Streaming之Input DStream and Receivers
如同Storm的Spout，Spark Streaming需要指明数据源。如上例所示的socketTextStream，Spark Streaming以socket连接作为数据源读取数据。当然Spark Streaming支持多种不同的数据源，包括Kafka、 Flume、HDFS/S3、Kinesis和Twitter等数据源；
3.6 Spark Streaming之Transformations
https://www.cnblogs.com/yhl-yh/p/7651558.html
3.7 Spark Streaming之Operations
https://www.cnblogs.com/yhl-yh/p/7651558.html
3.8 Spark Streaming之Join Operations
（1）DStream对象之间的Join
这种join一般应用于窗口函数形成的DStream对象之间，具体可以参考第一部分中的join操作，除了简单的join之外，还有leftOuterJoin, rightOuterJoin和fullOuterJoin。
（2）DStream和dataset之间的join
这一种join，可以参考前面transform操作中的示例。
3.9 Spark Streaming之Output Operations
https://www.cnblogs.com/yhl-yh/p/7651558.html
四、Spark SQL简介
4.1 Spark SQL是什么
是Spark用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame,并且作为分布式SQL查询引擎的作用
4.2 Spark SQL基础数据模型
DataFrame是由“命名列”（类似关系表的字段定义）所组织起来的一个分布式数据集合。你可以把它看成是一个关系型数据库的表。
DataFrame可以通过多种来源创建：结构化数据文件，hive的表，外部数据库，或者RDDs
4.3 Spark SQL运行流程
1、将SQL语句通过词法和语法解析生成未绑定的逻辑执行计划（Unresolved LogicalPlan），包含Unresolved Relation、Unresolved Function和Unresolved Attribute，然后在后续步骤中使用不同的Rule应用到该逻辑计划上
2、Analyzer使用Analysis Rules，配合元数据（如SessionCatalog 或是 Hive Metastore等）完善未绑定的逻辑计划的属性而转换成绑定的逻辑计划。具体流程是县实例化一个Simple Analyzer，然后遍历预定义好的Batch，通过父类Rule Executor的执行方法运行Batch里的Rules，每个Rule会对未绑定的逻辑计划进行处理，有些可以通过一次解析处理，有些需要多次迭代，迭代直到达到FixedPoint次数或前后两次的树结构没变化才停止操作。
3、Optimizer使用Optimization Rules，将绑定的逻辑计划进行合并、列裁剪和过滤器下推等优化工作后生成优化的逻辑计划
4、Planner使用Planning Strategies，对优化的逻辑计划进行转换（Transform）生成可以执行的物理计划。根据过去的性能统计数据，选择最佳的物理执行计划CostModel，最后生成可以执行的物理执行计划树，得到SparkPlan
5、在最终真正执行物理执行计划之前，还要进行preparations规则处理，最后调用SparkPlan的execute执行计算RDD。
4.4 Spark SQL的诞生
但是，随着Spark的发展，对于野心勃勃的Spark团队来说，Shark对于Hive的太多依赖（如采用Hive的语法解析器、查询优化器等等），制约了Spark的One Stack Rule Them All的既定方针，制约了Spark各个组件的相互集成，所以提出了SparkSQL项目。SparkSQL抛弃原有Shark的代码，汲取了Shark的一些优点，如内存列存储（In-Memory Columnar Storage）、Hive兼容性等，重新开发了SparkSQL代码；由于摆脱了对Hive的依赖性，SparkSQL无论在数据兼容、性能优化、组件扩展方面都得到了极大的方便，真可谓“退一步，海阔天空”。
4.5 Spark SQL和Hive On Spark的区别
很多人误以为Spark SQL和 Hive On Spark是一个东西，其实不然，这里对二者做一个简单的介绍：Spark SQL的前身是Shark, Shark 刚开始也是使用了hive里面一些东西的,但随着后来的发展，Shark始终不能很好的支持Hive里面的一些功能，这是因为在版本的更迭和Hive的多年发展，有些东西并不能始终支持，Shark 终止后，SparkSQL作为Spark生态的一员继续发展，而不再受限于Hive，只是兼容Hive；而Hive On Spark 是Hive 想要将底层的执行引擎在Spark上做很好的支持，并且想要对Hive的底层执行引擎做更多的支持.
4.6 Spark SQL架构
SparkSQL架构分为前，后，中，三个部分，其中中间部分的Catalyst optimizer是架构的核心，也是SparkSQL 优化的关键所在。
4.7 Spark SQL出现原因
（1）使用SQL来进行大数据处理，这本身就为传统的RDBMS人员降低了要求，因为这不需要使用人员掌握像mapreduce的编程方法。
（2）SparkSQL 可以支持SQL API，DataFrame和Dataset API多种API，这意味着开发人员可以采用DataFrame和Dataset API 进行编程，这和采用Spark core 的RDD API 进行编程有很大的不同，首先，并不像使用RDD进行编程时，开发人员在采用不同的编程语言和不同的方式开发应用程序时，其应用程序的性能千差万别，但如果使用DataFrame和Dataset进行开发时，资深开发人员和初级开发人员开发的程序性能差异很小，这是因为SparkSQL 使用Catalyst optimizer 对执行计划做了很好的优化。
（3）其次，SparkSQL既然对底层使用了很好的优化方式，这就让开发人员在不熟悉底层优化方式时也可以进行很好的程序开发，降低了开发人员对程序开发的门槛。
（4）最后，SparkSQL 对外部数据源(External Datasource)做了很好的支持，像对关系型数据库，如mysql,可以使用外部数据源的方式使用SparkSQL对数据库里面的数据直接进行处理，而不需要像以前使用sqoop导入导出。
4.8 Spark SQL如何使用
首先，利用sqlContext从外部数据源加载数据为DataFrame
然后，利用DataFrame上丰富的api进行查询、转换
最后，将结果进行展现或存储为各种外部数据形式
4.9 Spark SQL愿景
1.写更少的代码
2.读更少的数据
3.把优化的工作交由底层的优化器运行
4.14 SparkContext运行原理分析
前面我们队Spark SQL运行架构进行分析，知道从输入SQL语句到生成Dataset分为5个步骤，但实际运行过程中在输入SQL语句之前，Spark还有加载SessionCatalog步骤。
4.14.1 使用SessionCatalog保存元数据
在解析SQL语句之前需要初始化SQLContext，它定义了Spark SQL执行的上下文，并把元数据保存在SessionCatalog中，这些元数据包括表名称、表字段名称和字段类型等。
SessionCatalog中保存的是表名和逻辑执行计划对应的哈希列表，这些数据将在解析未绑定的逻辑计划上使用。（SessionCatalog中的表名对应的逻辑执行计划是什么？是这个Dataset对应的逻辑执行计划）
4.14.2 使用Antlr生成未绑定的逻辑计划
Spark 2.0版本起使用Antlr进行词法和语法解析。使用Antlr生成未绑定的逻辑计划分为两个阶段：第一阶段的过程为词法分析（Lexical Analysis），负责将符号（Token）分组成符号类(Token class or Token type)，第二阶段就是真正的Parser，默认Antlr会构建出一颗分析树（Parser Tree）或者叫语法树（Syntax
Tree）。
SQLContext类中定义了SQL的解析方法parseSql。具体的SQL解析在AbastrctSqlParser抽象类中的parse方法进行，解析完毕后生成语法树，语法树会根据系统初始化的AstBuilder解析生成表达式、逻辑计划或表标识对象。
在AbstractSqlParse的parse方法中，先实例化词法解析器SqlBaseLexer和语法解析器SqlBaseParser，然后尝试用Antlr较快的解析模式SLL，如果解析失败，则会再尝试使用普通解析模型LL，解析完毕后返回解析结果。
4.14.3 使用Analyzer绑定逻辑执行计划
该阶段Analyzer使用Analysis Rules，结合SessionCatalog元数据，对未绑定的逻辑计划进行解析，生成了已绑定的逻辑计划。在该处理过程中，先实例化一个Analyzer，在Analyzer中定义了FiexedPoint和Seq[Batch]两个变量，其中FixedPoint为迭代次数的上线，而Seq[Batch]为所定义需要执行批处理的序列，每个批处理由一系列Rule和策略所组成，策略一般分为Once和FixedPoint（可理解为迭代次数）
4.14.4 使用Optimizer优化逻辑计划
Optimizer的实现和处理方式同Analyzer类似，在该类中定义一系列Rule并同样继承于RuleExecutor。利用这些Rule对逻辑计划和Expression进行迭代处理，从而达到对树的节点进行合并和优化。其中主要的优化策略总结起来是合并、列裁剪和过滤器下推等几大类。
Optimizer的优化策略不仅对已绑定的逻辑计划进行优化，而且对逻辑计划中的Expression也进行了优化，其原理就是遍历树，然后应用优化Rule，但是注意一点，对逻辑计划处理时先序遍历，而对Expression的处理是后续遍历（根据树节点类型来判断是逻辑计划还是Expression吗？）
4.14.5 使用SparkPlanner生成可执行物理计划
SparkPlanner使用Planning Strategies对优化的逻辑计划进行转换(Transform)，生成可以执行的物理计划。在QueryExecution类代码中，调用SparkPlanner.plan方法对优化的逻辑计划进行处理，而SparkPlanner并未定义plan方法，实际是调用SparkPlanner的祖父类QueyrPlanner的plan方法，然后会返回一个Iterator[Physicalplan]。
SparkPlanner继承于SparkStrategies，而SparkStrategies继承了QueryPlanner。其中SparkStrategies包含了一系列特定的Strategies，这些Strategies是继承自QueryPlanner中定义的GenericStrategy。在SparkPlanner通过改写祖父类QueryPlanner中strategies策略变量，在该变量中定义了转换成物理计划所执行的策略。
4.14.6 使用QueryExecution执行物理计划
在该步骤中先调用SparkPlanner.preparations对物理计划进行准备工作，规范返回数据行的格式等，然后调用SparkPlan.execute执行物理计划，从数据库中查询数据并生成RDD。
SparkPlan的preparations其实是一个RuleExecutor[SparkPlan]，它会调用RuleExecutor的execut方法对前面生成的物理计划应用Rule进行匹配，最终生成一个SparkPlan。
SparkPlan继承于QueryPlan，SparkPlan中定义了SQL语句执行的execute方法，执行完execute方法返回的是一个RDD，之后可以运行Spark作业对该RDD进行操作。
4.15 Spark SQL小结
1）Spark SQL的应用并不局限于SQL；
2）访问hive、json、parquet等文件的数据；
3）SQL只是Spark SQL的一个功能而已；
4）Spark SQL提供了SQL的api、DataFrame和Dataset的API；
五、SparkMLlib简介
5.1 SparkMLlib是什么
MLlib是Spark的机器学习（Machine Learning）库，旨在简化机器学习的工程实践工作，并方便扩展到更大规模。MLlib由一些通用的学习算法和工具组成，包括分类、回归、聚类、协同过滤、降维等，同时还包括底层的优化原语和高层的管道API。具体来说，其主要包括以下几方面的内容：
（1）算法工具：常用的学习算法，如分类、回归、聚类和协同过滤；
（2）特征化公交：特征提取、转化、降维，和选择公交；
（3）管道(Pipeline)：用于构建、评估和调整机器学习管道的工具;
（4）持久性：保存和加载算法，模型和管道;
（5）实用工具：线性代数，统计，数据处理等工具。
5.2 SparkMLlib调优
MLlib是spark的可以扩展的机器学习库，由以下部分组成：通用的学习算法和工具类，包括分类，回归，聚类，协同过滤，降维，当然也包括调优的部分
Data
typesBasic
statistics(基本统计)
summary statistics概括统计correlations 相关性stratified sampling 分层取样hypothesis testing 假设检验random data generation 随机数生成Classification
and regression(分类一般针对离散型数据而言的，回归是针对连续型数据的。本质上是一样的)
linear
models (SVMs, logistic regression, linear regression)线性模型（支持向量机，逻辑回归，线性回归）naive
Bayes贝叶斯算法decision
trees决策树ensembles
of trees(Random Forests and Gradient-Boosted Trees)多种树（随机森林和梯度增强树）Collaborative
filtering协同过滤
alternating least squares (ALS)(交替最小二乘法(ALS) )Clustering
聚类
k-means k均值算法Dimensionality
reduction (降维)
singular value decomposition (SVD)奇异值分解principal component analysis (PCA)主成分分析Feature
extraction and transformation特征提取和转化Optimization (developer)优化部分
stochastic gradient descent随机梯度下降limited-memory BFGS (L-BFGS) 短时记忆的BFGS (拟牛顿法中的一种,解决非线性问题)
六、SparkGraphX简介
6.1 Spark GraphX是什么
Spark GraphX是一个分布式图处理框架，它是基于Spark平台提供对图计算和图挖掘简洁易用的而丰富的接口，极大的方便了对分布式图处理的需求。
6.2 Spark GraphX的框架
设计GraphX时，点分割和GAS都已成熟，在设计和编码中针对它们进行了优化，并在功能和性能之间寻找最佳的平衡点。如同Spark本身，每个子模块都有一个核心抽象。GraphX的核心抽象是Resilient Distributed Property Graph，一种点和边都带属性的有向多重图。它扩展了Spark RDD的抽象，有Table和Graph两种视图，而只需要一份物理存储。两种视图都有自己独有的操作符，从而获得了灵活操作和执行效率。
如同Spark，GraphX的代码非常简洁。GraphX的核心代码只有3千多行，而在此之上实现的Pregel模式，只要短短的20多行。GraphX的代码结构整体下图所示，其中大部分的实现，
都是围绕Partition的优化进行的。这在某种程度上说明了点分割的存储和相应的计算优化，的确是图计算框架的重点和难点。
6.3 Spark GraphX实现分析
如同Spark本身，每个子模块都有一个核心抽象。GraphX的核心抽象是Resilient Distributed Property Graph，一种点和边都带属性的有向多重图。它扩展了Spark RDD的抽象，有Table和Graph两种视图，而只需要一份物理存储。两种视图都有自己独有的操作符，从而获得了灵活操作和执行效率。
GraphX的底层设计有以下几个关键点。
对Graph视图的所有操作，最终都会转换成其关联的Table视图的RDD操作来完成。这样对一个图的计算，最终在逻辑上，等价于一系列RDD的转换过程。因此，Graph最终具备了RDD的3个关键特性：Immutable、Distributed和Fault-Tolerant，其中最关键的是Immutable（不变性）。逻辑上，所有图的转换和操作都产生了一个新图；物理上，GraphX会有一定程度的不变顶点和边的复用优化，对用户透明。
两种视图底层共用的物理数据，由RDD[Vertex-Partition]和RDD[EdgePartition]这两个RDD组成。点和边实际都不是以表Collection[tuple]的形式存储的，而是由VertexPartition/EdgePartition在内部存储一个带索引结构的分片数据块，以加速不同视图下的遍历速度。不变的索引结构在RDD转换过程中是共用的，降低了计算和存储开销。
图的分布式存储采用点分割模式，而且使用partitionBy方法，由用户指定不同的划分策略（PartitionStrategy）。划分策略会将边分配到各个EdgePartition，顶点Master分配到各个VertexPartition，EdgePartition也会缓存本地边关联点的Ghost副本。划分策略的不同会影响到所需要缓存的Ghost副本数量，以及每个EdgePartition分配的边的均衡程度，需要根据图的结构特征选取最佳策略。目前有EdgePartition2d、EdgePartition1d、RandomVertexCut和CanonicalRandomVertexCut这四种策略。
