【大数据学习 | Spark-Core】RDD的概念与Spark任务的执行流程

1. RDD的设计背景

在实际应用中,存在许多迭代式计算,这些应用场景的共同之处是,不同计算阶段之间会重用中间结果,即一个阶段的输出结果会作为下一个阶段的输入。但是,目前的MapReduce框架都是把中间结果写入到HDFS中,带来了大量的数据复制、磁盘IO和序列化开销。显然,如果能将结果保存在内存当中,就可以大量减少IO。RDD就是为了满足这种需求而出现的,它提供了一个抽象的数据架构,我们不必担心底层数据的分布式特性,只需将具体的应用逻辑表达为一系列转换处理,不同RDD之间的转换操作形成依赖关系,可以实现管道化,从而避免了中间结果的落地存储,大大降低了数据复制、磁盘IO和序列化开销。

2. RDD的概念

RDD(Resilient Distributed Datasets,弹性分布式数据集)代表可并行操作元素不可变分区集合。

一个RDD就是一个分布式对象集合,本质上是一个只读的分区记录集合每个RDD可以分成多个分区,每个分区就是一个数据集片段(HDFS上的块),并且一个RDD的不同分区可以被保存到集群中不同的节点上,从而可以在集群中的不同节点上进行并行计算

RDD提供了一种高度受限的共享内存模型,即RDD是只读的记录分区的集合,不能直接修改,只能基于稳定的物理存储中的数据集来创建RDD,或者通过在其他RDD上执行确定的转换操作(如map、join和groupBy)而创建得到新的RDD。

RDD提供了一组丰富的操作以支持常见的数据运算,分为“行动”(Action)和“转换”(Transformation)两种类型,前者用于执行计算并指定输出的形式,后者指定RDD之间的相互依赖关系。两类操作的主要区别是,转换操作(比如map、filter、groupBy、join等)接受RDD并返回RDD,而行动操作(比如count、collect等)接受RDD但是返回非RDD(即输出一个值或结果)。

RDD典型的执行过程

Spark用Scala语言实现了RDD的API,程序员可以通过调用API实现对RDD的各种操作。RDD典型的执行过程如下:

1)RDD读入外部数据源(或者内存中的集合)进行创建;

2)RDD经过一系列的“转换”操作,每一次都会产生不同的RDD,供给下一个“转换”使用;

3)最后一个RDD经“行动”操作进行处理,并输出到外部数据源(或者变成Scala/JAVA集合或变量)。

需要说明的是,RDD采用了惰性调用,即在RDD的执行过程中,真正的计算发生在RDD的“行动”操作(行动算子底层代码调用了runJob函数),对于“行动”之前的所有“转换”操作,Spark只是记录下“转换”操作应用的一些基础数据集以及RDD生成的轨迹,即相互之间的依赖关系,而不会触发真正的计算。

a0844dd491e547fe8225702380a01647.png

5b2fc5dbf419459ba82dca1f98c68f5a.png

val conf = new SparkConf
val sparkContext = new SparkContext(conf)
val lines :RDD = sparkContext.textFile(logFile)
//lines.filter((a:String) => a.contains("hello world"))
val count = lines.filter(_.contains("hello world")).count()
println(count)

可以看出,一个Spark应用程序,基本是基于RDD的一系列计算操作。

第1行代码用于创建JavaSparkContext对象;

第2行代码从HDFS文件中读取数据创建一个RDD;

第3行代码对fileRDD进行转换操作得到一个新的RDD,即filterRDD;

count()是一个行动操作,用于计算一个RDD集合中包含的元素个数。

这个程序的执行过程如下:

1)创建这个Spark程序的执行上下文,即创建SparkContext对象;

2)从外部数据源(即HDFS文件)中读取数据创建fileRDD对象;

3)构建起fileRDD和filterRDD之间的依赖关系,形成DAG图,这时候并没有发生真正的计算,只是记录转换的轨迹;

4)执行action代码时,count()是一个行动类型的操作,触发真正的计算,开始执行从fileRDD到filterRDD的转换操作,并把结果持久化到内存中,最后计算出filterRDD中包含的元素个数。

3. spark任务的执行过程

每一个应用都是由driver端组成的,并且driver端可以解析用户的代码,并且在集群中并行执行,spark给大家提供了一个编程对象,它是一个抽象的,叫做弹性分布式数据集,这个数据集和一堆数据的集合并且是被分区的,因为分区的数据可以被并行的进行操作,rdd的创建方式有两种 1.读取hdfs的文件 2.在driver的一个集合可以转换为rdd,rdd可以被持久化到内存中,并且rdd可以实现更好的失败恢复容错。

711e8bdb7e874c55a214b920cf31d793.png

