Spark Executor解析

目录

 

1、Spark Executor如何工作

2、Spark Executor工作源码


1、Spark Executor如何工作

当Driver发送过来Task的时候,其实是发送给CoarseGrainedExecutorBackend这个RPCEndpoint,而不是直接发送给Executor(Executor由于不是消息循环体永远也无法接收远程发过来的信息)。

Driver向ExecutorBackend发送LaunchTask,这里实际上是把内容交给线程池中的线程去执行。首先判断Executor是否为空,反序列化TaskDescription,然后调用Executor.launchTask ,launchTask 里面是将Task封装在TaskRunner(是一个runnable对象)里面然后交给线程池中的线程处理。在TaskRunner的run方法里面会导致runTask的执行。

2、Spark Executor工作源码

(1)Worker接收到Master发送过来启动Executor的指令,通过ExecutorRunner启动另外一个进程来运行Executor。在这个基础上会启动一个CoarseGrainedExecutorBackend(粗粒度),ExecutorBackend启动的时候这个时候要向Driver注册,通过RegisterExecutor。Driver(CoarseGrainedSchedulerBackend的内部成员DriverEndpoint接收注册)在接收到RegisterExecutor信息后会返回一个信息RegisteredExecutor给CoarseGrainedExecutorBackend。

CoarseGrainedExecutorBackend:是Executor运行所在的进程的名称,他本身不会完成具体任务的计算,这个进程里面有个Executor对象和CoarseGrainedExecutorBackend是一一对应的。

Executor:完成具体的计算,真正处理Task的对象,内部通过线程池的方式完成Task的计算。

(2)Driver的CoarseGrainedSchedulerBackend内部的成员DriverEndpoint收到RegisterExecutor后,首先判断ExecutorDataMap中是否有这个ExecutorId。ExecutorDataMap是个内存数据结构(他是CoarseGrainedSchedulerBackend的成员,所以最终是注册给CoarseGrainedSchedulerBackend),他是一个HashMap。

如果ExecutorDataMap中不存在,就进行注册,注册的时候先看看address存在不存在,如果存在直接获取,如果不存在就获取senderAddress。这从实现的角度看adress就是senderAddress。然后相关数据结构添加数据

CoarseGrainedSchedulerBackend.this.synchronized这里加上synchronized是因为集群中有很多ExecutorBackend向Driver注册,担心注册的时候写冲突,所以加上一个同步代码块

最后把消息发还给CoarseGrainedExecutorBackend

(3)CoarseGrainedExecutorBackend收到RegisterExecutor后就new了一个Executor这个Executor对象是事实上负责Task计算的

(4)在Executor默认构造器中有一个非常关键的内容,有个成员threadPool(线程池)

创建线程池,线程池里面要有线程,线程怎么产生的呢?不会平白无故的产生,所以就搞了一个线程工厂,就是按照某种你需要的格式去产生线程,背后还是new出一个线程

setDamon(True)设置每个线程是后台运行的方式

(5)线程池也准备好后就是等待Driver端发任务过来,是发给CoarseGrainedExecutorBackend不是Executor,因为Executor不可能接收到消息的他本身就不是一个消息循环体。CoarseGrainedExecutorBackend收到LaunchTask后,这里实际上是把内容交给线程池中的线程去执行。判断Executor是否为空,反序列化TaskDescription,然后调用Executor.launchTask

(6)launchTask 里面是将Task封装在TaskRunner(是一个runnable对象)里面然后交给线程池中的线程处理。并把这个任务加入ConcurrentHashMap类型的名称为runningTasks的数据结构中管理。

(7)在TaskRunner的run方法里面会导致runTask的执行

补充:为什么要在worker接收到master发送过来的指令后为什么要启动另外一个进程,也就是说为什么开辟另外一个进程,在另外一个进程中注册给Driver,然后启动Executor。必须启动另外一个进程的原因:①Worker本身是管理当前机器上的资源的,当前机器上的资源变动的时候要汇报给Master,Worker不是用来做计算的不能在Worker里面计算;②Spark集群中可能有很多应用程序就可能有很多的Executor,如果不是为每个Executor启动一个进程而是所有Executor在Worker里面,那么一个程序奔溃了会导致其他程序奔溃。

 

 

 

 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/473637.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Cassandra1.2文档学习(12)—— hint机制

参考文档:http://www.datastax.com/documentation/cassandra/1.2/webhelp/index.html#cassandra/dml/dml_about_hh_c.html Hint机制是Cassandra的特性当一致性不要求时保证了写入的高可用性。但临时故障发生如网络问题,Hint机制显著地提升了反应的一致性。通过配置c…

LeetCode 381. O(1) 时间插入、删除和获取随机元素 - 允许重复(vector + 哈希)

文章目录1. 题目2. 解题1. 题目 设计一个支持在平均 时间复杂度 O(1) 下, 执行以下操作的数据结构。 注意: 允许出现重复元素。 insert(val):向集合中插入元素 val。remove(val):当 val 存在时,从集合中移除一个 val。getRando…

Stage划分和Task最佳位置

目录 1、Job Stage划分 2、Task最佳位置 3、总结 3.1 Stage划分总结: 3.2 Task最佳位置总结: 1、Job Stage划分 Spark Application中因为不同的Action触发众多的Job,也就是说一个Application中可以有很多的Job,每个Job是由是…

LwIP编译方法以及选项说明

条件编译命令作用说明IP_SOF_BROADCAST LWIP_IGMP 转载于:https://www.cnblogs.com/motadou/p/3537581.html

