Apache Spark源码走读之15 -- Standalone部署模式下的容错性分析

欢迎转载,转载请注明出处,徽沪一郎。

概要

本文就standalone部署方式下的容错性问题做比较细致的分析,主要回答standalone部署方式下的包含哪些主要节点,当某一类节点出现问题时,系统是如何处理的。

Standalone部署的节点组成

介绍Spark的资料中对于RDD这个概念涉及的比较多,但对于RDD如何运行起来,如何对应到进程和线程的,着墨的不是很多。

在实际的生产环境中,Spark总是会以集群的方式进行运行的,其中standalone的部署方式是所有集群方式中最为精简的一种,另外是Mesos和YARN,要理解其内部运行机理,显然要花更多的时间才能了解清楚。

standalone cluster的组成

standalone集群由三个不同级别的节点组成,分别是

  • Master 主控节点,可以类比为董事长或总舵主,在整个集群之中,最多只有一个Master处在Active状态
  • Worker 工作节点 ,这个是manager,是分舵主, 在整个集群中,可以有多个worker,如果worker为零,什么事也做不了
  • Executor 干苦力活的,直接受worker掌控,一个worker可以启动多个executor,启动的个数受限于机器中的cpu核数

这三种不同类型的节点各自运行于自己的JVM进程之中

Driver Application

提交到standalone集群的应用程序称之为Driver Applicaton。

Standalone集群启动及任务提交过程详解

上图总结了正常情况下Standalone集群的启动以及应用提交时,各节点之间有哪些消息交互。下面分集群启动和应用提交两个过程来作详细说明。

集群启动过程

正常启动过程如下所述

step 1: 启动master

$SPARK_HOME/sbin/start-master.sh

step 2: 启动worker

./bin/spark-class org.apache.spark.deploy.worker.Worker spark://localhost:7077

worker启动之后,会做两件事情

  1. 将自己注册到Master, RegisterWorker
  2. 定期发送心跳消息给Master

任务提交过程

step 1: 提交application

利用如下指令来启动spark-shell

MASTER=spark://127.0.0.1:7077 $SPARK_HOME/bin/spark-shell

运行spark-shell时,会向Master发送RegisterApplication请求

日志位置: master运行产生的日志在$SPARK_HOME/logs目录下

step 2: Master处理RegisterApplication的请求之后

收到RegisterApplication请求之后,Mastet会做如下处理

  1. 如果有worker已经注册上来,发送LaunchExecutor指令给相应worker
  2. 如果没有,则什么事也不做

step 3: 启动Executor

Worker在收到LaunchExecutor指令之后,会启动Executor进程

step 4: 注册Executor

启动的Executor进程会根据启动时的入参,将自己注册到Driver中的SchedulerBackend

日志位置: executor的运行日志在$SPARK_HOME/work目录下

step 5: 运行Task

SchedulerBackend收到Executor的注册消息之后,会将提交到的Spark Job分解为多个具体的Task,然后通过LaunchTask指令将这些Task分散到各个Executor上真正的运行

如果在调用runJob的时候,没有任何的Executor注册到SchedulerBackend,相应的处理逻辑是什么呢?

  1. SchedulerBackend会将Task存储在TaskManager中
  2. 一旦有Executor注册上来,就将TaskManager管理的尚未运行的task提交到executor中
  3. 如果有多个job处于pending状态,默认调度策略是FIFO,即先提交的先运行

测试步骤

  1. 启动Master
  2. 启动spark-shell
  3. 执行 sc.textFile("README.md").count
  4. 启动worker
  5. 注意worker启动之后,spark-shell中打印出来的日志消息

Job执行结束

任务运行结束时,会将相应的Executor停掉。

可以做如下的试验

  1. 停止spark-shell
  2. 利用ps -ef|grep -i java查看java进程,可以发现CoarseGrainedExecutorBackend进程已经退出

小结

通过上面的控制消息原语之间的先后顺序可以看出

  1. Master和worker进程必须显式启动
  2. executor是被worker隐式的带起
  3. 集群的启动顺序
    1. Master必须先于其它节点启动
    2. worker和driver哪个先启动,无所谓
    3. 但driver提交的job只有在有相应的worker注册到Master之后才可以被真正的执行