为什么rdd是抽象的呢?因为rdd并不存在数据,它是虚拟的,我们在定义逻辑的时候要标识一个节点,表示数据在流动到此处的时候要进行什么样的处理,我们可以理解rdd是一个代理对象。

686d9c3356ae479ba0f6b98e743da76e.png

上述任务执行过程可以划分为两个stage,从创建rdd开始到groupBy的shuffle,划分为一个stage,然后该shuffle到任务执行结束,又是一个stage。后面读源码我们会发现,当出现shuffle时,就要划分出一个阶段。因为业务逻辑发生了变化。

任务的执行和层架关系:

读取hdfs数据的时候映射应该是一个blk块对应一个分区

  • 在一个任务中,一个action算子会生成一个job。(行动算子的源码都会包含runJob函数)
  • 在一个job中存在shuffle算子,比如group sort切分阶段,shuffle+1个阶段。
  • shuffle是任务的划分的重点,前面的任务会将数据放入到自己的本地存储,后续的任务进行数据的拉取。
  • 在一个stage中任务都是管道形式执行的,避免了io,序列化和反序列化,这个就是dag切分的原理。
  • 在一个阶段中分区数量就是task任务的数量,task任务就是一堆非shuffle类算子的整体任务链。
  • 有几个分区就会并行的执行几个task任务。
  • 有几个分区是根据读取的文件来进行适配的,比如有三个blk那么就会生成三个分区,因为我们可以在每个分区中进行处理数据,实现本地化的处理,避免远程io。

我们知道,分区的个数与读取的文件的Split切片数量有关。假如textFile读取文件的大小为400M,则会被物理切分为3个block,因为每个block-size的大小最大为128M,block1为128M,block2为128M,block3为144M。默认逻辑切片split-size的大小与block-size相适配,为128M,所以有三个分区。三个分区就会并行的执行3个task任务。

spark中一个executor可以执行多个task任务。这是通过将executor配置为拥有多个cores来实现的。每个核心可以并行执行一个task。即executor是一个JVM进程,负责在节点上运行任务。可以为executor配置多个核心来并行处理多个任务。

如果分区数多于executor的核心数,某些task必须等待其他task任务完成才能开始执行。

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

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

相关文章

Flume日志采集系统的部署,实现flume负载均衡,flume故障恢复

目录 安装包 flume的部署 负载均衡测试 故障恢复 安装包 在这里给大家准备好了flume的安装包 通过网盘分享的文件:apache-flume-1.9.0-bin.tar.gz 链接: https://pan.baidu.com/s/1DXMA4PxdDtUQeMB4J62xoQ 提取码: euz7 --来自百度网盘超级会员v4的分享 ----…

centos7 安装helm v3

文章目录 1. 安装 Helm v3步骤 1:下载 Helm 安装包步骤 2:解压安装包步骤 3:将 Helm 移动到 /usr/local/bin步骤 4:验证安装 2. 使用 Helm 配置 Kubernetes步骤 1:安装并配置 kubectl步骤 2:初始化 Helm步骤…

Hive基础面试-如何理解复用率的

1. 模型的复用率你们是怎么做的? 简单直白的说就是你的模型复用率如何,在业务方是否认可该模型,也是衡量模型建设的一个标准,复用率数:数仓模型涉及的核心是追求模型的复用和共享,引用系数越高,…

eduSRC挖洞思路

声明 学习视频来自 B 站UP主泷羽sec,如涉及侵权马上删除文章。 笔记的只是方便各位师傅学习知识,以下网站只涉及学习内容,其他的都与本人无关,切莫逾越法律红线,否则后果自负。 ✍🏻作者简介:致…

Banana Pi BPI-CanMV-K230D-Zero 采用嘉楠科技 K230D RISC-V芯片设计

概述 Banana Pi BPI-CanMV-K230D-Zero 采用嘉楠科技 K230D RISC-V芯片设计,探索 RISC-V Vector1.0 的前沿技术,选择嘉楠科技的 Canmv K230D Zero 开发板。这款创新的开发板是由嘉楠科技与香蕉派开源社区联合设计研发,搭载了先进的勘智 K230D 芯片。 K230…

昆山网站建设在移动互联网时代的作用

在当今的移动互联网时代,昆山网站建设的重要性愈加凸显。随着智能手机和移动设备的普及,用户获取信息和进行消费的方式发生了根本性的变革。企业在此背景下,必须重新审视自己的在线形象和运营策略,以适应这一变化带来的挑战和机遇…

AI赋能 Python编程之2. 从构思到优化:用AI快速实现Python项目

AI赋能 Python编程 2. 从构思到优化:用AI快速实现Python项目 利用AI生成完整的Python项目,可以帮助程序员,尤其是初学者,高效完成从项目构思到实现的全过程。本文以开发一个简单的计算器应用为例,分享了如何借助AI的…

Spring ApplicationListener

【JavaWeb】Spring ApplicationListener-CSDN博客

