Spark知识体系完整解读

 

　　做者：杨思义，2014年6月至今工做于北京亚信智慧数据科技有限公司 BDX大数据事业部，从2014年9月开始从事项目spark相关应用开发。

　　来源：数盟

　　Spark简介

　　Spark是整个BDAS的核心组件，是一个大数据分布式编程框架，不只实现了MapReduce的算子map 函数和reduce函数及计算模型，还提供更为丰富的算子，如filter、join、groupByKey等。是一个用来实现快速而同用的集群计算的平台。

　　Spark将分布式数据抽象为弹性分布式数据集（RDD），实现了应用任务调度、RPC、序列化和压缩，并为运行在其上的上层组件提供API。其底层采用Scala这种函数式语言书写而成，而且所提供的API深度借鉴Scala函数式的编程思想，提供与Scala相似的编程接口

　　Sparkon Yarn

　　

　　从用户提交做业到做业运行结束整个运行期间的过程分析。

　　1、客户端进行操做

1. 根据yarnConf来初始化yarnClient，并启动yarnClient

2. 建立客户端Application，并获取Application的ID，进一步判断集群中的资源是否知足executor和ApplicationMaster申请的资源，若是不知足则抛出IllegalArgumentException；

3. 设置资源、环境变量：其中包括了设置Application的Staging目录、准备本地资源（jar文件、log4j.properties）、设置Application其中的环境变量、建立Container启动的Context等；

4. 设置Application提交的Context，包括设置应用的名字、队列、AM的申请的Container、标记该做业的类型为Spark；

5. 申请Memory，并最终经过yarnClient.submitApplication向ResourceManager提交该Application。

　　看成业提交到YARN上以后，客户端就没事了，甚至在终端关掉那个进程也没事，由于整个做业运行在YARN集群上进行，运行的结果将会保存到HDFS或者日志中。

　　2、提交到YARN集群，YARN操做

1. 运行ApplicationMaster的run方法；

2. 设置好相关的环境变量。

3. 建立amClient，并启动；

4. 在Spark UI启动以前设置Spark UI的AmIpFilter；

5. 在startUserClass函数专门启动了一个线程（名称为Driver的线程）来启动用户提交的Application，也就是启动了Driver。在Driver中将会初始化SparkContext；

6. 等待SparkContext初始化完成，最多等待spark.yarn.applicationMaster.waitTries次数（默认为10），若是等待了的次数超过了配置的，程序将会退出；不然用SparkContext初始化yarnAllocator；

7. 当SparkContext、Driver初始化完成的时候，经过amClient向ResourceManager注册ApplicationMaster

8. 分配并启动Executeors。在启动Executeors以前，先要经过yarnAllocator获取到numExecutors个Container，而后在Container中启动Executeors。

   　　那么这个Application将失败，将Application Status标明为FAILED，并将关闭SparkContext。其实，启动Executeors是经过ExecutorRunnable实现的，而ExecutorRunnable内部是启动CoarseGrainedExecutorBackend的。

9. 最后，Task将在CoarseGrainedExecutorBackend里面运行，而后运行情况会经过Akka通知CoarseGrainedScheduler，直到做业运行完成。

　　Spark节点的概念

　　1、Spark驱动器是执行程序中的main()方法的进程。它执行用户编写的用来建立SparkContext(初始化)、建立RDD，以及运行RDD的转化操做和行动操做的代码。

　　驱动器节点driver的职责：

1. 把用户程序转为任务task(driver)

   　　Spark驱动器程序负责把用户程序转化为多个物理执行单元，这些单元也被称之为任务task(详解见备注)

2. 为执行器节点调度任务(executor)

   　　有了物理计划以后，Spark驱动器在各个执行器节点进程间协调任务的调度。Spark驱动器程序会根据当前的执行器节点，把全部任务基于数据所在位置分配给合适的执行器进程。当执行任务时，执行器进程会把缓存的数据存储起来，而驱动器进程一样会跟踪这些缓存数据的位置，并利用这些位置信息来调度之后的任务，以尽可能减小数据的网络传输。（就是所谓的移动计算，而不移动数据)。

　　2、执行器节点

　　做用：

1. 负责运行组成Spark应用的任务，并将结果返回给驱动器进程；

2. 经过自身的块管理器(blockManager)为用户程序中要求缓存的RDD提供内存式存储。RDD是直接缓存在执行器进程内的，所以任务能够在运行时充分利用缓存数据加快运算。

　　驱动器的职责：