异常场景分析

上面说明的是正常情况下,各节点的消息分发细节。那么如果在运行中,集群中的某些节点出现了问题,整个集群是否还能够正常处理Application中的任务呢?

异常分析1: worker异常退出

在Spark运行过程中,经常碰到的问题就是worker异常退出,当worker退出时,整个集群会有哪些故事发生呢? 请看下面的具体描述

  1. worker异常退出,比如说有意识的通过kill指令将worker杀死
  2. worker在退出之前,会将自己所管控的所有小弟executor全干掉
  3. worker需要定期向master改善心跳消息的,现在worker进程都已经玩完了,哪有心跳消息,所以Master会在超时处理中意识到有一个“分舵”离开了
  4. Master非常伤心,伤心的Master将情况汇报给了相应的Driver
  5. Driver通过两方面确认分配给自己的Executor不幸离开了,一是Master发送过来的通知,二是Driver没有在规定时间内收到Executor的StatusUpdate,于是Driver会将注册的Executor移除

后果分析

worker异常退出会带来哪些影响

  1. executor退出导致提交的task无法正常结束,会被再一次提交运行
  2. 如果所有的worker都异常退出,则整个集群不可用
  3. 需要有相应的程序来重启worker进程,比如使用supervisordrunit

测试步骤

  • 启动Master
  • 启动worker
  • 启动spark-shell
  • 手工kill掉worker进程
  • 用jps或ps -ef|grep -i java来查看启动着的java进程

异常退出的代码处理

定义于ExecutorRunner.scala的start函数

def start() {
    workerThread = new Thread("ExecutorRunner for " + fullId) {
      override def run() { fetchAndRunExecutor() }
    }
    workerThread.start()
    // Shutdown hook that kills actors on shutdown.
    shutdownHook = new Thread() {
      override def run() {
        killProcess(Some("Worker shutting down"))
      }
    }
    Runtime.getRuntime.addShutdownHook(shutdownHook)
  }

killProcess的过程就是停止相应CoarseGrainedExecutorBackend的过程。

worker停止的时候,一定要先将自己启动的Executor停止掉。这是不是很像水浒中宋江的手段,李逵就是这样不明不白的把命给丢了。

小结

需要特别指出的是,当worker在启动Executor的时候,是通过ExecutorRunner来完成的,ExecutorRunner是一个独立的线程,和Executor是一对一的关系,这很重要。Executor作为一个独立的进程在运行,但会受到ExecutorRunner的严密监控。

异常分析2: executor异常退出

Executor作为Standalone集群部署方式下的最底层员工,一旦异常退出,其后果会是什么呢?

  1. executor异常退出,ExecutorRunner注意到异常,将情况通过ExecutorStateChanged汇报给Master
  2. Master收到通知之后,非常不高兴,尽然有小弟要跑路,那还了得,要求Executor所属的worker再次启动
  3. Worker收到LaunchExecutor指令,再次启动executor

作为一名底层员工,想轻易摞挑子不干是不成的。"人在江湖,身不由己“啊。

测试步骤

  • 启动Master
  • 启动Worker
  • 启动spark-shell
  • 手工kill掉CoarseGrainedExecutorBackend

fetchAndRunExecutor

fetchAndRunExecutor负责启动具体的Executor,并监控其运行状态,具体代码逻辑如下所示

