Spark源码学习2

转自:http://www.cnblogs.com/hseagle/p/3673123.html

在源码阅读时,需要重点把握以下两大主线。

  • 静态view 即 RDD, transformation and action
  • 动态view 即 life of a job, 每一个job又分为多个stage,每一个stage中可以包含多个rdd及其transformation,这些stage又是如何映射成为task被distributed到cluster中

一、概要

本文以wordCount为例,详细说明spark创建和运行job的过程,重点是在进程及线程的创建。

二、实验环境搭建

在进行后续操作前,确保下列条件已满足。

  1. 下载spark binary 0.9.1
  2. 安装scala
  3. 安装sbt
  4. 安装java

三、启动spark-shell

3.1 单机模式运行,即local模式

local模式运行非常简单,只要运行以下命令即可,假设当前目录是$SPARK_HOME

MASTER=local bin/spark-shell

"MASTER=local"就是表明当前运行在单机模式

3.2 local cluster方式运行

local cluster模式是一种伪cluster模式,在单机环境下模拟standalone的集群,启动顺序分别如下

  1. 启动master
  2. 启动worker
  3. 启动spark-shell

3.3 master

$SPARK_HOME/sbin/start-master.sh

注意运行时的输出,日志默认保存在$SPARK_HOME/logs目录。

master主要是运行类 org.apache.spark.deploy.master.Master,在8080端口启动监听,日志如下图所示

3.4 修改配置

  1. 进入$SPARK_HOME/conf目录
  2. 将spark-env.sh.template重命名为spark-env.sh
  3. 修改spark-env.sh,添加如下内容
export SPARK_MASTER_IP=localhost
export SPARK_LOCAL_IP=localhost

3.5 运行worker

bin/spark-class org.apache.spark.deploy.worker.Worker spark://localhost:7077 -i 127.0.0.1  -c 1 -m 512M

worker启动完成,连接到master。打开maser的web ui可以看到连接上来的worker. Master WEb UI的监听地址是http://localhost:8080

3.6 启动spark-shell

MASTER=spark://localhost:7077 bin/spark-shell

如果一切顺利,将看到下面的提示信息。

Created spark context..
Spark context available as sc.

可以用浏览器打开localhost:4040来查看如下内容

  1. stages
  2. storage
  3. environment
  4. executors

wordcount

上述环境准备妥当之后,我们在sparkshell中运行一下最简单的例子,在spark-shell中输入如下代码

scala>sc.textFile("README.md").filter(_.contains("Spark")).count

上述代码统计在README.md中含有Spark的行数有多少

四、部署过程详解

Spark布置环境中组件构成如下图所示。

  • Driver Program 简要来说在spark-shell中输入的wordcount语句对应于上图的Driver Program.
  • Cluster Manager 就是对应于上面提到的master,主要起到deploy management的作用
  • Worker Node 与Master相比,这是slave node。上面运行各个executor,executor可以对应于线程。executor处理两种基本的业务逻辑,一种就是driver programme,另一种就是job在提交之后拆分成各个stage,每个stage可以运行一到多个task

    Notes: 在集群(cluster)方式下, Cluster Manager运行在一个jvm进程之中,而worker运行在另一个jvm进程中。在local cluster中,这些jvm进程都在同一台机器中,如果是真正的standalone或Mesos及Yarn集群,worker与master或分布于不同的主机之上。

五、JOB的生成和运行

job生成的简单流程如下

  1. 首先应用程序创建SparkContext的实例,如实例为sc
  2. 利用SparkContext的实例来创建生成RDD
  3. 经过一连串的transformation操作,原始的RDD转换成为其它类型的RDD
  4. 当action作用于转换之后RDD时,会调用SparkContext的runJob方法
  5. sc.runJob的调用是后面一连串反应的起点,关键性的跃变就发生在此处

调用路径大致如下

  1. sc.runJob->dagScheduler.runJob->submitJob
  2. DAGScheduler::submitJob会创建JobSummitted的event发送给内嵌类eventProcessActor
  3. eventProcessActor在接收到JobSubmmitted之后调用processEvent处理函数
  4. job到stage的转换,生成finalStage并提交运行,关键是调用submitStage
  5. 在submitStage中会计算stage之间的依赖关系,依赖关系分为宽依赖窄依赖两种
  6. 如果计算中发现当前的stage没有任何依赖或者所有的依赖都已经准备完毕,则提交task
  7. 提交task是调用函数submitMissingTasks来完成
  8. task真正运行在哪个worker上面是由TaskScheduler来管理,也就是上面的submitMissingTasks会调用TaskScheduler::submitTasks
  9. TaskSchedulerImpl中会根据Spark的当前运行模式来创建相应的backend,如果是在单机运行则创建LocalBackend
  10. LocalBackend收到TaskSchedulerImpl传递进来的ReceiveOffers事件
  11. receiveOffers->executor.launchTask->TaskRunner.run

代码片段executor.lauchTask

