Spark分布式计算执行模型

引言

相对Hadoop, Spark在处理需要迭代运算的机器学习训练等任务上有着很大性能提升,同时提供了批处理、实时数据处理、机器学习以及图算法等一站式的服务,因此最近大家一起来学习Spark,特别是MLLib

Spark中使用了RDD(Resilient Distributed Datasets, 弹性分布式数据集)抽象分布式计算,即使用RDD以及对应的transform/action等操作来执行分布式计算;并且基于RDD之间的依赖关系组成lineage以及checkpoint等机制来保证整个分布式计算的容错性。因此,在学习MLLib的时候,很多时候看到的都是RDD的一些操作,而没有涉及到分布式计算上来,如下面的代码:

// 创建SparkContext对象,作为Spark的Driver应用程序,同Spark集群进行交互
val conf = new SparkConf().setAppName(s"LinearRegression with $params")
val sc = new SparkContext(conf)
    
// 加载libsvm格式的训练、测试instances数据
val examples = MLUtils.loadLibSVMFile(sc, params.input, multiclass = true).cache()

// 将数据分为训练集和测试集
val splits = examples.randomSplit(Array(0.8, 0.2))
val training = splits(0).cache()
val test = splits(1).cache()

examples.unpersist(blocking = false)

// 设置离线训练算法
val updater = params.regType match {
  case NONE => new SimpleUpdater()
  case L1 => new L1Updater()
  case L2 => new SquaredL2Updater()
}

val algorithm = new LinearRegressionWithSGD()
algorithm.optimizer
  .setNumIterations(params.numIterations)
  .setStepSize(params.stepSize)
  .setUpdater(updater)
  .setRegParam(params.regParam)

// 根据设置的训练算法和训练数据集,得到模型
val model = algorithm.run(training)

// 使用模型来预测测试集合上对应的值
val prediction = model.predict(test.map(_.features))

由于Spark使用RDD抽象了分布式计算的操作,因此,上面的代码只是涉及到训练集和测试集以及上面的操作,感觉不到该程序是单机模型训练还是分布式模型训练。因此,有同学提出来问题,这些RDD操作如何进行分布式计算的呢?这涉及Spark分布式计算执行模型。下面从Spark集群部署和应用程序提交、执行模型来展开介绍Spark如何进行分布式计算。

Spark集群部署和应用程序提交

Spark应用程序部署

Spark集群部署

Spark集群是由Cluster Manager和Worker Node组成。当前Spark支持如下三种不同的Cluster Manager:

  1. Standalone – 使用./sbin/start-master.sh和./sbin/start-slave.sh或者./sbin/start-all.sh即可非常容易地将Spark集群搭建起来,Spark内置的集群资源管理器。当前我们的实验Spark集群是基于Standalone的集群管理模式搭建起来的
  2. Apache Mesos – 基于Mesos资源管理器来管理机器资源
  3. Hadoop Yarn – 基于Hadoop 2里面的资源管理器来管理机器资源

提交Spark应用程序

不管Spark集群是基于什么样资源管理器进行管理,通过spark-submit往Spark集群上提交应用程序(包括有SparkContext对象的应用程序叫做Driver),提交Driver应用程序的时候,需指定Cluster Manager的地址,所需要的CPU核数、内存数目等。具体例子如下:

# Run on a Spark standalone cluster
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://207.184.161.138:7077 \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

# Run on a YARN cluster
export HADOOP_CONF_DIR=XXX
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn-cluster \  # can also be `yarn-client` for client mode
  --executor-memory 20G \
  --num-executors 50 \
  /path/to/examples.jar \
  1000

提交完应用程序之后,即Driver通过Cluster Manager去Worker Node上分配所需要的CPU和内存。从而有了分布式计算所需要的物理资源,即拥有分布在Spark集群中Worker Node上的ExecutorBackend进程,该进程等待来自Driver的Task任务。

下面以Standalone集群模式为例说明整个过程,Master对应的时Cluster Manager, Worker对应为Worker Node. 整个操作过程如下:

