Spark是一个由加州大学伯克利分校(UC Berkeley AMP)开发的一个分布式数据快速分析项目。它的核心技术是弹性分布式数据集(Resilient distributed datasets),提供了比Hadoop更加丰富的MapReduce模型,可以快速在内存中对数据集进行多次迭代,来支持复杂的数据挖掘算法和图计算算法。
Spark使用Scala开发,使用Mesos作为底层的调度框架,可以和hadoop和Ec2紧密集成,直接读取hdfs或S3的文件进行计算并把结果写回hdfs或S3,是Hadoop和Amazon云计算生态圈的一部分。Spark是一个小巧玲珑的项目,项目的core部分的代码只有63个Scala文件,充分体现了精简之美。
Spark之依赖
Spark与Hadoop的对比
Spark与Hadoop的结合
Spark可以直接对HDFS进行数据的读写,同样支持Spark on YARN。Spark可以与MapReduce运行于同集群中,共享存储资源与计算,数据仓库Shark实现上借用Hive,几乎与Hive完全兼容。
Spark的核心概念
1、Resilient Distributed Dataset (RDD)弹性分布数据集
RDD是Spark的最基本抽象,是对分布式内存的抽象使用,实现了以操作本地集合的方式来操作分布式数据集的抽象实现。RDD是Spark最核心的东西,它表示已被分区,不可变的并能够被并行操作的数据集合,不同的数据集格式对应不同的RDD实现。RDD必须是可序列化的。RDD可以cache到内存中,每次对RDD数据集的操作之后的结果,都可以存放到内存中,下一个操作可以直接从内存中输入,省去了MapReduce大量的磁盘IO操作。这对于迭代运算比较常见的机器学习算法, 交互式数据挖掘来说,效率提升比较大。
RDD的特点:
RDD的好处:
RDD的存储与分区:
RDD的内部表示:
RDD的存储级别:RDD根据useDisk、useMemory、deserialized、replication四个参数的组合提供了11种存储级别。RDD定义了各种操作,不同类型的数据由不同的RDD类抽象表示,不同的操作也由RDD进行抽实现。
RDD有两种创建方式:
2、Spark On Mesos
Spark支持Local调用和Mesos集群两种模式,在Spark上开发算法程序,可以在本地模式调试成功后,直接改用Mesos集群运行,除了文件的保存位置需要考虑以外,算法理论上不需要做任何修改。Spark的本地模式支持多线程,有一定的单机并发处理能力。但是不算很强劲。本地模式可以保存结果在本地或者分布式文件系统,而Mesos模式一定需要保存在分布式或者共享文件系统。
为了在Mesos框架上运行,安装Mesos的规范和设计,Spark实现两个类,一个是SparkScheduler,在Spark中类名是MesosScheduler;一个是SparkExecutor,在Spark中类名是Executor。有了这两个类,Spark就可以通过Mesos进行分布式的计算。Spark会将RDD和MapReduce函数,进行一次转换,变成标准的Job和一系列的Task。提交给SparkScheduler,SparkScheduler会把Task提交给Mesos Master,由Master分配给不同的Slave,最终由Slave中的Spark Executor,将分配到的Task一一执行,并且返回,组成新的RDD,或者直接写入到分布式文件系统。
3、Transformations & Actions
对于RDD可以有两种计算方式:转换(返回值还是一个RDD)与操作(返回值不是一个RDD)。
它们本质区别是:Transformation返回值还是一个RDD。它使用了链式调用的设计模式,对一个RDD进行计算后,变换成另外一个RDD,然后这个RDD又可以进行另外一次转换。这个过程是分布式的。Action返回值不是一个RDD。它要么是一个Scala的普通集合,要么是一个值,要么是空,最终或返回到Driver程序,或把RDD写入到文件系统中。关于这两个动作,在Spark开发指南中会有就进一步的详细介绍,它们是基于Spark开发的核心。这里将Spark的官方ppt中的一张图略作改造,阐明一下两种动作的区别。
4、Lineage(血统)
利用内存加快数据加载,在众多的其它的In-Memory类数据库或Cache类系统中也有实现,Spark的主要区别在于它处理分布式运算环境下的数据容错性(节点实效/数据丢失)问题时采用的方案。为了保证RDD中数据的鲁棒性,RDD数据集通过所谓的血统关系(Lineage)记住了它是如何从其它RDD中演变过来的。相比其它系统的细颗粒度的内存数据更新级别的备份或者LOG机制,RDD的Lineage记录的是粗颗粒度的特定数据转换(Transformation)操作(filter, map, join etc.)行为。当这个RDD的部分分区数据丢失时,它可以通过Lineage获取足够的信息来重新运算和恢复丢失的数据分区。这种粗颗粒的数据模型,限制了Spark的运用场合,但同时相比细颗粒度的数据模型,也带来了性能的提升。
RDD在Lineage依赖方面分为两种Narrow Dependencies与Wide Dependencies用来解决数据容错的高效性。
在RDD计算,通过checkpint进行容错,做checkpoint有两种方式,一个是checkpoint data,一个是logging the updates。用户可以控制采用哪种方式来实现容错,默认是logging the updates方式,通过记录跟踪所有生成RDD的转换(transformations)也就是记录每个RDD的lineage(血统)来重新计算生成丢失的分区数据。
Spark的Shuffle过程介绍
1.Shuffle Writer
Spark丰富了任务类型,有些任务之间数据流转不需要通过Shuffle,但是有些任务之间还是需要通过Shuffle来传递数据,比如wide dependency的group by key。
Spark中需要Shuffle输出的Map任务会为每个Reduce创建对应的bucket,Map产生的结果会根据设置的partitioner得到对应的bucketId,然后填充到相应的bucket中去。每个Map的输出结果可能包含所有的Reduce所需要的数据,所以每个Map会创建R个bucket(R是reduce的个数),M个Map总共会创建M*R个bucket。
Map创建的bucket其实对应磁盘上的一个文件,Map的结果写到每个bucket中其实就是写到那个磁盘文件中,这个文件也被称为blockFile,是Disk Block Manager管理器通过文件名的Hash值对应到本地目录的子目录中创建的。每个Map要在节点上创建R个磁盘文件用于结果输出,Map的结果是直接输出到磁盘文件上的,100KB的内存缓冲是用来创建Fast Buffered OutputStream输出流。这种方式一个问题就是Shuffle文件过多。
针对上述Shuffle过程产生的文件过多问题,Spark有另外一种改进的Shuffle过程:consolidation Shuffle,以期显著减少Shuffle文件的数量。在consolidation Shuffle中每个bucket并非对应一个文件,而是对应文件中的一个segment部分。Job的map在某个节点上第一次执行,为每个reduce创建bucket对应的输出文件,把这些文件组织成ShuffleFileGroup,当这次map执行完之后,这个ShuffleFileGroup可以释放为下次循环利用;当又有map在这个节点上执行时,不需要创建新的bucket文件,而是在上次的ShuffleFileGroup中取得已经创建的文件继续追加写一个segment;当前次map还没执行完,ShuffleFileGroup还没有释放,这时如果有新的map在这个节点上执行,无法循环利用这个ShuffleFileGroup,而是只能创建新的bucket文件组成新的ShuffleFileGroup来写输出。
比如一个Job有3个Map和2个reduce:(1) 如果此时集群有3个节点有空槽,每个节点空闲了一个core,则3个Map会调度到这3个节点上执行,每个Map都会创建2个Shuffle文件,总共创建6个Shuffle文件;(2) 如果此时集群有2个节点有空槽,每个节点空闲了一个core,则2个Map先调度到这2个节点上执行,每个Map都会创建2个Shuffle文件,然后其中一个节点执行完Map之后又调度执行另一个Map,则这个Map不会创建新的Shuffle文件,而是把结果输出追加到之前Map创建的Shuffle文件中;总共创建4个Shuffle文件;(3) 如果此时集群有2个节点有空槽,一个节点有2个空core一个节点有1个空core,则一个节点调度2个Map一个节点调度1个Map,调度2个Map的节点上,一个Map创建了Shuffle文件,后面的Map还是会创建新的Shuffle文件,因为上一个Map还正在写,它创建的ShuffleFileGroup还没有释放;总共创建6个Shuffle文件。
2.Shuffle Fetcher
Reduce去拖Map的输出数据,Spark提供了两套不同的拉取数据框架:通过socket连接去取数据;使用netty框架去取数据。
每个节点的Executor会创建一个BlockManager,其中会创建一个BlockManagerWorker用于响应请求。当Reduce的GET_BLOCK的请求过来时,读取本地文件将这个blockId的数据返回给Reduce。如果使用的是Netty框架,BlockManager会创建ShuffleSender用于发送Shuffle数据。并不是所有的数据都是通过网络读取,对于在本节点的Map数据,Reduce直接去磁盘上读取而不再通过网络框架。
Reduce拖过来数据之后以什么方式存储呢?Spark Map输出的数据没有经过排序,Spark Shuffle过来的数据也不会进行排序,Spark认为Shuffle过程中的排序不是必须的,并不是所有类型的Reduce需要的数据都需要排序,强制地进行排序只会增加Shuffle的负担。Reduce拖过来的数据会放在一个HashMap中,HashMap中存储的也是<key, value>对,key是Map输出的key,Map输出对应这个key的所有value组成HashMap的value。Spark将Shuffle取过来的每一个<key, value>对插入或者更新到HashMap中,来一个处理一个。HashMap全部放在内存中。
Shuffle取过来的数据全部存放在内存中,对于数据量比较小或者已经在Map端做过合并处理的Shuffle数据,占用内存空间不会太大,但是对于比如group by key这样的操作,Reduce需要得到key对应的所有value,并将这些value组一个数组放在内存中,这样当数据量较大时,就需要较多内存。
当内存不够时,要不就失败,要不就用老办法把内存中的数据移到磁盘上放着。Spark意识到在处理数据规模远远大于内存空间时所带来的不足,引入了一个具有外部排序的方案。Shuffle过来的数据先放在内存中,当内存中存储的<key, value>对超过1000并且内存使用超过70%时,判断节点上可用内存如果还足够,则把内存缓冲区大小翻倍,如果可用内存不再够了,则把内存中的<key, value>对排序然后写到磁盘文件中。最后把内存缓冲区中的数据排序之后和那些磁盘文件组成一个最小堆,每次从最小堆中读取最小的数据,这个和MapReduce中的merge过程类似。
3.MapReduce和Spark的Shuffle过程对比
Spark的资源管理与作业调度
Spark对于资源管理与作业调度可以使用本地模式,Standalone(独立模式),Apache Mesos及Hadoop YARN来实现。Spark on Yarn在Spark0.6时引用,但真正可用是在现在的branch-0.8版本。Spark on Yarn遵循YARN的官方规范实现,得益于Spark天生支持多种Scheduler和Executor的良好设计,对YARN的支持也就非常容易,Spark on Yarn的大致框架图。
让Spark运行于YARN上与Hadoop共用集群资源可以提高资源利用率。
编程接口
Spark通过与编程语言集成的方式暴露RDD的操作,类似于DryadLINQ和FlumeJava,每个数据集都表示为RDD对象,对数据集的操作就表示成对RDD对象的操作。Spark主要的编程语言是Scala,选择Scala是因为它的简洁性(Scala可以很方便在交互式下使用)和性能(JVM上的静态强类型语言)。
Spark和Hadoop MapReduce类似,由Master(类似于MapReduce的Jobtracker)和Workers(Spark的Slave工作节点)组成。用户编写的Spark程序被称为Driver程序,Dirver程序会连接master并定义了对各RDD的转换与操作,而对RDD的转换与操作通过Scala闭包(字面量函数)来表示,Scala使用Java对象来表示闭包且都是可序列化的,以此把对RDD的闭包操作发送到各Workers节点。 Workers存储着数据分块和享有集群内存,是运行在工作节点上的守护进程,当它收到对RDD的操作时,根据数据分片信息进行本地化数据操作,生成新的数据分片、返回结果或把RDD写入存储系统。
Spark生态系统
Spark的适用场景
在业界的使用
Spark项目在2009年启动,2010年开源, 现在使用的有:Berkeley, Princeton, Klout, Foursquare, Conviva, Quantifind, Yahoo! Research & others, 淘宝等,豆瓣也在使用Spark的python克隆版Dpark。
联系客服