1 def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) {
2     val tr = new TaskRunner(context, taskId, serializedTask)
3     runningTasks.put(taskId, tr)
4     threadPool.execute(tr)
5   }

说了这么一大通,也就是讲最终的逻辑处理切切实实是发生在TaskRunner这么一个executor之内,运算结果是包装成为MapStatus然后通过一系列的内部消息传递,反馈到DAGScheduler,这一个消息传递路径不是过于复杂,有兴趣可以自行勾勒

时间: 2024-11-06 11:36:13

Spark源码学习2的相关文章

Spark源码学习和总结1

一. Spark源码编译 解析: wget http://archive.apache.org/dist/spark/spark-1.6.0/spark-1.6.0-bin-hadoop2.6.tgz gtar -zxvf spark-1.6.0-bin-hadoop2.6.tgz cd spark-1.6.0-bin-hadoop2.6 ./sbt/sbt gen-idea 说明:漫长的等待后,上述命令生成SBT工程,我们就可以使用IDEA以SBT Project的方式打开. 二. RDD实现

Spark源码学习1.1——DAGScheduler.scala

本文以Spark1.1.0版本为基础. 经过前一段时间的学习,基本上能够对Spark的工作流程有一个了解,但是具体的细节还是需要阅读源码,而且后续的科研过程中也肯定要修改源码的,所以最近开始Spark的源码的学习.首先以重要文件为基础分别分析,然后再整体的分析. (一)DAGScheduler.scala文件的主要功能 DAGScheduler是划分Job为stage的调度,它是在作业所需要的数据已经被分为RDD之后执行的.DAGScheduler将Job划分为DAG图,以stage为图的结点,

spark源码学习-withScope

withScope是最近的发现版中新增加的一个模块,它是用来做DAG可视化的(DAG visualization on SparkUI) 以前的sparkUI中只有stage的执行情况,也就是说我们不可以看到上个RDD到下个RDD的具体信息.于是为了在 sparkUI中能展示更多的信息.所以把所有创建的RDD的方法都包裹起来,同时用RDDOperationScope 记录 RDD 的操作历史和关联,就能达成目标.下面就是一张WordCount的DAG visualization on Spark

Spark源码学习1.4——MapOutputTracker.scala

相关类:MapOutputTrackerMessage,GetMapOutputStatuses extends MapPutputTrackerMessage,StopMapOutputTracker extends MapOutputTrackerMessage,MapOutputTrackerMasterActor,MapOutputTrackerMaster. 首先重写MapOutputTrackerMasterActor的receiveWithLogging:判断MapOutputTr

Spark源码学习1.5——BlockManager.scala

一.BlockResult类 该类用来表示返回的匹配的block及其相关的参数.共有三个参数: data:Iterator [Any]. readMethod: DataReadMethod.Value. bytes: Long. 实例化InputMetrics类. 二.BlockManager类 关系的参数有executorId,blockManagerMaster,mapOutPutTracker等.关系的类有ShuffleBlockManager类,DiskBlockManager类,Co

Spark源码学习和总结2

2. SchedulerBackend SchedulerBackend是一个trait,主要的功能是向当前等待分配计算资源的Task分配计算资源Executor,并且在分配的Executor上启动Task,完成计算的调度过程.上述任务调度是通过方法def reviveOffers(): Unit实现的.需要说明的是CoarseGrainedSchedulerBackend是SchedulerBackend的一个具体实现,而YARN.Standalone和Mesos都是基于SchedulerBa

Spark源码学习1.6——Executor.scala

Executor.scala 一.Executor类 首先判断本地性,获取slaves的host name(不是IP或者host: port),匹配运行环境为集群或者本地.如果不是本地执行,需要启动一个handler来监控所有的executor进程,避免阻塞.然后,初始化Spark执行环境.向SparkEnv注册executor资源,即registerSource方法.第三步,装载类,序列化类到内存中.第四,启动worker的线程池.第五,收集所有的task任务.接下就可以分配资源给task了,

Spark源码学习1.3——TaskSetManager.scala

TaskSetManager.scala TaskSet是指一系列被提交的task,一般是代表特定的stage中丢失的partition.TaskSetManager通过一个TaskSchedulerImpl实体来对TaskSet进行管理. 该方法定义了以下参数: EXECUTOR_TASK_BLACKLIST_TIMEOUT:executor加入黑名单的时间.如果executor意外终止导致任务执行失败,那么会暂时将这个executor加入黑名单,不再分配任务给它. SPECULATION_Q

Spark 消息队列机制源码学习

源码学习 spark源码注释中有下面一句话: Asynchronously passes SparkListenerEvents to registered SparkListeners 即所有spark消息SparkListenerEvents 被异步的发送给已经注册过的SparkListeners. 在SparkContext中, 首先会创建LiveListenerBus实例,这个类主要功能如下: 保存有消息队列,负责消息的缓存 保存有注册过的listener,负责消息的分发 该类的继承层次