def fetchAndRunExecutor() {
    try {
      // Create the executor‘s working directory
      val executorDir = new File(workDir, appId + "/" + execId)
      if (!executorDir.mkdirs()) {
        throw new IOException("Failed to create directory " + executorDir)
      }

      // Launch the process
      val command = getCommandSeq
      logInfo("Launch command: " + command.mkString("\"", "\" \"", "\""))
      val builder = new ProcessBuilder(command: _*).directory(executorDir)
      val env = builder.environment()
      for ((key, value)  {
        logInfo("Runner thread for executor " + fullId + " interrupted")
        state = ExecutorState.KILLED
        killProcess(None)
      }
      case e: Exception => {
        logError("Error running executor", e)
        state = ExecutorState.FAILED
        killProcess(Some(e.toString))
      }
    }
  }

异常分析3: master 异常退出

worker和executor异常退出的场景都讲到了,我们剩下最后一种情况了,master挂掉了怎么办?

带头大哥如果不在了,会是什么后果呢?

  • worker没有汇报的对象了,也就是如果executor再次跑飞,worker是不会将executor启动起来的,大哥没给指令
  • 无法向集群提交新的任务
  • 老的任务即便结束了,占用的资源也无法清除,因为资源清除的指令是Master发出的

怎么样,知道后果很严重了吧?别看老大平时不干活,要真的不在,仅凭小弟们是不行的。

Master单点失效问题的解决

那么怎么解决Master单点失效的问题呢?

你说再加一个Master就是了,两个老大。两个老大如果同时具有指挥权,结果也将是灾难性的。设立一个副职人员,当目前的正职挂掉之后,副职接管。也就是同一时刻,有且只有一个active master。

注意不错,如何实现呢?使用zookeeper的ElectLeader功能,效果图如下

配置细节

如何搭建zookeeper集群,这里不再废话,哪天有空的话再整一整,或者可以参考写的storm系列中谈到的zookeeper的集群安装步骤。

假设zookeeper集群已经设置成功,那么如何启动standalone集群中的节点呢?有哪些特别的地方?

conf/spark-env.sh

在conf/spark-env.sh中,为SPARK_DAEMON_JAVA_OPTS添加如下选项

System property Meaning
spark.deploy.recoveryMode Set to ZOOKEEPER to enable standby Master recovery mode (default: NONE).
spark.deploy.zookeeper.url The ZooKeeper cluster url (e.g., 192.168.1.100:2181,192.168.1.101:2181).
spark.deploy.zookeeper.dir The directory in ZooKeeper to store recovery state (default: /spark).

设置SPARK_DAEMON_JAVA_OPTS的实际例子

SPARK_DAEMON_JAVA_OPTS="$SPARK_DAEMON_JAVA_OPTS -Dspark.deploy.recoveryMode=ZOOKEEPER"

应用程序启动

应用程序运行的时候,指定多个master地址,用逗号分开,如下所示

MASTER=spark://192.168.100.101:7077,spark://192.168.100.102:7077 bin/spark-shell

小结

Standalone集群部署方式下的容错性分析让我们对于Spark的任务分发过程又有了进一处的认识。前面的篇章从整体上匆匆过了一遍Spark所涉及的知识点,分析的不够深,不够细。

此篇尝试着就某一具体问题做深入的分析。套用书画中的说法,在框架分析的时候,我们可以”大开大合,疏可走马,计白当黑“,在细节分析的时候,又要做到“密不透风,条分缕析,层层递进”。

Apache Spark源码走读之15 -- Standalone部署模式下的容错性分析,布布扣,bubuko.com

时间: 2024-10-18 01:09:11

Apache Spark源码走读之15 -- Standalone部署模式下的容错性分析的相关文章

Apache Spark源码走读之19 -- standalone cluster模式下资源的申请与释放

欢迎转载,转载请注明出处,徽沪一郎. 概要 本文主要讲述在standalone cluster部署模式下,Spark Application在整个运行期间,资源(主要是cpu core和内存)的申请与释放. 构成Standalone cluster部署模式的四大组成部件如下图所示,分别为Master, worker, executor和driver,它们各自运行于独立的JVM进程. 从资源管理的角度来说 Master  掌管整个cluster的资源,主要是指cpu core和memory,但Ma

Apache Spark源码走读之7 -- Standalone部署方式分析

欢迎转载,转载请注明出处,徽沪一郎. 楔子 在Spark源码走读系列之2中曾经提到Spark能以Standalone的方式来运行cluster,但没有对Application的提交与具体运行流程做详细的分析,本文就这些问题做一个比较详细的分析,并且对在standalone模式下如何实现HA进行讲解. 没有HA的Standalone运行模式 先从比较简单的说起,所谓的没有ha是指master节点没有ha. 组成cluster的两大元素即Master和Worker.slave worker可以有1到

Apache Spark技术实战之8:Standalone部署模式下的临时文件清理

未经本人同意严禁转载,徽沪一郎. 概要 在Standalone部署模式下,Spark运行过程中会创建哪些临时性目录及文件,这些临时目录和文件又是在什么时候被清理,本文将就这些问题做深入细致的解答. 从资源使用的方面来看,一个进程运行期间会利用到这四个方面的资源,分别是CPU,内存,磁盘和网络.进程退出之后,CPU,内存和网络都会由操作系统负责释放掉,但是运行过程中产生临时文件如果进程自己不在退出之前有效清除,就会留下一地鸡毛,浪费有效的存储空间. 部署时的第三方依赖 再提出具体的疑问之前,先回顾

Apache Spark源码走读之3 -- Task运行期之函数调用关系分析

概要 本篇主要阐述在TaskRunner中执行的task其业务逻辑是如何被调用到的,另外试图讲清楚运行着的task其输入的数据从哪获取,处理的结果返回到哪里,如何返回. 准备 1. spark已经安装完毕 2. spark运行在local mode或local-cluster mode local-cluster mode local-cluster模式也称为伪分布式,可以使用如下指令运行 MASTER=local[1,2,1024] bin/spark-shell [1,2,1024] 分别表

Apache Spark源码走读之21 -- WEB UI和Metrics初始化及数据更新过程分析

欢迎转载,转载请注明出处,徽沪一郎. 概要 WEB UI和Metrics子系统为外部观察监测Spark内部运行情况提供了必要的窗口,本文将简略的过一下其内部代码实现. WEB UI 先上图感受一下spark webui 假设当前已经在本机运行standalone cluster模式,输入http://127.0.0.1:8080将会看到如下页面 driver application默认会打开4040端口进行http监听,可以看到application相关的详细信息 显示每个stage的详细信息

Apache Spark源码走读之12 -- Hive on Spark运行环境搭建

欢迎转载,转载请注明出处,徽沪一郎. 楔子 Hive是基于Hadoop的开源数据仓库工具,提供了类似于SQL的HiveQL语言,使得上层的数据分析人员不用知道太多MapReduce的知识就能对存储于Hdfs中的海量数据进行分析.由于这一特性而收到广泛的欢迎. Hive的整体框架中有一个重要的模块是执行模块,这一部分是用Hadoop中MapReduce计算框架来实现,因而在处理速度上不是非常令人满意.由于Spark出色的处理速度,有人已经成功将HiveQL的执行利用Spark来运行,这就是已经非常

Apache Spark源码走读之13 -- hiveql on spark实现详解

欢迎转载,转载请注明出处,徽沪一郎 概要 在新近发布的spark 1.0中新加了sql的模块,更为引人注意的是对hive中的hiveql也提供了良好的支持,作为一个源码分析控,了解一下spark是如何完成对hql的支持是一件非常有趣的事情. Hive简介 Hive的由来 以下部分摘自Hadoop definite guide中的Hive一章 "Hive由Facebook出品,其设计之初目的是让精通SQL技能的分析师能够对Facebook存放在HDFS上的大规模数据集进行分析和查询. Hive大大

Apache Spark源码走读之9 -- Spark源码编译

欢迎转载,转载请注明出处,徽沪一郎. 概要 本来源码编译没有什么可说的,对于java项目来说,只要会点maven或ant的简单命令,依葫芦画瓢,一下子就ok了.但到了Spark上面,事情似乎不这么简单,按照spark officical document上的来做,总会出现这样或那样的编译错误,让人懊恼不已. 今天闲来无事,又重试了一把,居然o了,做个记录,以备后用. 准备 我的编译机器上安装的Linux是archlinux,并安装后如下软件 scala 2.11 maven git 下载源码 第

Apache Spark源码走读之21 -- 浅谈mllib中线性回归的算法实现

欢迎转载,转载请注明出处,徽沪一郎. 概要 本文简要描述线性回归算法在Spark MLLib中的具体实现,涉及线性回归算法本身及线性回归并行处理的理论基础,然后对代码实现部分进行走读. 线性回归模型 机器学习算法是的主要目的是找到最能够对数据做出合理解释的模型,这个模型是假设函数,一步步的推导基本遵循这样的思路 假设函数 为了找到最好的假设函数,需要找到合理的评估标准,一般来说使用损失函数来做为评估标准 根据损失函数推出目标函数 现在问题转换成为如何找到目标函数的最优解,也就是目标函数的最优化