SparkCore(11):RDD概念和创建RDD两种方法,以及RDD的Partitions以及并行度理解

本文主要是介绍SparkCore(11):RDD概念和创建RDD两种方法,以及RDD的Partitions以及并行度理解,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

一、RDD概念

1.概念

Resilient Distributed Datasets弹性分布式数据集,默认情况下:每一个block对应一个分区,一个分区会开启一个task来处理。

(a)Resilient:可以存在给定不同数目的分区、数据缓存的时候可以缓存一部分数据也可以缓存全部数据
(b)Distributed:分区可以分布到不同的executor执行(也就是不同的worker/NM上执行)
(c)Datasets:内部存储是数据

2.特性

(1)是一系列的分片,分区
(2)每个分片有一个方法来做计算
(3)rdd会有依赖其他rdd的操作,可以通过wordCountRDD.toDebugString来查看
(4)(可选项)如果rdd是二元组,就会存在分区器(默认是hashpartition)
(5)(可选项)最佳位置。数据在哪台机器上,任务就启在哪个机器上,数据在本地上,不用走网络。不过数据进行最后汇总的时候就要走网络。(hdfs file的block块)

二、RDD创建方法

1.外部数据

val path = "hdfs://192.168.31.3:8020/page_views.data"
val originalRdd: RDD[String] = sc.textFile(path)

2.内存中数据:基于序列化进行创建

scala> val seq = List(1,2,3,4,5,6,7)
seq: List[Int] = List(1, 2, 3, 4, 5, 6, 7)scala> val rdd2 = sc.parallelize(seq)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:29

三、关键:Partitions以及并行度

1.RDD的partitions数目

(1)读取数据阶段,对于textFile来说,没有在方法中的指定分区数,则默认为min(defaultParallelism,2),而defaultParallelism对应的就是spark.default.parallelism。如果是从hdfs上面读取文件,其分区数为文件block数(128MB/block)
(2)在Map阶段partition数目保持不变。
(3)在Reduce阶段,RDD的聚合会触发shuffle操作,聚合后的RDD的partition数目跟具体操作有关,例如repartition操作会聚合成指定分区数,还有一些算子是可配置的。

 2.并行度

定义:一个job一次所能执行的task数目,即一个job对应的总的core资源个数

执行一个job的task的并行数 = job的Executor数目 * 每个Executor的core个数。
例如提交scalaProjectMaven.jar的spark任务

date=`date +"%Y%m%d%H%M"`
/opt/modules/spark-2.1.0-bin-2.7.3/bin/spark-submit \
--master yarn \
--deploy-mode client \
--name spark_shell_${date} \
--class wordcount \
--driver-memory   8G \
--driver-cores 4 \    
--executor-memory 4G \
--executor-cores 4 \
--num-executors 3 \
--conf spark.app.coalesce=1 \
/opt/datas/scalaProjectMaven.jar

提交job的并行数=3*4=12,即每一个批次执行12个task,对应12个partitions。

3.partitions和并行度关系

一个partition对应一个要做的task,一个executor的core执行一个task
Tasks(一个RDD的总task数)=该RDD的partitions
Doing(并行执行任务数)= job的Executor数目 * 每个Executor核心数
总共执行批次=Tasks/Doing  (如果不是整除,则加1)

例如:Tasks=50,Doing=30,则执行两次,第一次执行30个task,第二次执行20个task。

4.参考

(1)https://blog.csdn.net/yu0_zhang0/article/details/80454517

 

这篇关于SparkCore(11):RDD概念和创建RDD两种方法,以及RDD的Partitions以及并行度理解的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


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

相关文章

Python错误AttributeError: 'NoneType' object has no attribute问题的彻底解决方法

《Python错误AttributeError:NoneTypeobjecthasnoattribute问题的彻底解决方法》在Python项目开发和调试过程中,经常会碰到这样一个异常信息... 目录问题背景与概述错误解读:AttributeError: 'NoneType' object has no at

postgresql使用UUID函数的方法

《postgresql使用UUID函数的方法》本文给大家介绍postgresql使用UUID函数的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录PostgreSQL有两种生成uuid的方法。可以先通过sql查看是否已安装扩展函数,和可以安装的扩展函数

Java中Arrays类和Collections类常用方法示例详解

《Java中Arrays类和Collections类常用方法示例详解》本文总结了Java中Arrays和Collections类的常用方法,涵盖数组填充、排序、搜索、复制、列表转换等操作,帮助开发者高... 目录Arrays.fill()相关用法Arrays.toString()Arrays.sort()A

Nginx安全防护的多种方法

《Nginx安全防护的多种方法》在生产环境中,需要隐藏Nginx的版本号,以避免泄漏Nginx的版本,使攻击者不能针对特定版本进行攻击,下面就来介绍一下Nginx安全防护的方法,感兴趣的可以了解一下... 目录核心安全配置1.编译安装 Nginx2.隐藏版本号3.限制危险请求方法4.请求限制(CC攻击防御)

python生成随机唯一id的几种实现方法

《python生成随机唯一id的几种实现方法》在Python中生成随机唯一ID有多种方法,根据不同的需求场景可以选择最适合的方案,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习... 目录方法 1:使用 UUID 模块(推荐)方法 2:使用 Secrets 模块(安全敏感场景)方法

MyBatis-Plus通用中等、大量数据分批查询和处理方法

《MyBatis-Plus通用中等、大量数据分批查询和处理方法》文章介绍MyBatis-Plus分页查询处理,通过函数式接口与Lambda表达式实现通用逻辑,方法抽象但功能强大,建议扩展分批处理及流式... 目录函数式接口获取分页数据接口数据处理接口通用逻辑工具类使用方法简单查询自定义查询方法总结函数式接口

MySQL深分页进行性能优化的常见方法

《MySQL深分页进行性能优化的常见方法》在Web应用中,分页查询是数据库操作中的常见需求,然而,在面对大型数据集时,深分页(deeppagination)却成为了性能优化的一个挑战,在本文中,我们将... 目录引言:深分页,真的只是“翻页慢”那么简单吗?一、背景介绍二、深分页的性能问题三、业务场景分析四、

JAVA中安装多个JDK的方法

《JAVA中安装多个JDK的方法》文章介绍了在Windows系统上安装多个JDK版本的方法,包括下载、安装路径修改、环境变量配置(JAVA_HOME和Path),并说明如何通过调整JAVA_HOME在... 首先去oracle官网下载好两个版本不同的jdk(需要登录Oracle账号,没有可以免费注册)下载完

IntelliJ IDEA2025创建SpringBoot项目的实现步骤

《IntelliJIDEA2025创建SpringBoot项目的实现步骤》本文主要介绍了IntelliJIDEA2025创建SpringBoot项目的实现步骤,文中通过示例代码介绍的非常详细,对大家... 目录一、创建 Spring Boot 项目1. 新建项目2. 基础配置3. 选择依赖4. 生成项目5.

Linux线程之线程的创建、属性、回收、退出、取消方式

《Linux线程之线程的创建、属性、回收、退出、取消方式》文章总结了线程管理核心知识:线程号唯一、创建方式、属性设置(如分离状态与栈大小)、回收机制(join/detach)、退出方法(返回/pthr... 目录1. 线程号2. 线程的创建3. 线程属性4. 线程的回收5. 线程的退出6. 线程的取消7.