Spark的内核调度

这篇具有很好参考价值的文章主要介绍了Spark的内核调度。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

概述

RDD的依赖

 DAG和Stage

 DAG执行流程图形成和Stage划分

 Stage内部流程

Spark Shuffle

Spark中shuffle的发展历程

优化前的Hash shuffle

 经过优化后的Hash shuffle

 Sort shuffle

Sort shuffle的普通机制

Job调度流程

Spark RDD并行度


概述

Spark内核调度任务:

1.构建DAG有向无环图

2.划分stage夹断

3.Driver底层的运转

4.分区的划分(线程)

的Spark内核调度的目的:尽可能用最少的资源高效地完成任务计算

RDD的依赖

RDD的依赖:一个RDD的形成可能由一个或者多个RDD得到的,此时这个RDD和之前的RDD之间产生依赖关系

Spark中,RDD之间的依赖关系,只要有两种类型:宽依赖和窄依赖

窄依赖:

作用:能够让Spark程序并行计算,也就是一个分区数据计算出现问题的时候,其它分区不受影响

特点:父RDD的分区和子RDD的分区是一对一关系,也就是父RDD分区的数据会整个被下游子RDD的分区接收

Spark的内核调度,spark,大数据,分布式

宽依赖:

作用:划分stage的重要依据,宽依赖也叫shuffle依赖

特点:父RDD的分区和子RDD的分区关系是一对多的关系,也就是父RDD的分区数据会被划成多份给到下游子RDD的多个分区做接收

注意:如果有宽依赖,shuffle下游的其他操作,必须等待shuffle执行完成以后才能够继续执行,为了避免数据的不完整

算子中一般以ByKey结尾的会发生shuffle;另外是重分区算子会发生shuffle

Spark的内核调度,spark,大数据,分布式

 DAG和Stage

DAG:有向无环图,只要描述一段执行任务,从开始一直往下走,不允许出现回调操作

Spark应用程序中,遇到一个Action算子,就会触发一个JOB任务的产生

对于每个JOB的任务,都会产生一个DAG执行流程图,流程图的形成的层级关系如下:

层级关系:

1.一个spark应用程序→遇到一个Action算子,就会触发形成一个JOB任务

2.一个JOB任务只有一个DAG有向无环图

3.一个DAG有向无环图→有多个stage

4.一个stage→有多个Task线程

5.一个RDD→有多个分区

6.一个分区会被一个Task线程所处理

 DAG执行流程图形成和Stage划分

Spark的内核调度,spark,大数据,分布式

 1.spark应用程序遇到Action算子后,就会触发一个JOB任务的产生,JOB任务就会将它所依赖的算子全部加载进来,形成一个stage

2.接着从action算子从后往前回溯,遇到窄依赖就将算子放在同一个stage中,如果遇到宽依赖,就划分形成新的stage,最后一直到回溯完成

 Stage内部流程

Spark的内核调度,spark,大数据,分布式

 默认并行度值的确认:

1.使用textFile读取HDFS上的文件,因此RDD分区数=max(文件的block块数量,defaultminpartition),继续需要知道defaultminpartition的值是多少

2.defaultminpartition=min(spark.default.parallelism,2)取最小值,最终确认spark.default.parallelism的参数值就能最终确认RDD的分区数有多少个

spark.default.parallelism参数值的确认:

1.如果有父RDD,就取父RDD的最大分区数

2.如果没有父RDD,根据集群模式进行取值

        本地模式:机器的最大cpu核数

        Mesos:默认是8

        其它模式:所有执行节点上的核总数或2,以较大者为准

Spark Shuffle

Spark中shuffle的发展历程

1- 在1.1版本以前,Spark采用Hash shuffle (优化前 和 优化后)

2- 在1.1版本的时候,Spark推出了Sort Shuffle

3- 在1.5版本的时候,Spark引入钨丝计划(优化为主)

4- 在1.6版本的时候,将钨丝计划合并到sortShuffle中

5- 在2.0版本的时候,将Hash Shuffle移除,将Hash shuffle方案移植到Sort Shuffle

优化前的Hash shuffle

Spark的内核调度,spark,大数据,分布式

 存在的问题:

        上游(map端)的每个Task会产生与下游Task个数相等的小文件个数,导致上游有非常多的小文件,下游(reduce端)来拉取文件的时候,会有大量的网络IO和磁盘IO过程,因为要打开和读取多个小文件

 经过优化后的Hash shuffle

Spark的内核调度,spark,大数据,分布式

优化后的Hash shuffle:

变成了由每个Executor进程产生与下游Task个数相等的小文件数,这样可以大量减少小文件的产生,以及降低下游拉取文件时候的网络IO和磁盘IO过程

 Sort shuffle

