Spark
Spark
基于MapReduce的计算引擎通常会将中间结果输出到磁盘上，进行存储和容错。出于任务管道承接的考虑，当一些查询翻译到MapReduce任务时，往往会产生多个Stage，而这些串联的Stage依赖底层文件系统（如HDFS）来存储每一个Stage的输出结果。Spark是MapReduce的替代方案，而且兼容HDFS、Hive，可融入Hadoop的生态系统，以弥补MapReduce的不足。
Spark的特点：【快】与Hadoop的MapReduce相比，Spark基于内存的运算要快100倍以上，基于硬盘的运算也要块10倍以上。Spark实现了高效的DAG执行引擎，可以通过基于内存来高效处理数据流。【通用】Spark可以用于批处理、交互式查询（Spark sql）、实时流处理（Spark Streaming）、机器学习（Spark MLlib）和图计算（Graphx）。【兼容性】Spark可以使用Hadoop的YARN和Apache Mesos作为它的资源管理和调度器，并且可以处理所有Hadoop支持的数据，包括HDFS、HBase和Cassandra等，使得不需要做任何数据迁移就可以使用Spark的强大处理能力。Spark实现了Standalone作为其内置的资源管理和调度框架，因此不依赖第三方的资源管理和调度器。
Spark里面有两种角色，一种是Master，一种是Worker。Master负责管理所有的Worker，判断这些Worker的状态，进而进行资源的调度。Worker管理当前计算节点，Worker会启动一个Executor来完成真正的计算。Spark会有专门的Driver，Driver会监控每一个任务(Spark每提交一个应用叫作app)。由于Master节点存在单点故障，要解决此问题需要借助Zookeeper并且启动至少两个Master节点来实现高可靠。如何启动Spark集群？
将spark-env.sh拷贝到其他节点上。如果想启动集群是在172.16.0.11这台机器上，在slaves文件中配置集群中的Worker所在机器的ip。执行sbin/start-all.sh在本机上启动一个Master进程。启动Master进程后，会读取slaves配置文件，发送SSH请求，通过调用脚本启动远端的Worker。与Hadoop类似，现在本机启动NameNode，然后通过SSH协议读取slaves文件，将远端的Datanode启动起来。Master和Worker通过TCP协议进行通信。
Spark提交任务流程
Master先启动，随后Worker向Master注册、发送心跳。启动Driver，Driver相当于Yarn中的ApplicationMaster，Driver负责管理它提交的任务，因为它提交任务会产生Executor，即提交100个Application就相应有100个Driver。Driver和Master建立连接（通过SparkContext）。Master会去找符合条件的Worker，给Worker分配资源。Master告诉Worker启动Executor，以后Executor主动寻找Driver进行RPC通信，不经过Master，然后在触发Action后Driver将任务以stage的形式提交，stage中包含一个task set。
Spark RDD
RDD（Resilient distributed Dataset）叫作分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显示地将工作集缓存在内存中，后续的查询能够重用工作集，极大地提升查询速度。Spark核心内核是一系列的RDD。RDD是一个弹性的、分布式的数据集，以并行化的方式操作它。
*Internally,each RDD is characterized by five main properties:
*
*A list of partitions
*A function for computing each split
*A list of dependencies on other RDDs
*Optionally,a Partitioner for key-value RDDs(e.g. to say that the RDD is hash-partitioned)
*Oprionally,a list of preferred locations to compute each split on(e.g. block location for an HDFS file)
- 一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值是程序所分配到的cpu Core的数目。
- 一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会以迭代器进行复合，不需要保存每次计算的结果。
- RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。
- 一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于key-value的RDD，才会有Partitioner，非key-value的RDD的Partitioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
- 一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能将任务分配到其所要处理数据块的存储位置。
创建RDD有两种方式。【1】通过HDFS支持的文件系统创建RDD，RDD里面没有真正要计算的数据，只记录了下一个元数据，比如：是哪里的数据、调用什么函数，一旦触发一个Action就会提交一个任务【2】通过Scala集合或数组以并行化的方式创建RDD
sc.textFile("hdfs://root:9000/file")
val rdd1=sc.parallelize(Array(1,2,3,4,5,6))//通过并行比如化Scala集合创建RDD
【RDD编程API】
在Spark里面对RDD操作分为两类，一类叫作Transformation（转换），转换具有延迟执行（软加载）的特点。Transformation会记录元数据信息，当计算任务触发Action时才会真正开始计算。为什么要进行软加载？因为存在很多这种算子，有一个算子就触发计算，需要计算很多次。另一种叫作Action（动作）。
常用的Transformation：
| 转换 | 含义 |
|---|---|
| map(func) | 返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成 |
| filter(func) | 返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成 |
| flatMap(func) | 类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素） |
| mapPartitions(func) | 类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T]=>Iterator{…} |
| mapPartitionsWithIndex(func) | 类似于mapPartitions，但func带有一个整数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(int,Iterator[T])=>iterator{…} |
| sample(withReplacement,Fraction,seed) | 根据fraction指定的比例对数据进行采样，可以选择是否使用随机数进行替换，seed用于指定随机数生成器种子 |
| union(other Dataset) | 对源RDD和参数RDD求并集后返回一个新的RDD |
| intersection(other Dataset) | 对源RDD和参数RDD求交集后返回一个新的RDD |
| distinct([numTasks]) | 对源RDD进行去重后后返回一个新的RDD |
| groupByKey([numTasks]) | 在一个(k,v)的RDD上调用，返回一个(k,Iterator[V])的RDD |
| reduceByKey(func [numTasks]) | 在一个(k,v)的RDD上调用，返回一个(k,v)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置 |
| aggregateByKey(zerovalue)(seqOp)([numTasks]) | |
| sortByKey([ascending],[numTasks]) | 在一个(k,v)的RDD上调用，k必须实现Ordered接口，返回一个按照key进行排序的(k,v)的RDD |
| sortBy(func,[ascending],[numTasks]) | 与sortByKey类似 |
| join(otherDataset,[numTasks]) | 在类型为(k,v)和(k,w)的RDD上调用，返回一个相同key对应的所有元素在一起的(k,fv,w)的RDD |
| cogroup(otherDataset,[numTasks]) | 在类型为(k,v)和(k,w)的RDD上调用，返回一个(k,iterable,iterable)类型的RDD |
| cartesian(otherDataset) | 笛卡尔积 |
| pipe(command,[envVars]) | |
| coalesce(numPartitions) | |
| repartition(numPartitions) | |
| repartitionAndSortWithinPartitions(partitioner) |
mapPartitionsWithIndex：传入两个参数，第一个参数是分区编号，第二个参数是迭代器，经过转换后返回一个迭代器itrator。
aggregate（聚合）：现对每个partition进行聚合，再对个部分的结果的进行聚合。
定义的数组内容在Master上，完成计算是在Worker上。RDD会在Master机器上启动一个Driver进程，专门用来提交程序。
aggregateByKey：对相同key的value进行操作
combineByKey：传进去3个函数，第一个函数相对于key进行分组，分组之后对分组里面的第一个元素进行操作，第二个函数是局部进行操作，第三个是对所有partition进行操作。reduceByKey底层是由combineByKey实现的。
val rdd1=sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"),3)
val rdd2=sc.parallelize(List(1,1,2,2,2,1,2,2,2),3)
val rdd3=rdd1.zip(rdd2)
val rdd=rdd3.combineByKey(List(),(x:List(String),y:String)=>x:+y,(m:List[String],n:List[String])=>m++n)
repartition：重新分区，假如数据原先在两个分区，运行这个命令重新分区为3个，数据要通过网络传输到另外的机器上。
Action：
| 动作 | 含义 |
|---|---|
| reduce(func) | 通过func函数聚集RDD中的所有元素，这个功能必须是可交换且可并联的 |
| collect() | 在驱动程序中，以数组的形式返回数据集的所有元素 |
| count() | 返回RDD的元素个数 |
| first() | 返回RDD的第一个元素（类似于take(1)） |
| take(n) | 返回一个由数据集的前n个元素组成的数组 |
| takeSample(withReplacement,num,[seed]) | 返回一个数组，该数组由从数据集中随机采样的num各元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子 |
| takeOrdered(n,[ordering]) | |
| saveAsTextFile(path) | 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它转换为文件中的文本 |
【RDD的依赖关系】
RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。窄依赖指每一个父RDD的Partition最多被子RDD的一个Partition使用，宽依赖指多个子RDD的Partition会依赖同一个父RDD的Partition。
【Lineage】
RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（即血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。
【RDD的缓存以及缓存方式】
Spark速度非常快的原因之一是在不同操作中可以在内存中持久化或缓存数据集，当持久化某个RDD后，每一个节点都将把计算的分片结果保存在内存中，并在对此RDD或衍生出的RDD进行的其他动作中重用。这使得后续的动作变得更加迅速。RDD相关的持久化和缓存是Spark最重要的特征之一，缓存是Spark构建迭代式算法和快速交互式查询的关键。
RDD通过persist方法或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中并供后面重用，其中cache调用persist方法。rdd.upersist(true)方法释放缓存。若数据大于可用内存，例如机器只有512M内存，HDFS中数据有1G，调用cache方法，会缓存512M数据，剩余数据还是在HDFS中。
/**Persist this RDD with the default storage level('MEMORY_ONLY')
def cache():this.type=persist()
/**Persist this RDD with the default storage level('MEMORY_ONLY')
def persist():this.type=persist(StorageLevel.MEMORY_ONLY)
RDD缓存可以对应不同的存储级别。
RDD转换
针对于下过程分析该计算中会产生多少RDD以及RDD之间的依赖关系。
val rdd=sc.textFile("hdfs://root:9000/app")
.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.saveAsTextFile(args(1))//产生6个RDD
Spark往HDFS中写数据，需要拿到HDFS一个流，然后往HDFS中写数据，如果是一条数据写一次，就会频繁打开、关闭一个流，效率会很低。可以拿到针对一个分区拿到一个流，把一个分区中的数据写到HDFS中。
【.textFile(“hdfs://root:9000/app”)】 会产生HadoopRDD、MapPartitionsRDD，因为通过Hadoop从HDFS中读数据，数据为K-V类型，Key是LongWritable，Value是caches，因为它用的InputFormat。之后发现偏移量没有用，把每一条数据进行处理（只保留Value），这时候需要经过一次map(RDD的Map)，生成MapPartitionsRDD，把每个分区的数据拿出来，调用Scala的map方法 。【.flatMap(.split(" ")) 】和【.map((,1))】产生一个RDD：MapPartitionsRDD。【.reduceByKey(+)】将reduceByKey转换成pariRDDFunction，reduceByKey调用combineByKeyWithclasstag，这个函数要求传入三个参数【1】把每个分区的key拿出来处理【2】局部的分区进行聚合操作【3】全部的数据拿出来聚合操作。reduceByKey会产生一个ShuffledRDD，会产生一个局部聚和。【.saveAsTextFile(args(1))】saveAsTextFile往HDFS中写数据，调用MapPartitionsRDD方法，把一个分区的数据拿出来然后建立一个流写数据。
CheckPoint
RDD中有很多分区，这些分区由多个Executor处理，每个Executor只处理一个或两个分区。一个分区肯定在一个Executor上，但是一台机器上可能有多个分区。Executor把它处理的分区中的数据缓存到内存/磁盘。数据序列化省内存但是耗cpu，不序列化耗内存但cpu负载会低些。尽管可以把数据保存到内存或磁盘，但是内存和磁盘都有可能坏掉，当进行下一步计算找不到数据了。由于RDD有容错机制，RDD和RDD会记录之间的依赖关系，即使以前的数据丢失了一部分，可以通过血统恢复，但是数据要经过几百次迭代，代价很大。这里可以把中间结果做一个checkpoint，把中间的结果保存到HDFS（HDFS是高可用的）。将分区中的数据以二进制的形式序列化到HDFS中。首先还是从内存中读取，如果内存中的数据丢失，就会去找RDD有没有checkpoint，RDD会去HDFS中读取数据。
/**
*Mark this RDD for checkpointing.It will be saved to a file inside the checkpoint
*directory set with 'SparkContextsetCheckpointDir' adn all references to its parent
*RDDs will be removed.This function must be called before any job has been
*executed on this RDD.It is strongly recommended that this RDD is persisted in
*memory,otherwise saving it on a file will require recomputation.
**/
在CheckPoint之前一定要指定一个存储目录：sc.setcheckpointDir("…")
并且rdd.checkpint是一个Transformation，在触发Action时会执行两个任务【1】计算Action任务【2】把数据保存到HDFS。在使用CheckPoint之前最好把数据缓存到内存中，速度会快很多，不用重新启动一个job，而是直接从内存当中把数据写入到HDFS中。之后的RDD在计算Transformation或Action时直接从CheckPoint中读取数据，不再依赖血统关系。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