　　全部的Spark程序都遵循一样的结构：程序从输入数据建立一系列RDD，再使用转化操做派生成新的RDD，最后使用行动操做手机或存储结果RDD，Spark程序实际上是隐式地建立出了一个由操做组成的逻辑上的有向无环图DAG。当驱动器程序执行时，它会把这个逻辑图转为物理执行计划。

　　这样 Spark就把逻辑计划转为一系列步骤(stage)，而每一个步骤又由多个任务组成。这些任务会被打包送到集群中。

　　Spark初始化

1. 每一个Spark应用都由一个驱动器程序来发起集群上的各类并行操做。驱动器程序包含应用的main函数，而且定义了集群上的分布式数据集，以及对该分布式数据集应用了相关操做。

2. 驱动器程序经过一个SparkContext对象来访问spark,这个对象表明对计算集群的一个链接。（好比在sparkshell启动时已经自动建立了一个SparkContext对象，是一个叫作SC的变量。(下图，查看变量sc)

   　　

3. 一旦建立了sparkContext，就能够用它来建立RDD。好比调用sc.textFile()来建立一个表明文本中各行文本的RDD。（好比vallinesRDD = sc.textFile(“yangsy.text”),val spark = linesRDD.filter(line=>line.contains(“spark”),spark.count()）

   　　执行这些操做，驱动器程序通常要管理多个执行器,就是咱们所说的executor节点。

4. 在初始化SparkContext的同时，加载sparkConf对象来加载集群的配置，从而建立sparkContext对象。

   　　从源码中能够看到，在启动thriftserver时，调用了spark- daemon.sh文件，该文件源码如左图，加载spark_home下的conf中的文件。

   　　

   　　（在执行后台代码时，须要首先建立conf对象，加载相应参数， val sparkConf = newSparkConf().setMaster("local").setAppName("cocapp").set("spark.executor.memory","1g"), val sc: SparkContext = new SparkContext(sparkConf))

　　RDD工做原理：

　　RDD(Resilient DistributedDatasets)[1] ,弹性分布式数据集，是分布式内存的一个抽象概念，RDD提供了一种高度受限的共享内存模型，即RDD是只读的记录分区的集合，只能经过在其余RDD执行肯定的转换操做（如map、join和group by）而建立，然而这些限制使得实现容错的开销很低。对开发者而言，RDD能够看做是Spark的一个对象，它自己运行于内存中，如读文件是一个RDD，对文件计算是一个RDD，结果集也是一个RDD ，不一样的分片、数据之间的依赖、key-value类型的map数据均可以看作RDD。

　　主要分为三部分：建立RDD对象，DAG调度器建立执行计划，Task调度器分配任务并调度Worker开始运行。

　　SparkContext(RDD相关操做)→经过(提交做业)→(遍历RDD拆分stage→生成做业)DAGScheduler→经过（提交任务集）→任务调度管理(TaskScheduler)→经过（按照资源获取任务)→任务调度管理(TaskSetManager)

　　Transformation返回值仍是一个RDD。它使用了链式调用的设计模式，对一个RDD进行计算后，变换成另一个RDD，而后这个RDD又能够进行另一次转换。这个过程是分布式的。

　　Action返回值不是一个RDD。它要么是一个Scala的普通集合，要么是一个值，要么是空，最终或返回到Driver程序，或把RDD写入到文件系统中

　　转换(Transformations)(如：map, filter, groupBy, join等)，Transformations操做是Lazy的，也就是说从一个RDD转换生成另外一个RDD的操做不是立刻执行，Spark在遇到Transformations操做时只会记录须要这样的操做，并不会去执行，须要等到有Actions操做的时候才会真正启动计算过程进行计算。

　　操做(Actions)(如：count, collect, save等)，Actions操做会返回结果或把RDD数据写到存储系统中。Actions是触发Spark启动计算的动因。

　　它们本质区别是：Transformation返回值仍是一个RDD。它使用了链式调用的设计模式，对一个RDD进行计算后，变换成另一个RDD，而后这个RDD又能够进行另一次转换。这个过程是分布式的。Action返回值不是一个RDD。它要么是一个Scala的普通集合，要么是一个值，要么是空，最终或返回到Driver程序，或把RDD写入到文件系统中。关于这两个动做，在Spark开发指南中会有就进一步的详细介绍，它们是基于Spark开发的核心。

　　RDD基础

