Spark RDD简介与运行机制概述
RDD工作原理:
主要分为三部分:创建RDD对象,DAG调度器创建执行计划,Task调度器分配任务并调度Worker开始运行。
SparkContext(RDD相关操作)→通过(提交作业)→(遍历RDD拆分stage→生成作业)DAGScheduler→通过(提交任务集)→任务调度管理(TaskScheduler)→通过(按照资源获取任务)→任务调度管理(TaskSetManager)
举例:以下面一个按 A-Z 首字母分类,查找相同首字母下不同姓名总个数的例子来看一下 RDD 是如何运行起来的。
步骤 1 :创建 RDD 。 上面的例子除去最后一个 collect 是个动作,不会创建 RDD 之外,前面四个转换都会创建出新的 RDD 。因此第一步就是创建好所有 RDD( 内部的五项信息 ) 。
步骤 2 :创建执行计划。 Spark 会尽可能地管道化,并基于是否要重新组织数据来划分 阶段 (stage) ,例如本例中的 groupBy() 转换就会将整个执行计划划分成两阶段执行。最终会产生一个 DAG(directed acyclic graph ,有向无环图 ) 作为逻辑执行计划。
步骤 3 :调度任务。 将各阶段划分成不同的 任务 (task) ,每个任务都是数据和计算的合体。在进行下一阶段前,当前阶段的所有任务都要执行完成。因为下一阶段的第一个转换一定是重新组织数据的,所以必须等当前阶段所有结果数据都计算出来了才能继续。
假设本例中的 hdfs://names 下有四个文件块,那么 HadoopRDD 中 partitions 就会有四个分区对应这四个块数据,同时 preferedLocations 会指明这四个块的最佳位置。现在,就可以创建出四个任务,并调度到合适的集群结点上。
Task管理和序列化:
Task的运行要解决的问题不外乎就是如何以正确的顺序,有效地管理和分派任务,如何将Task及运行所需相关数据有效地发送到远端,以及收集运行结果
Task的派发源起于DAGScheduler调用TaskScheduler.submitTasks将一个Stage相关的一组Task一起提交调度。
在TaskSchedulerImpl中,这一组Task被交给一个新的TaskSetManager实例进行管理,所有的TaskSetManager经由SchedulableBuilder根据特定的调度策略进行排序,TaskSchedulerImpl的resourceOffers函数中,当前被选择的TaskSetManager的ResourceOffer函数被调用并返回包含了序列化任务数据的TaskDescription,最后这些TaskDescription再由SchedulerBackend派发到ExecutorBackend去执行
系列化的过程中,上一节中所述App依赖文件相关属性URL等通过DataOutPutStream写出,而Task本身通过可配置的Serializer来序列化,当前可配制的Serializer包括如JavaSerializer ,KryoSerializer等
Task的运行结果在Executor端被序列化并发送回SchedulerBackend,由于受到Akka Frame Size尺寸的限制,如果运行结果数据过大,结果会存储到BlockManager中,这时候发送到SchedulerBackend的是对应数据的BlockID,TaskScheduler最终会调用TaskResultGetter在线程池中以异步的方式读取结果,TaskSetManager再根据运行结果更新任务状态(比如失败重试等)并汇报给DAGScheduler等
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Spark配置参数调优
1.配置多个executor 在项目中,由于数据量为几百万甚至千万级别,如果一个executor装载的对象过多,会导致GC很慢。项目中,我们使一个worker节点执行app时启动多个executor,从而加大并发度,解决full GC慢的问题。同时,由于启动了多个exeucute,在内存与核数不变的情况下,需要调整分配给每个execute的内存数及核数。 2.配置数据序列化 Spark默认序列化方式为Java的ObjectOutputStream序列化一个对象,速度较慢,序列化产生的结果有时也比较大。所以项目中我们使用kryo序列化方式,通过kryo序列化,使产生的结果更为紧凑,减少内存的占用空间,同时减少了对象本身的元数据信息与基本数据类型的开销,从而更好地提高了性能。 3.优化缓存大小 Spark默认用于缓存RDD的空间为一个executor的60%,项目中由于考虑到标签数量为成百个,使用同样规则与数量的标签进行客户群探索及客户群生成的概率很小。所以修改spark.storage.memoryFaction=0.4,这样使百分之60%的内存空间可以在task执行过程中缓存创建...
- 下一篇
Spark运行流程概述
Application 指用户编写的Spark应用程序,其中包含了一个Driver功能的代码和分布在集群中多个节点上运行的Executor代码。 Driver Spark中的Driver即运行上述Application的main()函数并创建SparkContext.创建的目的是为了初始化Spark的运行环境。SparkContext负责与ClusterManager通信,进行资源的申请、任务的分配和监控等。当Executor部分运行完毕后,Driver同时负责将SparkContext关闭,通常用SparkContext代表Driver. Executor 某个Application运行在Worker节点上的一个进程,该进程负责运行某些Task,并且负责将数据存在内存或者磁盘上。每个Application都有各自独立的一批Executor.在spark on Yarnm模式下,名为CoarseGrainedExecutor Backend进程有且只有一个executor,它负责将task包装秤taskRunner,并从线程池中抽取出一个空闲线程进行Task.这样 每个CoarseGr...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- Linux系统CentOS6、CentOS7手动修改IP地址
- SpringBoot2初体验,简单认识spring boot2并且搭建基础工程
- SpringBoot2全家桶,快速入门学习开发网站教程
- CentOS8编译安装MySQL8.0.19
- CentOS7,CentOS8安装Elasticsearch6.8.6
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Red5直播服务器,属于Java语言的直播服务器
- CentOS6,CentOS7官方镜像安装Oracle11G
- Windows10,CentOS7,CentOS8安装Nodejs环境
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池