MFC 下拉框显示问题和控件自适应窗口大小

在mfc开发过程中,遇到了下拉框的两个问题,现在记录一下 1、把下拉框点开显示不全我放进去的多条文本 解决办法:把鼠标放到下拉框的倒三角去点一下,鼠标就会变成双向箭头,或者先选中一下下拉框再把鼠标移动到倒三角上去…

【Python-办公自动化】实现自动化输出模板表格报告

import pandas as pd import numpy as np# 定义时间范围 date_range = pd.date_range(start=2023-11-01, end=2024-10-31, freq=M

规则引擎aviatorEvaluator注意点

规则引擎aviatorEvaluator注意点 官方文档&#xff1a;AviatorScript 编程指南&#xff08;5.0&#xff09; 语雀 1、依赖 <!--规则引擎--><dependency><groupId>com.googlecode.aviator</groupId><artifactId>aviator</artifactId><…

接上一主题,C++14中如何设计类似于std::any,使集合在C++中与Python一样支持任意数据?

这篇文章的重点是C多态的应用&#xff0c;但是如果你是C新手&#xff0c; 你需要了解以下C知识&#xff1a; 类 构造函数 拷贝构造函数 虚拟函数 纯虚拟函数 析构函数 类的继承 运算符重写 模板类 模板参数 数组 数组的传递 指针与动态内存分配 Python&#xff1a; s …

SpringBoot3与JUnit5集成测试

你可以在 Spring Boot 3 中轻松设置和运行 JUnit 集成测试。合理使用 Spring 提供的注解和工具&#xff0c;可以确保测试的高效性和可靠性。以下是集成测试的步骤和示例&#xff1a; 1. 添加依赖 在 pom.xml 中添加 Spring Boot Starter Test 依赖&#xff0c;它包含 JUnit 5 …

【系统架构设计师】真题论文: 论网络安全体系设计(包括解题思路和素材)

更多内容请见: 备考系统架构设计师-专栏介绍和目录 文章目录 真题题目(2014年 试题4)解题思路论文素材参考网络安全体系设计目标与原则网络安全体系的主要组成部分网络安全体系设计流程真题题目(2014年 试题4) 随着社会信息化的普及,计算机网络已经在各行各业得到了广泛…

5.STM32之通信接口《精讲》之USART通信---实验串口接收程序

根据上节&#xff0c;我们一已经完成了串口发送程序的代码&#xff0c;并且深入的解析探索了串口的原理&#xff0c;接下来&#xff0c;Whappy小编将带领大家进入串口接收程序的探索与实验&#xff0c;并将结合上一节串口发送一起来完成串口的发送和接收实验。 上来两张图 上图…

微服务即时通讯系统的实现(服务端)----(1)

目录 1. 项目介绍和服务器功能设计2. 基础工具安装3. gflags的安装与使用3.1 gflags的介绍3.2 gflags的安装3.3 gflags的认识3.4 gflags的使用 4. gtest的安装与使用4.1 gtest的介绍4.2 gtest的安装4.3 gtest的使用 5 Spdlog日志组件的安装与使用5.1 Spdlog的介绍5.2 Spdlog的安…

macos 使用 nvm 管理 node 并自定义安装目录

系统环境&#xff1a;MacOS Version 参考文章&#xff1a; Github 地址&#xff1a;https://github.com/nvm-sh/nvm 安装的方式是很简单的&#xff0c;直接执行下面的命令即可&#xff1a; curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.40.0/install.sh | bas…

数字IC后端实现时钟树综合系列教程 | Clock Tree,Clock Skew Group之间的区别和联系

Q: Clock&#xff0c;Clock Tree和Skew Group有何区别&#xff1f;Innovus CCOPT引擎是如何使用这些的&#xff1f; Clock是时序约束SDC中的时钟定义点。 create_clock -name clk_osc -period $period_24m [get_ports xin_osc0_func] 时钟树综合(Clock Tree Synthesis)之前应…

tcpdump抓包 wireShark

TCPdump抓包工具介绍 TCPdump&#xff0c;全称dump the traffic on anetwork&#xff0c;是一个运行在linux平台可以根据使用者需求对网络上传输的数据包进行捕获的抓包工具。 tcpdump可以支持的功能: 1、在Linux平台将网络中传输的数据包全部捕获过来进行分析 2、支持网络层…

安装支持ssl的harbor 2.1.4 docker 19.03.8 docker-compose 1.24.0

版本&#xff1a; docker 19.03.8 docker-compose 1.24.0 harbor: harbor-offline-installer-v2.1.4.tgz 1、先在/root下生成证书 #生成ca根证书 openssl genrsa -out ca.key 4096#创建ca证书 openssl req -x509 -new -nodes -sha512 -days 3650 \-subj "/CCN/STShenzh…