Spark Submit Application Process

  1. Driver通过AppClient向Master发送了RegisterApplication消息来注册Application
  2. Master收到消息之后会发送RegisteredApplication通知Driver注册成功,Driver的接收类还是AppClient
  3. Master接受到RegisterApplication之后会触发调度过程,在资源足够的情况下会向Woker和Driver分别发送LaunchExecutor、ExecutorAdded消息
  4. Worker接收到LaunchExecutor消息之后,会执行消息中携带的命令,执行CoarseGrainedExecutorBackend类(图中仅以它继承的接口ExecutorBackend代替),执行完毕之后会发送ExecutorStateChanged消息给Master
  5. Master接收ExecutorStateChanged之后,立即发送ExecutorUpdated消息通知Driver。Driver中的AppClient接收到Master发过来的ExecutorAdded和ExecutorUpdated后进行相应的处理
  6. 启动之后的CoarseGrainedExecutorBackend会向Driver发送RegisterExecutor消息
  7. Driver中的SparkDeploySchedulerBackend(具体代码在CoarseGrainedSchedulerBackend里面)接收到RegisterExecutor消息,回复注册成功的消息RegisteredExecutor给ExecutorBackend,并且立马准备给它发送任务

最后,CoarseGrainedExecutorBackend接收到RegisteredExecutor消息之后,实例化一个Executor等待任务的到来

Spark执行模型

Spark执行模型分如下三步:

  1. 创建应用程序计算RDD DAG (Directed acyclic graph,有向无环图)
  2. 创建RDD DAG逻辑执行方案,即将整个计算过程对应到Stage上
  3. 根据上面介绍获取到Executor来进行调度并执行各个Stage对应的ShuffleMapResult和ResultTask等任务。必须是执行一个Stage完成之后,才能往下执行接下来的Stage

下面以WordCount例子来说明Spark执行模型

WordCount Job例子

下面以map-reduce中经典例子WordCount为例解释Spark执行模型,整个WordCount Job是计算README.md这个文档中各个单词出现的频次,并且将最终结果保存到wordcount_result目录中去

val wordCountResult = sc.textFile("README.md", 4)
  .flatMap(line => line.split(" ")).map(word => (word, 1))
  .reduceByKey(_ + _, 2)
vordCountResult.saveAsTextFile("wordcount_result")

RDD DAG

RDD DAG描述的是各个RDD之间的依赖关系。上面例子从RDD DAG的角度来看如下:

RDD DAG