Spark的内核调度,spark,大数据,分布式

 Sort shuffle分成了两种:普通机制和bypass机制,具体使用哪种由spark底层决定

Sort shuffle的普通机制

 普通机制的运行过程:

每个上游task线程处理数据,数据处理完以后,先放在内存中,接着对内存中的数据进行分区,排序,将内存中的数据溢写到磁盘,形成一个个小文件,溢写完成后,将多个小文件合并成一个大的磁盘文件,并且针对每个大的磁盘文件,提供一个索引文件,接着是下游Task根据索引文件来读取相应的数据

Sort shuffle的bypass机制 

bypass机制 :就是在普通机制的基础上,省略了排序的过程

bypass机制的触发条件:

1.上游的RDD数量不能超过100个

2.上游不能对数据进行提前聚合操作(因为提前聚合,需要先进行分组操作,而分组的操作实际上是有排序的操作)

Job调度流程

Spark的内核调度,spark,大数据,分布式

主要是讨论:在Driver内部,是如何调度任务

1.Driver进程启动后,底层PY4J创建SparkContext顶级对象,在创建该对象的进程中,还会创建另外两个对象,分别是:DAGScheduler和TaskScheduler

        DAGScheduler:DAG调度器,将Job任务形成DAG有向无环图和划分Stage的阶段

        TaskScheduler:Task调度器,将Task线程分配给到具体的Executor执行

2.一个saprk程序遇到一个action算子触发产生一个job任务,SparkContext将job任务给到DAG调度器,拿到job任务后,会将job任务形成有向无环图和划分stage阶段,并且确定每个stage有多少个Task线程,会将众多的Task线程放到TaskSet的集合中,DAG调度器将TaskSet集合给到Task调度器

3.Task调度器拿到TaskSet集合以后,将Task分配给到具体的Executor执行,底层是基于SchedulerBackend调度队列来实现的

4.Executor开始执行任务,并且Driver会监控各个Executor的执行状态,知道所有的Executor执行完成,就认为任务运行结束

5.Driver通知Namenote释放资源

Spark RDD并行度

整个Spark应用中,影响并行度的因素有以下两个原因:

        1.资源的并行度:Executor数量和CPU核数以及内存的大小

        2.数据的并行度:Task的线程和分区数量

一般将Task想层数量设置为CPU核数的2-3被,另外每个线程分配3-5GB的内存资源

说明: spark.default.parallelism该参数是SparkCore中的参数。该参数只会影响shuffle以后的分区数量。另外该参数对parallelize并行化本地集合创建的RDD不起作用。

import time

from pyspark import SparkConf, SparkContext
import os

# 绑定指定的Python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    print("Spark入门案例: WordCount词频统计")

    # 1- 创建SparkContext对象
    conf = SparkConf()\
        .set("spark.default.parallelism", "5")\
        .setAppName('spark_wordcount_demo')\
        .setMaster('local[*]')

    # 设置并行度参数方式一
    # conf.set("spark.default.parallelism", "4")

    sc = SparkContext(conf=conf)

    # 2- 数据输入
    init_rdd = sc.textFile("file:///export/data/gz16_pyspark/01_spark_core/data/content.txt")

    # 3- 数据处理
    flatmap_rdd = init_rdd.flatMap(lambda line: line.split(" "))

    map_rdd = flatmap_rdd.map(lambda word: (word,1))

    # shuffle前分区数
    print("shuffle前分区数",map_rdd.getNumPartitions())

    result = map_rdd.reduceByKey(lambda agg,curr: agg+curr)

    # shuffle后分区数
    print("shuffle后分区数", result.getNumPartitions())

    # 4- 数据输出
    print(result.collect())

    # 5- 释放资源
    sc.stop()

通过parallelize构建得到RDD的分区情况(了解):文章来源地址https://www.toymoban.com/news/detail-797781.html

from pyspark import SparkConf, SparkContext
import os

# 绑定指定的Python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    print("并行化本地集合创建RDD")

    # 1- 创建SparkContext对象
    conf = SparkConf().setAppName('parallelize_rdd').setMaster('local[1]')

    # 设置并行度参数
    conf.set("spark.default.parallelism", 4)

    sc = SparkContext(conf=conf)

    # 2- 数据输入
    # 并行化本地集合得到RDD
    init_rdd = sc.parallelize([1,2,3,4,5])

    # shuffle前分区数
    print("分区数", init_rdd.getNumPartitions())

    # 3- 数据处理
    # 4- 数据输出
    # 获取分区数
    print(init_rdd.getNumPartitions())

    # 获取具体分区内容
    print(init_rdd.glom().collect())


    # 5- 释放资源
    sc.stop()

