Spark的核心RDD
Resilient Distributed Datasets(弹性分布式数据集)
Spark运行原理与RDD理论
Spark与MapReduce对比,MapReduce的计算和迭代是基于磁盘的,而Spark的迭代和计算是尽量基于内存,只有在内存空间不能容纳计算结果时才将溢出的部分数据缓冲到磁盘存储,因此Spark是将内存与磁盘结合起来使用的一种架构,它既可以适应超大型的批量离线数据集处理(因为它可以基于磁盘),也可以适应基于实时的流数据分析计算(因为它可以基于内存迭代)。
Spark是分布式的基于内存的迭代式计算框架,当然它也可以基于磁盘做迭代计算,根据官方说法,基于磁盘的迭代计算也会比基于磁盘的MapReduce计算快10多倍,而基于内存的迭代计算则比基于内存的MapReduce计算快100倍以上;同时由于Spark是迭代式的计算,因此它天生擅长于将多步骤的Job通过应用层面向过程的流水化操作将其转换成底层的多个Job串联操作,这在第一代计算框架MapReduce中需要我们程序员手动分解多个步骤的操作到各个Job作业中去,于此我们如果基于MR来编写Job应用会耗费大量的代码;虽然Hive可以通过HQL来简化MR的编写过程,但是毕竟Hive的简化是很有限度的,而且Hive数据仓库的HQL查询仍然是基于磁盘的(因为底层仍然是执行MR代码)。
Spark的迭代计算过程中,从一个节点计算完成之后的数据到下一个节点的传送过程需要历经一个shuffle过程,如果一个应用需要多个Job步骤来支撑则自然会涉及到多个节点传送数据的多次shuffle过程。
Spark的主要功能模块
交互式查询:基于Spark-Shell的交互式查询和基于Spark-Sql的交互式查询
流式数据处理:实时计算分析
批量数据处理:离线分析计算
Spark客户端Driver—>SparkContext
RDD(Resilient Distributed Datasets(弹性分布式数据集))
RDD是Spark的运行时(Spark Runtime)内核实现,内核涵盖两个部分:
A、数据源：
Spark处理的数据来源可以是HDFS、Hive、HBase、S3(亚马逊云储存平台)、MySql、Oracle等数据源;处理之后的数据也可以被输出到HDFS、Hive、HBase、S3、MySql、Oracle及其当前的控制台终端
B、调度器：Yarn、Mesos、AWS
RDD被称为弹性分布式数据集,它具有如下特征:
A、弹性与伸缩性:
1)、自动进行磁盘数据迭代和内存数据迭代两种操作模式的切换,数据优先在内存中进行迭代,当内存容量有限时再将溢出部分数据缓冲到磁盘进行迭代
2)、作业链步骤的高度容错机制
当一个应用存在多个操作步骤(Stage:阶段)时,如果在作业执行过程中某个步骤出错,则只需要从出错步骤的前一个步骤后重新执行后面的步骤来恢复;而无需从整个应用的第一个步骤开始恢复
3)、作业步骤如果出错或失败则会自动进行特定次数(默认为3次)的重试(该重试包括步骤本身的重试和基于步骤操作底层节点Task计算失败的重试(默认4次))
4)、重新提交或重试时仅提交计算失败的那些数据分片来重新计算,这可以从细粒度上控制重新计算的数据量
注:Spark中对操作步骤(操作阶段)Stage的定义:
1)、如果数据涉及到从一个节点转移并重新分配数据到另一个节点则表示从一个Stage过渡到另一个Stage;从MapReduce原理的角度来理解就是当数据发生reduce、shuffle、sort等操作时都将涉及到数据在节点之间转移;而从SQL概念的角度来理解就是当发生group by、distinct、order by、join、union等操作时都将涉及到数据在节点之间转移;
2)、如果数据操作并没有导致数据在节点之间发生转移(即在同一个节点上进行的本地操作)则表示为Spark概念中的一个Stage内执行的操作,一个Stage内执行的操作相当于MapReduce概念中的map任务操作,这种操作执行的并发度很高(因为这种操作充分利用了集群节点的并发计算能力),而对于SQL概念而言则相当于where和having操作。
application>job>stage>task(jvm)
在以下情况下RDD会缓存操作步骤中的中间数据(注意:并不一定会缓存每一个步骤的中间结果):
1、某个计算步骤非常耗时则会缓存此步骤的计算结果
2、计算步骤链很长时会加大缓存步骤计算结果的频次
3、shuffle到其它节点上的数据会被shuffle到的目标节点缓存一次
4、从一个作业切换到下一个作业时将发生一次checkpoint操作,在此,于checkpoint操作之前会缓存上一个作业的中间结果;checkpoint操作会将数据放置于磁盘文件系统中去以保证数据不发生丢失
RDD从逻辑上看是一个抽象分布式数据集的概念,它的底层数据存储于集群中不同节点上的磁盘文件系统中,存储是按照分区(partition)方式进行存储;所有Spark操作都可以看成是一系列对RDD对象的操作,而RDD是数据集合的抽象,它可以使用SparkContext(Spark上下文)来创建,SparkContext是Spark集群操作的入口,如果是在Spark-Shell下操作,则Spark会自动创建一个基于已有配置的默认SparkContext对象,如果是自己编写作业Jar则需要自己手动创建(与Hadoop中的FileSystem一样可以通过Configution配置参数来构建,也可以基于classpath中的配置文件来构建);
一个简单的测试：
//下面是通过Spark上下文调用textFile函数创建一个包装好底层数据集的RDD对象: //这里没有指定前缀hdfs://YunMaster01:9000,但是SparkContext(Spark上下文,即sc对象)中已经封装了这些默认的配置信息,即创建Spark上下文对象时就已经封装了这些上下文配置信息(包括操作的底层数据源,如这里的HDFS);这里最好不要加上hdfs前缀,因为加上了之后代码就被编译后固化了,造成不可配置,如果不加则SparkContext可以基于配置来自动选择(数据源类型可以通过配置来改变),提升高可维护性。 scala>val dataSet=sc.textFile("/spark/input") //SparkContext调用textFile函数将返回一个org.apache.spark.rdd.RDD[String]类型的对象(RDD在Spark的架构源码中被Scala定义为一个接口(trait)类型;方括号中的String表示RDD集合中的元素类型为字符串类型);此RDD的具体实现是MapPartitionRDD(即:字典分区RDD);textFile操作仅仅是执行数据抽象,并没有立即执行数据的读取(read)操作,数据的读取操作会延迟到执行action动作时才发生;因此textFile函数属于一种transformation动作,transformation动作是lazy级别的操作。 //查看数据源的输入路径(如果数据源使用HDFS则显示为HDFS路径),以下如不特别加以说明默认都是基于HDFS的数据源(因为这个最常用) //toDebugString函数会根据数据集的来源路径(RDD依赖)进行反向推理并从上到下依次显示RDD路径源列表 scala>dataSet.toDebugString() //返回数据集中的记录数量,这会启动一个action(textFile函数是一种转换(transformation),不会启动action);action是指Spark会在底层启动作业(Job)或任务(Task)的计算操作;而转换(transformation)仅仅是实现数据的封装和抽象(只涉及到数据在节点本地上的抽象读(注意:没有真正意义上的读,因为没有产生任何IO操作),但是数据的写操作则属于action操作,因为它涉及到IO操作过程,比如:saveAsTextFile函数) scala>dataSet.count //Spark中的一个分区(partition)相当于HDFS中的一个块(Block),即Spark中一个partition的尺寸等于一个HDFS文件块的尺寸(BlockSize=128M) //下面调用flatMap函数处理dataSet集合中的每一行,此函数调用完成之后又将产生一个新的RDD对象结果集,实际上所有的Spark操作都是基于RDD集合对象的操作(一切皆为RDD操作) scala>val dataSet02=dataSet.flatMap(_.split(" ")) //实现第二次映射操作产生新的RDD集合对象 scala>val dataSet03=dataSet02.map(word=>(word,1)) //上面的两个map算子相当于MapReduce中的map方法的操作;而下面与reduce相关的算子则相当于MapReduce中的reduce方法的操作 //由于reduce操作涉及到在节点之间转移和重新分配数据,因此此过程涉及到shuffle机制,执行reduce操作过程中产生的shuffle是为了归并相同Key的记录到同一分组以执行SQL概念中的分组统计 scala>val dataSet04=dataSet03.reduceByKey(_+_) //从上面的执行过程来看,Spark的底层计算原理与Hadoop的底层计算原理是类同的,只是Spark在此基础上引入了RDD的概念,将一切的操作归入集合RDD的操作(实际上,Hadoop中MapReduce操作本身就是针对集合对象进行操作,只不过Hadoop没有抽象出这样一个概念而已),由于引入了RDD的概念,所以Spark可以将一切操作(应用的所有操作步骤)的输入源和输出源都抽象为RDD对象的操作,这样以来,各个操作之间就可以通过RDD对象链接成操作链(上一个操作的输出源是一个RDD对象,该RDD对象可以直接作为下一个操作的输入源) //将应用执行过程中处理后的数据最终存储到HDFS文件系统中去(此过程是action操作) scala>dataSet04.saveAsTextFile("/spark/output")
从浏览器终端看到的Tasks任务详情列表中的Locality Level字段表示RDD迭代计算的数据来源:
| A、PROCESS_LOCAL:来自于内存 |
| B、NODE_LOCAL:来自于磁盘 |
| C、ANY:来自于其它节点的shuffle |
WordCount计算原理介绍:
A、textFile函数:返回一个HadoopRDD,HadoopRDD在执行action操作计算数据时会将数据从分布式磁盘读取到分布式内存中,即从worker节点的本地磁盘读取到worker节点的本地内存中,RDD更多的是指分布式内存存储,一种数据的逻辑存储系统；
HadoopRDD产生的是一个元组集合(即集合中的每个元素是一个元组类型),元组的第一个元素是行记录的索引(Key,字节偏移量),元组的第二个元素是行记录内容本身(Value)；
基于HadoopRDD会继续产生一个字符串集合MapPartitionRDD,此集合中的每个字符串代表行记录内容(没有行记录的索引,丢弃了字节偏移量Key)
B、flatMap函数:此函数仍然产生一个字符串MapPartitionRDD集合
C、map函数:此函数产生一个二元素元组类型的MapPartitionRDD集合
D、reduceByKey函数:基于相同Key的Value进行统计;先进行本地统计(执行卡宾函数统计可以减少数据传送量,降低网络负载),再进行集群模式统计(shuffle);本地统计之后的结果会根据分区策略将统计后的基于二元素元组的MapPartitionRDD集合数据写出到不同的本地文件中;在数据被传送到另一个节点之前的所有操作都是在同一个节点上的操作;在此,于同一个节点上的所有操作都是在同一个Stage中;同一个Stage中的所有操作都是基于内存的链式迭代操作(你也可以在这个迭代过程中手动缓存中间结果);我们通常说Spark是基于内存迭代的,其缘由就在于此。
E、shuffle是整个分布式计算的性能瓶颈所在,shuffle过程是从一个节点将数据传送到另一个节点的过程(即便在同一个节点上也是切换到另一个JVM进程来处理),这个过程中会针对每一份数据至少产生两次IO(本次磁盘IO、网络IO);shuffle过程是依据数据分区策略来进行的,shuffle之后将从一个Stage阶段过渡到另一个新的Stage阶段,shuffle之后将在另一个新的Stage阶段产生一个基于二元素元组的ShuffledRDD集合。
F、saveAsTextFile函数:将ShuffledRDD集合转换为MapPartitionRDD(实际上就是追加一个字节偏移量Key将每一个元组再包装成一个二元素元组(第一个元素是字节偏移量Key)),然后将其写出到HDFS文件系统中去。