即该RDD DAG主要是包括有MappedRDD->FlatMappedRDD->MappedRDD->ShuffledRDD四个RDD的转换(Transform), 根据Spark实现,RDD的转换操作是不会提交给Spark集群来执行的,因此,上面的操作必须要由Spark的行为(Action)来触发,因此,在最后调用saveAsTextFile这个行为来将整个WordCount Job提交到Spark集群中来执行。(备注:所有的Spark的转换、行为操作可以参考文档Spark Programming Guide

RDD DAG逻辑执行方案

RDD DAG只是从整体的RDD角度来查看整个Job的执行过程。在RDD DAG逻辑执行方案,需要查看各个RDD中各个Partition的情况,以及各个RDD的Partition的依赖情况来决定如何划分Stage。

根据RDD论文,将RDD的各个Partition的依赖情况划分为Narrow Dependencies和Wide Dependencies:

  • Narrow Dependencies – parent RDD中的一个Partition最多被child RDD中的一个Partition所依赖
  • Wide Dependencies – parent RDD中的一个Partition被child RDD中的多个Partition所依赖

Narrow and Wide Dependencies

如图所示,map是Narrow Dependencies, groupByKey是Wide Dependencies。若在Job中存在有Wide Dependencies,就划分为不同的Stage。

具体到WordCount Job,具体的Stage划分如下:

Spark Stage Partition划分

由于flatMap、map等操作对RDD进行转换得到的RDD的partition和parent RDD的partition是Narrow Dependencies关系,因此处于在同一个Stage中,即都在Stage 1中;而reduceByKey这个转换,其对应的是Wide Dependencies关系,因此,需要新建一个Stage出来,即所在为Stage 2,独立于Stage 1。

当Job执行saveAsTextFile这个行为的时候,其依赖于Stage 2中ShuffledRDD,而Stage 2又依赖于Stage 1,因此,需要先执行完Stage 1中所有的Task之后,才执行Stage 2中的所有的Task。当Stage 2中所有的任务执行完之后,整个Job即执行完成。

RDD Task执行

Spark通过分析各个RDD的依赖关系生成了RDD DAG,然后再通过分析各个RDD中的partition之间的依赖关系来将执行过程进行逻辑划分成不同的Stage。有了这些Stage的依赖关系之后,从最parent stage开始执行,执行完了parent stage的所有的task再执行child stage中的所有的task,直到所有的Stage都执行完成。

针对WordCount这个例子来看,Stage 2依赖Stage 1,因此,先执行Stage 1中的Task。而Stage 1中各个RDD中是有4个partition(见textFile(“README.md”, 4)中的第二个参数来指定RDD中需要划分多少partition,当然对于RDD也可以通过调用repartition和coalesce来改变partition数目),因此,在Stage 1中由Driver应用程序生成4个ShuffleMapTask并提交给之前分配得到的Executor中执行。当Stage 1中4个ShuffledMapTask执行完成之后,再开始执行Stage 2中的2个ResultTask(由于reduceByKey(_ + _, 2)中的第二个参数指定只需要2个reduce,因此,在ShuffledRDD中只有2个partition,因此,也只有对应的2个ResultTask).当Stage 2中的2个ResultTask执行完之后,saveAsTextFile会将ShuffledRDD中的内容落地到文件中中,即保存到wordcount_result目录中去。从而完成了整个WordCount Job的执行任务。

从上面的描述可以看到,Stage 1中会生成4个ShuffleMapTask, 在提交WordCount Job应用程序给Spark集群时候,获取得到的Executor数目大于等于4个,那么该4个ShuffleMapTask可以在这些Executor进行并行运行,从而实现了在不同的Executor进行分布式计算。

最后说明下,RDD的Partition数目决定了执行过程中生成多少个Task,即决定于并行计算的数目,该参数是Spark应用程序中非常重要的参数,Partition设置的越大,并行度越高,在Executor资源有限的情况下,任务之间调度开销会变大,同时若有Wide Dependencies的时候,Shuffle的代价也比较多,因此在实际应用中需要谨慎调整该参数。Spark作者推荐的“比较合理的partition数目”为:

  1. 100-10000
  2. 最少要有2倍于申请的CPU核数
  3. 每个Partition对应的Task最少要运行100ms以上

结束语

RDD为Spark抽象了分布式计算的操作,即将任务进行分布式计算转成RDD的转换和行为上。为了了解Spark究竟如何进行分布式计算的,本文首先介绍提交Driver应用程序给Spark集群,通过同Cluster Manager和Worker Node进行交互,得到该Driver所需要的Executor资源,然后再由Spark应用程序通过分析RDD DAG依赖关系,以及各个RDD之间partition的依赖关系来生成不同的Stage,再将Stage中的任务,按照RDD的partition个数生成相同数目的Task提交给Executor来执行,从而实现了Task在不同的Executor中进行分布式计算,最终实现整个Driver应用程序的分布式计算。

本文链接:Spark分布式计算执行模型
本站文章若无特别说明,皆为原创,转载请注明来源:火光摇曳,谢谢!^^


火光摇曳

Spark分布式计算执行模型》有3个想法

  1. 相亲时相中一个男的,他离过婚,据他说是他老婆不能生过了两年又离婚的!我不想嫁个二婚就没愿意,但他后来老是给我打电话,我对他又不反感也就聊着,直到有一天他表达的意思是要我先怀孕再结婚我就拉黑他了!

发表评论

电子邮件地址不会被公开。 必填项已用*标注

*