LeetCode 1636. 按照频率将数组升序排序(哈希+排序)

文章目录1. 题目2. 解题1. 题目 给你一个整数数组 nums ,请你将数组按照每个值的频率 升序 排序。 如果有多个值的频率相同,请你按照数值本身将它们 降序 排序。 请你返回排序后的数组。 示例 1: 输入:nums [1,1,2,2,2,3] 输出…

Ubuntu16.04直接修改root密码

Ubuntu16.04系统中,因为误操作导致管理员密码丢失或无效,并且忘记root密码,此时无法进行任何root/sudo权限操作。可以通过GRUB重新设置root密码,并恢复管理员账户到正常状态。 工具/原料 • Ubuntu16.04操作系统 重置root密码 启…

TaskScheduler相关

目录 1、DAGScheduler与TaskScheduler 2、TaskScheduler与SchedulerBackend 3、任务调度过程总结 1、DAGScheduler与TaskScheduler DAGScheduler面向我们整个Job划分出了Stage,划分了Stage是从后往前划分的,执行的时候是从前往后,每个Stag…

网上支付心案例payment

案例的作用及用法参考该类的说明注释 工具类分别有: package cn.itcast.utils; import java.util.Properties; /** * 读取配置文件 * */ public class ConfigInfo { private static Properties cache new Properties(); static{ try …

LeetCode 1637. 两点之间不包含任何点的最宽垂直面积

文章目录1. 题目2. 解题1. 题目 给你 n 个二维平面上的点 points ,其中 points[i] [xi, yi] ,请你返回两点之间内部不包含任何点的 最宽垂直面积 的宽度。 垂直面积 的定义是固定宽度,而 y 轴上无限延伸的一块区域(也就是高度为…

有两个序列a,b,大小都为n,序列元素的值任意整数,无序;要求:通过交换a,b中的元素,使[序列a元素的和

原题:有一序列a,大小为n,分为2部分,序列元素的值任意整形数,无序; 要求:通过交换a,b中的元素,使[序列a元素的和]与[序列b元素的和]之间的差最小,用python写。 #codingutf-8 # 1.将两序列合并为一…

Task执行流程

1、源码走读 (1)当Driver中的SchedulerBackend(Standalone模式为CoarseGrainedSchedulerBackend)给ExecutorBackend(Standalone模式为CoarseGrainedExecutorBackend)发送LaunchTask之后,Coarse…

一段挂起进程中所有线程的代码

今天看书核心编程看到第7章,中的一段代码很有意思,win7下对记事本进程进行测试,可以挂起,挺有意思的 //windows核心编程 第5版中的一段代码 /* 函数功能:挂起进程中的所有线程 参数1:进程ID 参数2:若为TRUE时对进程中的所有线程调…

LeetCode 1638. 统计只差一个字符的子串数目(DP)

文章目录1. 题目2. 解题2.1 暴力枚举2.2 DP1. 题目 给你两个字符串 s 和 t ,请你找出 s 中的非空子串的数目,这些子串满足替换 一个不同字符 以后,是 t 串的子串。 换言之,请你找到 s 和 t 串中 恰好 只有一个字符不同的子字符串…

Airflow简介

1、什么是Airflow Airflow 是一个 Airbnb 的 Workflow 开源项目,使用Python编写实现的任务管理、调度、监控工作流平台。Airflow 是基于DAG(有向无环图)的任务管理系统,可以简单理解为是高级版的crontab,但是它解决了crontab无法解决的任务…

Flask简介与简单项目操作流程

Flask框架简介Flask诞生于2010年,是Armin ronacher(人名)用Python语言基于Werkzeug工具箱编写的轻量级Web开发框架。它主要面向需求简单的小应用。Flask本身相当于一个内核,其他几乎所有的功能都要用到扩展(邮件扩展Fl…

LeetCode 1640. 能否连接形成数组(哈希)

文章目录1. 题目2. 解题1. 题目 给你一个整数数组 arr ,数组中的每个整数 互不相同 。 另有一个由整数数组构成的数组 pieces,其中的整数也 互不相同 。请你以 任意顺序 连接 pieces 中的数组以形成 arr 。但是,不允许 对每个数组 pieces[i]…

python基本知识、数据库、网络、编程等总结

Python语言特性 1 Python的函数参数传递 看两个例子: a 1 def fun(a):a 2 fun(a) print a # 1 a [] def fun(a):a.append(1) fun(a) print a # [1] 所有的变量都可以理解是内存中一个对象的“引用”,或者,也可以看似c中void*的感觉。 通过id来看引…

Android中的Handler的具体用法

Android UI操作并不是线程安全的并且这些操作必须在UI线程中执行。Android利用Handler来实现UI线程的更新的。 Handler是Android中的消息发送器,其在哪个Activity中创建就属于且紧紧属于该Activity。还可以说其在哪个线程中new的,就是那个线程的Handler。…

LeetCode 1641. 统计字典序元音字符串的数目(DP)

文章目录1. 题目2. 解题1. 题目 给你一个整数 n,请返回长度为 n 、仅由元音 (a, e, i, o, u) 组成且按 字典序排列 的字符串数量。 字符串 s 按 字典序排列 需要满足:对于所有有效的 i,s[i] 在字母表中的位置总是与 s[i1] 相同或在 s[i1] 之…

mysql5.5中的MaxValue关键字

MaxValue是MySQL5.5中的关键字转载于:https://www.cnblogs.com/Neil223/p/3544271.html