1. Spark中的RDD就是一个不可变的分布式对象集合。每一个RDD都被分为多个分区，这些分区运行在集群的不一样节点上。建立RDD的方法有两种：一种是读取一个外部数据集；一种是在群东程序里分发驱动器程序中的对象集合，不如刚才的示例，读取文本文件做为一个字符串的RDD的示例。

2. 建立出来后，RDD支持两种类型的操做:转化操做和行动操做

   　　转化操做会由一个RDD生成一个新的RDD。（好比刚才的根据谓词筛选）

   　　行动操做会对RDD计算出一个结果，并把结果返回到驱动器程序中，或把结果存储到外部存储系统（好比HDFS）中。好比first()操做就是一个行动操做，会返回RDD的第一个元素。

   　　注：转化操做与行动操做的区别在于Spark计算RDD的方式不一样。虽然你能够在任什么时候候定义一个新的RDD，但Spark只会惰性计算这些RDD。它们只有第一个在一个行动操做中用到时，才会真正的计算。之因此这样设计，是由于好比刚才调用sc.textFile(...)时就把文件中的全部行都读取并存储起来，就会消耗不少存储空间，而咱们立刻又要筛选掉其中的不少数据。

   　　这里还须要注意的一点是，spark会在你每次对它们进行行动操做时从新计算。若是想在多个行动操做中重用同一个RDD，那么可使用RDD.persist()或RDD.collect()让Spark把这个RDD缓存下来。（能够是内存，也能够是磁盘)

3. Spark会使用谱系图来记录这些不一样RDD之间的依赖关系，Spark须要用这些信息来按需计算每一个RDD，也能够依靠谱系图在持久化的RDD丢失部分数据时用来恢复所丢失的数据。(以下图，过滤errorsRDD与warningsRDD,最终调用union()函数)

   　　

　　RDD计算方式

　　

　　RDD的宽窄依赖

　　

　　窄依赖 (narrowdependencies) 和宽依赖 (widedependencies) 。窄依赖是指 父 RDD 的每一个分区都只被子 RDD 的一个分区所使用 。相应的，那么宽依赖就是指父 RDD 的分区被多个子 RDD 的分区所依赖。例如， map 就是一种窄依赖，而 join 则会致使宽依赖

　　这种划分有两个用处。首先，窄依赖支持在一个结点上管道化执行。例如基于一对一的关系，能够在 filter 以后执行 map 。其次，窄依赖支持更高效的故障还原。由于对于窄依赖，只有丢失的父 RDD 的分区须要从新计算。而对于宽依赖，一个结点的故障可能致使来自全部父 RDD 的分区丢失，所以就须要彻底从新执行。所以对于宽依赖，Spark 会在持有各个父分区的结点上，将中间数据持久化来简化故障还原，就像 MapReduce 会持久化 map 的输出同样。

　　SparkExample

　　

　　步骤 1 ：建立 RDD 。上面的例子除去最后一个 collect 是个动做，不会建立 RDD 以外，前面四个转换都会建立出新的 RDD 。所以第一步就是建立好全部 RDD( 内部的五项信息 ) 。

　　步骤 2 ：建立执行计划。Spark 会尽量地管道化，并基因而否要从新组织数据来划分 阶段 (stage) ，例如本例中的 groupBy() 转换就会将整个执行计划划分红两阶段执行。最终会产生一个 DAG(directedacyclic graph ，有向无环图 ) 做为逻辑执行计划。

　　步骤 3 ：调度任务。 将各阶段划分红不一样的 任务 (task) ，每一个任务都是数据和计算的合体。在进行下一阶段前，当前阶段的全部任务都要执行完成。由于下一阶段的第一个转换必定是从新组织数据的，因此必须等当前阶段全部结果数据都计算出来了才能继续。

　　假设本例中的 hdfs://names 下有四个文件块，那么 HadoopRDD 中 partitions 就会有四个分区对应这四个块数据，同时 preferedLocations 会指明这四个块的最佳位置。如今，就能够建立出四个任务，并调度到合适的集群结点上。

　　Spark数据分区

1. Spark的特性是对数据集在节点间的分区进行控制。在分布式系统中，通信的代价是巨大的，控制数据分布以得到最少的网络传输能够极大地提高总体性能。Spark程序能够经过控制RDD分区方式来减小通信的开销。

2. Spark中全部的键值对RDD均可以进行分区。确保同一组的键出如今同一个节点上。好比，使用哈希分区将一个RDD分红了100个分区，此时键的哈希值对100取模的结果相同的记录会被放在一个节点上。

   　　（可以使用partitionBy(newHashPartitioner(100)).persist()来构造100个分区)

