创建RDD的常用方式【并行化集合创建、从外部存储系统创建、RDD衍生、小文件读取】,创建SparkContext对象的方式

本文主要是介绍创建RDD的常用方式【并行化集合创建、从外部存储系统创建、RDD衍生、小文件读取】,创建SparkContext对象的方式,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 创建 SparkContext 对象
  • RDD 创建方式
    • 1. 创建并行化集合RDD(parallelize、makeRDD)
      • parallelize
      • makeRDD
    • 2. 从外部存储系统创建RDD(textfile)
    • 3. 从RDD衍生新的RDD(原地计算)
    • 4. 小文件读取(wholeTextFiles)
  • RDD 分区数目(总结)
    • 获取 RDD 分区数目的俩种方式
    • RDD 分区的数据取决于那些因素?


创建 SparkContext 对象

SparkApplication应用,创建SparkContext对象,及应用结束关闭资源,代码如下:

import org.apache.spark.{SparkConf, SparkContext}object MydemoTest {def main(args: Array[String]): Unit = {// 构建SparkContext对象val sc: SparkContext = {// a. 创建SparkConf对象val sparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[2]")// b. 传递SparkConf对象,创建实例val context = SparkContext.getOrCreate(sparkConf) //有就获取,没有创建// c. 返回实例对象context}// 应用结束,关闭资源sc.stop()}}

或者写的简单点

val conf = new SparkConf().setAppName("word_count").setMaster("local[6]")
val sc = new SparkContext(conf)

RDD 创建方式

1. 创建并行化集合RDD(parallelize、makeRDD)

parallelize

源码:

def parallelize[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism
): RDD[T]

需要传入的是一个Seq集合,和一个分区数Int


makeRDD

源码结构和parallelize一样,还是调用的parallelize。

两者区别:(源码的区别)

  • parallelize可以不指定分区数,使用比较多。
  • makeRDD不指定就会调用成别的同名方法,可以看下源码来验证,一般不用。

上述俩种方式的写法如下:

//从本地集合创建
@Test
def rddCreationLocal() = {val seq = Seq("Hello1", "hello2", "Hello3")val rdd1: RDD[String] = sc.parallelize(seq, 2)  //指定分区数目val rdd2: RDD[String] = sc.makeRDD(seq, 2) //底层依旧调用的parallelize
}

2. 从外部存储系统创建RDD(textfile)

由外部存储系统的数据集创建,包括本地的文件系统,还有所有 Hadoop支持的数据集,比如HDFS、Cassandra、HBase 等。实际使用最多的方法:textFile,读取HDFS或LocalFS上文本文件,指定文件路径和RDD分区数目。

源码:

def textFile(path : scala.Predef.String, minPartitions : scala.Int = { /* compiled code */ }
) : org.apache.spark.rdd.RDD[scala.Predef.String] = { /* compiled code */ }

注意:

  • 传入的文件路径,可以是 hdfs://… , 也可以是 file://… 。这种方式分为在集群中执行和在本地执行,在集群中是hdfs://,本地则是file:// 。
  • 支持分区,如果传入的path是 hdfs://… ,分区则由hdfs文件的block决定。默认情况下,RDD分区数目等于HDFS上Block块数目。其中文件路径:最好是全路径,可以指定文件名称,可以指定文件目录,可以使用通配符指定。
  • 支持外部数据源,比如阿里源等等。

3. 从RDD衍生新的RDD(原地计算)

通过在RDD上进行算子操作,会生成新的RDD,那么新的RDD是原来的RDD吗?提一个概念,原地计算?类比一下字符串操作 str.substr 返回的是新的 str。那么这个叫做非原地计算,那么原来的字符串变了吗?当然没变?同理RDD不可变!!!

在这里插入图片描述


4. 小文件读取(wholeTextFiles)

在实际项目中,有时往往处理的数据文件属于小文件(每个文件数据数据量很小,比如KB,几十MB等),文件数量又很大,如果一个个文件读取为RDD的一个个分区,计算数据时很耗时性能低下,使用SparkContext中提供:wholeTextFiles类,专门读取小文件数据。

def wholeTextFiles(path: String,	//文件存储目录minPartitions: Int = defaultMinPartitions	//RDD分区数目
): RDD[(String, String)]

范例演示:读取100个小文件数据,每个文件大小小于1MB,设置RDD分区数目为2。

实际项目中,可以先使用wholeTextFiles方法读取数据,设置适当RDD分区,再将数据保存到文件系统,以便后续应用读取处理,大大提升性能。


RDD 分区数目(总结)

在讲解 RDD 属性时,多次提到了分区(partition)的概念。分区是一个偏物理层的概念,也是 RDD 并行计算的核心。数据在 RDD 内部被切分为多个子集合,每个子集合可以被认为是一个分区,运算逻辑最小会被应用在每一个分区上,每个分区是由一个单独的任务(task)来运行的,所以分区数越多,整个应用的并行度也会越高。

获取 RDD 分区数目的俩种方式

  • rdd.getNumPartitions
  • rdd.partitions.length

RDD 分区的数据取决于那些因素?

  • 第一点:RDD分区的原则是使得分区的个数尽量等于集群中的CPU核心(core)数目,这样可以充分利用CPU的计算资源;
  • 第二点:在实际中为了更加充分的压榨CPU的计算资源,会把并行度设置为cpu核数的2~3倍;
  • 第三点:RDD分区数和启动时指定的核数、调用方法时指定的分区数、如文件本身分区数有关系,具体如下说明:
    • 1.启动的时候指定的CPU核数确定了一个参数值:
      • spark.default.parallelism = 指定的CPU核数(集群模式最小为2)
    • 2.对于Scala集合调用parallellize(集合,分区数)方法:
      • 如果没有指定分区数目,就使用 spark.default.parallelism
      • 如果指定了就使用指定的分区数(不要指定大于spark.default.parallelism)
    • 3.对于textFile(文件,分区数)
      • defaultMinPartitons
        • 如果没有指定分区数目 sc.defaultMinPartitions=min(defaultParallelism, 2)
        • 如果指定了就使用指定的分区数 sc.defaultMinPartitions=指定的分区数rdd的分区数
      • rdd的分区数
        • 对于本地文件:
          • rdd的分区数 = max(本地file的分片数, sc.defaultMinPartitions)
        • 对于HDFS文件:
          • rdd的分区数 = max(hdfs文件的block数目,sc.defaultMinPartitions)
        • 所以如果分配的核数为多个,且从文件中读取数据创建RDD,即使hdfs文件只有1个切片,最后的Spark的RDD的partition数也有可能是2

这篇关于创建RDD的常用方式【并行化集合创建、从外部存储系统创建、RDD衍生、小文件读取】,创建SparkContext对象的方式的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/488191

相关文章

HarmonyOS学习(七)——UI(五)常用布局总结

自适应布局 1.1、线性布局(LinearLayout) 通过线性容器Row和Column实现线性布局。Column容器内的子组件按照垂直方向排列,Row组件中的子组件按照水平方向排列。 属性说明space通过space参数设置主轴上子组件的间距,达到各子组件在排列上的等间距效果alignItems设置子组件在交叉轴上的对齐方式,且在各类尺寸屏幕上表现一致,其中交叉轴为垂直时,取值为Vert

JS常用组件收集

收集了一些平时遇到的前端比较优秀的组件,方便以后开发的时候查找!!! 函数工具: Lodash 页面固定: stickUp、jQuery.Pin 轮播: unslider、swiper 开关: switch 复选框: icheck 气泡: grumble 隐藏元素: Headroom

【C++】_list常用方法解析及模拟实现

相信自己的力量,只要对自己始终保持信心,尽自己最大努力去完成任何事,就算事情最终结果是失败了,努力了也不留遗憾。💓💓💓 目录   ✨说在前面 🍋知识点一:什么是list? •🌰1.list的定义 •🌰2.list的基本特性 •🌰3.常用接口介绍 🍋知识点二:list常用接口 •🌰1.默认成员函数 🔥构造函数(⭐) 🔥析构函数 •🌰2.list对象

常用的jdk下载地址

jdk下载地址 安装方式可以看之前的博客: mac安装jdk oracle 版本:https://www.oracle.com/java/technologies/downloads/ Eclipse Temurin版本:https://adoptium.net/zh-CN/temurin/releases/ 阿里版本: github:https://github.com/

【Python编程】Linux创建虚拟环境并配置与notebook相连接

1.创建 使用 venv 创建虚拟环境。例如,在当前目录下创建一个名为 myenv 的虚拟环境: python3 -m venv myenv 2.激活 激活虚拟环境使其成为当前终端会话的活动环境。运行: source myenv/bin/activate 3.与notebook连接 在虚拟环境中,使用 pip 安装 Jupyter 和 ipykernel: pip instal

内核启动时减少log的方式

内核引导选项 内核引导选项大体上可以分为两类:一类与设备无关、另一类与设备有关。与设备有关的引导选项多如牛毛,需要你自己阅读内核中的相应驱动程序源码以获取其能够接受的引导选项。比如,如果你想知道可以向 AHA1542 SCSI 驱动程序传递哪些引导选项,那么就查看 drivers/scsi/aha1542.c 文件,一般在前面 100 行注释里就可以找到所接受的引导选项说明。大多数选项是通过"_

30常用 Maven 命令

Maven 是一个强大的项目管理和构建工具,它广泛用于 Java 项目的依赖管理、构建流程和插件集成。Maven 的命令行工具提供了大量的命令来帮助开发人员管理项目的生命周期、依赖和插件。以下是 常用 Maven 命令的使用场景及其详细解释。 1. mvn clean 使用场景:清理项目的生成目录,通常用于删除项目中自动生成的文件(如 target/ 目录)。共性规律:清理操作

在cscode中通过maven创建java项目

在cscode中创建java项目 可以通过博客完成maven的导入 建立maven项目 使用快捷键 Ctrl + Shift + P 建立一个 Maven 项目 1 Ctrl + Shift + P 打开输入框2 输入 "> java create"3 选择 maven4 选择 No Archetype5 输入 域名6 输入项目名称7 建立一个文件目录存放项目,文件名一般为项目名8 确定

uva 11178 计算集合模板题

题意: 求三角形行三个角三等分点射线交出的内三角形坐标。 代码: #include <iostream>#include <cstdio>#include <cstdlib>#include <algorithm>#include <cstring>#include <cmath>#include <stack>#include <vector>#include <

Java 创建图形用户界面(GUI)入门指南(Swing库 JFrame 类)概述

概述 基本概念 Java Swing 的架构 Java Swing 是一个为 Java 设计的 GUI 工具包,是 JAVA 基础类的一部分,基于 Java AWT 构建,提供了一系列轻量级、可定制的图形用户界面(GUI)组件。 与 AWT 相比,Swing 提供了许多比 AWT 更好的屏幕显示元素,更加灵活和可定制,具有更好的跨平台性能。 组件和容器 Java Swing 提供了许多