到了这里,关于Spark的内核调度的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 大数据开源框架环境搭建(七)——Spark完全分布式集群的安装部署

    前言:七八九用于Spark的编程实验 大数据开源框架之基于Spark的气象数据处理与分析_木子一个Lee的博客-CSDN博客_spark舆情分析 目录 实验环境: 实验步骤: 一、解压 二、配置环境变量:  三、修改配置文件  1.修改spark-env.sh配置文件: 2.修改配置文件slaves: 3.分发配置文件:

    2024年02月11日
    浏览(54)
  • 云计算与大数据第16章 分布式内存计算平台Spark习题

    1、Spark是Hadoop生态(  B  )组件的替代方案。 A. Hadoop     B. MapReduce        C. Yarn             D.HDFS 2、以下(  D  )不是Spark的主要组件。 A. Driver      B. SparkContext       C. ClusterManager D. ResourceManager 3、Spark中的Executor是(  A  )。 A.执行器      B.主节

    2024年02月14日
    浏览(119)
  • 分布式计算中的大数据处理:Hadoop与Spark的性能优化

    大数据处理是现代计算机科学的一个重要领域,它涉及到处理海量数据的技术和方法。随着互联网的发展,数据的规模不断增长,传统的计算方法已经无法满足需求。因此,分布式计算技术逐渐成为了主流。 Hadoop和Spark是目前最为流行的分布式计算框架之一,它们都提供了高

    2024年01月23日
    浏览(57)
  • 分布式计算框架:Spark、Dask、Ray 分布式计算哪家强:Spark、Dask、Ray

    目录 什么是分布式计算 分布式计算哪家强:Spark、Dask、Ray 2 选择正确的框架 2.1 Spark 2.2 Dask 2.3 Ray 分布式计算是一种计算方法,和集中式计算是相对的。 随着计算技术的发展, 有些应用需要非常巨大的计算能力才能完成,如果采用集中式计算,需要耗费相当长的时间来完成

    2024年02月11日
    浏览(70)
  • 数据存储和分布式计算的实际应用:如何使用Spark和Flink进行数据处理和分析

    作为一名人工智能专家,程序员和软件架构师,我经常涉及到数据处理和分析。在当前大数据和云计算的时代,分布式计算已经成为了一个重要的技术方向。Spark和Flink是当前比较流行的分布式计算框架,它们提供了强大的分布式计算和数据分析功能,为数据处理和分析提供了

    2024年02月16日
    浏览(62)
  • Spark单机伪分布式环境搭建、完全分布式环境搭建、Spark-on-yarn模式搭建

    搭建Spark需要先配置好scala环境。三种Spark环境搭建互不关联,都是从零开始搭建。 如果将文章中的配置文件修改内容复制粘贴的话,所有配置文件添加的内容后面的注释记得删除,可能会报错。保险一点删除最好。 上传安装包解压并重命名 rz上传 如果没有安装rz可以使用命

    2024年02月06日
    浏览(83)
  • 【Spark分布式内存计算框架——Spark 基础环境】1. Spark框架概述

    第一章 说明 整个Spark 框架分为如下7个部分,总的来说分为Spark 基础环境、Spark 离线分析和Spark实时分析三个大的方面,如下图所示: 第一方面、Spark 基础环境 主要讲述Spark框架安装部署及开发运行,如何在本地模式和集群模式运行,使用spark-shell及IDEA开发应用程序,测试及

    2024年02月11日
    浏览(67)
  • spark分布式解压工具

    ​ spark解压缩工具,目前支持tar、gz、zip、bz2、7z压缩格式,默认解压到当前路下,也支持自定义的解压输出路径。另外支持多种提交模式,进行解压任务,可通过自定义配置文件,作为spark任务的资源设定 2.1 使用hadoop的FileSystem类,对tos文件的进行读取、查找、写入等操作

    2024年02月02日
    浏览(48)
  • Spark分布式内存计算框架

    目录 一、Spark简介 (一)定义 (二)Spark和MapReduce区别 (三)Spark历史 (四)Spark特点 二、Spark生态系统 三、Spark运行架构 (一)基本概念 (二)架构设计 (三)Spark运行基本流程 四、Spark编程模型 (一)核心数据结构RDD (二)RDD上的操作 (三)RDD的特性 (四)RDD 的持

    2024年02月04日
    浏览(68)
  • 分布式计算MapReduce | Spark实验

    题目1 输入文件为学生成绩信息,包含了必修课与选修课成绩,格式如下: 班级1, 姓名1, 科目1, 必修, 成绩1 br (注: br 为换行符) 班级2, 姓名2, 科目1, 必修, 成绩2 br 班级1, 姓名1, 科目2, 选修, 成绩3 br ………., ………, ………, ………, ……… br 编写两个Hadoop平台上的MapRed

    2024年02月08日
    浏览(62)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包