3. Spark中的许多操做都引入了将数据根据键跨界点进行混洗的过程。(好比：join(),leftOuterJoin(),groupByKey(),reducebyKey()等)对于像reduceByKey()这样只做用于单个RDD的操做，运行在未分区的RDD上的时候会致使每一个键的全部对应值都在每台机器上进行本地计算。

　　SparkSQL的shuffle过程

　　

　　Spark SQL的核心是把已有的RDD，带上Schema信息，而后注册成相似sql里的”Table”，对其进行sql查询。这里面主要分两部分，一是生成SchemaRD，二是执行查询。

　　若是是spark-hive项目，那么读取metadata信息做为Schema、读取hdfs上数据的过程交给Hive完成，而后根据这俩部分生成SchemaRDD，在HiveContext下进行hql()查询。

　　SparkSQL结构化数据

1. 首先说一下ApacheHive，Hive能够在HDFS内或者在其余存储系统上存储多种格式的表。SparkSQL能够读取Hive支持的任何表。要把Spark SQL链接已有的hive上，须要提供Hive的配置文件。hive-site.xml文件复制到spark的conf文件夹下。再建立出HiveContext对象(sparksql的入口)，而后就可使用HQL来对表进行查询，并以由行足证的RDD的形式拿到返回的数据。

2. 建立Hivecontext并查询数据

   　　importorg.apache.spark.sql.hive.HiveContext

   　　valhiveCtx = new org.apache.spark.sql.hive.HiveContext(sc)

   　　valrows = hiveCtx.sql(“SELECT name,age FROM users”)

   　　valfitstRow – rows.first()

   　　println(fitstRow.getSgtring(0)) //字段0是name字段

3. 经过jdbc链接外部数据源更新与加载

   　　Class.forName("com.mysql.jdbc.Driver")

   　　val conn =DriverManager.getConnection(mySQLUrl)

   　　val stat1 =conn.createStatement()

   　　stat1.execute("UPDATE CI_LABEL_INFO set DATA_STATUS_ID = 2 , DATA_DATE ='" + dataDate +"' where LABEL_ID in ("+allCreatedLabels.mkString(",")+")")

   　　stat1.close()

   　　//加载外部数据源数据到内存

   　　valDIM_COC_INDEX_MODEL_TABLE_CONF =sqlContext.jdbc(mySQLUrl,"DIM_COC_INDEX_MODEL_TABLE_CONF").cache()

   　　val targets =DIM_COC_INDEX_MODEL_TABLE_CONF.filter("TABLE_DATA_CYCLE ="+TABLE_DATA_CYCLE).collect

　　SparkSQL解析

　　

　　首先说下传统数据库的解析，传统数据库的解析过程是按Rusult、Data Source、Operation的次序来解析的。传统数据库先将读入的SQL语句进行解析，分辨出SQL语句中哪些词是关键字（如select,from,where)，哪些是表达式，哪些是Projection，哪些是Data Source等等。进一步判断SQL语句是否规范，不规范就报错，规范则按照下一步过程绑定（Bind)。过程绑定是将SQL语句和数据库的数据字典(列,表,视图等）进行绑定，若是相关的Projection、Data Source等都存在，就表示这个SQL语句是能够执行的。在执行过程当中，有时候甚至不须要读取物理表就能够返回结果，好比从新运行刚运行过的SQL语句，直接从数据库的缓冲池中获取返回结果。在数据库解析的过程当中SQL语句时，将会把SQL语句转化成一个树形结构来进行处理，会造成一个或含有多个节点(TreeNode)的Tree,而后再后续的处理政对该Tree进行一系列的操做。

　　Spark SQL对SQL语句的处理和关系数据库对SQL语句的解析采用了相似的方法，首先会将SQL语句进行解析，而后造成一个Tree，后续如绑定、优化等处理过程都是对Tree的操做，而操做方法是采用Rule,经过模式匹配，对不一样类型的节点采用不一样的操做。SparkSQL有两个分支，sqlContext和hiveContext。sqlContext如今只支持SQL语法解析器（Catalyst)，hiveContext支持SQL语法和HiveContext语法解析器。

　　版权声明：

　　转载文章均来自公开网络，仅供学习使用，不会用于任何商业用途，若是出处有误或侵犯到原做者权益，请与咱们联系删除或受权事宜，联系邮箱：holly0801@163.com。转载大数据公众号文章请注明原文连接和做者，不然产生的任何版权纠纷与大数据无关。