当前位置: 首页 > news >正文

Spark---RDD算子(单值类型Value)

文章目录

  • 1.RDD算子介绍
  • 2.转换算子
      • 2.1 Value类型
          • 2.1.1 map
          • 2.1.2 mapPartitions
          • 2.1.3 mapPartitionsWithIndex
          • 2.1.4 flatMap
          • 2.1.5 glom
          • 2.1.6 groupBy
          • 2.1.7 filter
          • 2.1.8 sample
          • 2.1.9 distinct
          • 2.1.10 coalesce
          • 2.1.11 repartition
          • 2.1.12 sortBy

1.RDD算子介绍

RDD算子是用于对RDD进行转换(Transformation)或行动(Action)操作的方法或函数。通俗来讲,RDD算子就是RDD中的函数或者方法,根据其功能,RDD算子可以分为两大类:
转换算子(Transformation): 转换算子用于从一个RDD生成一个新的RDD,但是原始RDD保持不变。常见的转换算子包括map、filter、flatMap等,它们通过对RDD的每个元素执行相应的操作来生成新的RDD。
行动算子(Action): 行动算子触发对RDD的实际计算,并返回计算结果或将结果写入外部存储系统。与转换算子不同,行动算子会导致Spark作业的执行。如collect方法。

2.转换算子

RDD 根据数据处理方式的不同将算子整体上分为:
Value 类型:对一个RDD进行操作或行动,生成一个新的RDD。
双 Value 类型:对两个RDD进行操作或行动,生成一个新的RDD。
Key-Value类型:对键值对进行操作,如reduceByKey((x, y),按照key对value进行合并。

2.1 Value类型

2.1.1 map

将处理的数据逐条进行映射转换,这里的转换可以是类型的转换,也可以是值的转换。

函数定义
def map[U: ClassTag](f: T => U): RDD[U]

代码实现:

    //建立与Spark框架的连接val rdd = new SparkConf().setMaster("local[*]").setAppName("RDD") //配置文件val sparkRdd = new SparkContext(rdd) //读取配置文件val mapRdd: RDD[Int] = sparkRdd.makeRDD(List(1, 2, 3, 4))//对mapRdd进行转换val mapRdd1 = mapRdd.map(num => num * 2)//对mapRdd1进行转换val mapRdd2 = mapRdd1.map(num => num + "->")mapRdd2.collect().foreach(print)sparkRdd.stop();//关闭连接

在这里插入图片描述

2.1.2 mapPartitions

将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据。

函数定义
def mapPartitions[U: ClassTag](
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

Map 算子是分区内一个数据一个数据的执行,类似于串行操作。而 mapPartitions 算子是以分区为单位进行批处理操作。

mapPartitions在处理数据的时候因为是批处理,相对于map来说处理效率较高,但是如果数据量较大的情况下使用mapPartitions可能会造成内存溢出,因为mapPartitions会将分区内的数据全部加载到内存中。此时更推荐使用map。

2.1.3 mapPartitionsWithIndex

将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据,在处理时同时可以获取当前分区索引。

函数定义
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

实现只保留第二个分区的数据

    val mapRdd: RDD[Int] = sparkRdd.makeRDD(List(1, 2, 3, 4),2)val newRdd: RDD[Int] = mapRdd.mapPartitionsWithIndex((index, iterator) => {if (index == 1) iteratorelse Nil.iterator})newRdd.collect().foreach(println)
2.1.4 flatMap

将处理的数据进行扁平化后再进行映射处理,所以算子也称之为扁平映射

       //建立与Spark框架的连接val rdd = new SparkConf().setMaster("local[*]").setAppName("RDD") //配置文件val sparkRdd = new SparkContext(rdd) //读取配置文件val rdd1: RDD[List[Int]] = sparkRdd.makeRDD(List(List(1, 2), List(3, 4)))val rdd2: RDD[String] = sparkRdd.makeRDD(List("Hello Java", "Hello Scala"), 2)val frdd1: RDD[Int] =rdd1.flatMap(list=>{list})val frdd2: RDD[String] =rdd2.flatMap(str=>str.split(" "))frdd1.collect().foreach(println)frdd2.collect().foreach(println)sparkRdd.stop();//关闭连接

在这里插入图片描述

2.1.5 glom

将同一个分区的数据直接转换为相同类型的内存数组进行处理,分区不变,glom函数的作用就是将一组数据转换为数组。

函数定义
def glom(): RDD[Array[T]]

    /建立与Spark框架的连接val rdd = new SparkConf().setMaster("local[*]").setAppName("RDD") //配置文件val sparkRdd = new SparkContext(rdd) //读取配置文件val rdd1: RDD[Any] = sparkRdd.makeRDD(List(1,2,3,4),2)val value: RDD[Array[Any]] = rdd1.glom()value.collect().foreach(data=> println(data.mkString(",")))sparkRdd.stop();//关闭连接

在这里插入图片描述

2.1.6 groupBy

将数据根据指定的规则进行分组, 分区默认不变,但是数据会被打乱重新组合,我们将这样的操作称之为 shuffle。 极限情况下,数据可能被分在同一个分区中

函数定义
def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

	    //按照奇偶分组val rdd1: RDD[Int] = sparkRdd.makeRDD(List(1,2,3,4),2)val value = rdd1.groupBy(num => num % 2)value.collect().foreach(println)//将 List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。val rdd2: RDD[String] = sparkRdd.makeRDD(List("Hello", "hive", "hbase", "Hadoop"))val value1: RDD[(Char, Iterable[String])] = rdd2.groupBy(str => {str.charAt(0)})value1.collect().foreach(println)

在这里插入图片描述

2.1.7 filter

将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃。当数据进行筛选过滤后,分区不变,但是分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜。

函数定义
def filter(f: T => Boolean): RDD[T]

	//获取偶数val dataRDD = sparkRdd.makeRDD(List(1, 2, 3, 4), 1)val value1 = dataRDD.filter(_ % 2 == 0)
2.1.8 sample

函数定义
def sample(
withReplacement: Boolean,
fraction: Double,
seed: Long = Utils.random.nextLong): RDD[T]

根据指定的规则从数据集中抽取数据

参数具体意义:
1.抽取数据不放回withReplacement: Boolean, 该参数表示抽取不放回,此时采用伯努利算法(false)fraction: Double,该参数表示抽取的几率,范围在[0,1]之间,0:全不取;1:全取;seed: Long = Utils.random.nextLong): RDD[T] 该参数表示随机数种子2.抽取数据放回withReplacement: Boolean, 该参数表示抽取放回,此时采用泊松算法(true)fraction: Double,该参数表示重复数据的几率,范围大于等于 0.表示每一个元素被期望抽取到的次数seed: Long = Utils.random.nextLong): RDD[T] 该参数表示随机数种子
2.1.9 distinct

将数据集中重复的数据去重

def distinct()(implicit ord: Ordering[T] = null): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

    val dataRDD = sparkRdd.makeRDD(List(1, 2, 3, 4, 1, 2), 6)val value = dataRDD.distinct()

在这里插入图片描述

2.1.10 coalesce

根据数据量缩减分区,用于大数据集过滤后,提高小数据集的执行效率当 spark 程序中,存在过多的小任务的时候,可以通过 coalesce 方法,收缩合并分区,减少分区的个数,减小任务调度成本

def coalesce(numPartitions: Int, shuffle: Boolean = false,
partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
(implicit ord: Ordering[T] = null)
: RDD[T]

    //初始Rdd采用6个分区val dataRDD = sparkRdd.makeRDD(List(1, 2, 3, 4, 1, 2), 6)//将分区数量缩减至2个val value = dataRDD.coalesce(2)

在coalesce中默认不开启shuffle,在进行分区缩减的时候,数据不会被打散。
在这里插入图片描述

2.1.11 repartition

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

repartition内部其实执行的是 coalesce 操作,参数 shuffle 的默认值为 true。无论是将分区数多的RDD 转换为分区数少的 RDD,还是将分区数少的 RDD 转换为分区数多的 RDD,repartition操作都可以完成,因为无论如何都会经 shuffle 过程。
在这里插入图片描述

	//将分区数量从2个提升至4个val dataRDD = sparkRdd.makeRDD(List(1, 2, 3, 4, 1, 2), 2)val dataRDD1 = dataRDD.repartition(4)
2.1.12 sortBy

该操作用于排序数据。在排序之前,可以将数据通过 f 函数进行处理,之后按照 f 函数处理的结果进行排序,默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程

def sortBy[K](
f: (T) => K, 该参数表述用于处理的函数
ascending: Boolean = true, 该参数表示是否升序排序
numPartitions: Int = this.partitions.length) 该参数表示设置分区数量
(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

    val dataRDD = sparkRdd.makeRDD(List(1, 2, 3, 4, 1, 2), 2)//按照初始数据降序排列val dataRDD1 = dataRDD.sortBy(num => num, false, 4)

相关文章:

Spark---RDD算子(单值类型Value)

文章目录 1.RDD算子介绍2.转换算子2.1 Value类型2.1.1 map2.1.2 mapPartitions2.1.3 mapPartitionsWithIndex2.1.4 flatMap2.1.5 glom2.1.6 groupBy2.1.7 filter2.1.8 sample2.1.9 distinct2.1.10 coalesce2.1.11 repartition2.1.12 sortBy 1.RDD算子介绍 RDD算子是用于对RDD进…...

数据库中的MVCC--多版本并发控制

一、前言 1、定义:MVCC(Multi-Version Concurrency Control),多版本并发控制,主要为了提高数据库 的并发性能。是MySQL的InnoDB存储引擎实现隔离级别的一种具体方式。用于实现提交读和可重 复读这两种隔离级别。 2…...

wps将姓名处理格式为:姓**

1.打开wps,在要处理数据右侧一个单元格 输入公式:LEFT(A1,1)&"**",然后回车 2.按住ctrl和处理好的数据的右下角小方框,往下拖动即可生成格式为:姓** 格式的数据 3.复制生成的数据,右键选择 “…...

2023年我的编程之旅:技术演进与自我成长的纪录

2023年我的编程之旅:技术演进与自我成长的纪录 转眼间,2023年已经悄然走到了尾声。这一年,对我来说既是挑战也是机遇的一年。我的编程之旅如同坐上了一辆高速前进的列车,从新技术的学习探索到项目实战的沉浸经历,再到…...

好用免费的WAF---如何安装雷池社区版

什么是雷池​ 雷池(SafeLine)是长亭科技耗时近 10 年倾情打造的 WAF,核心检测能力由智能语义分析算法驱动。 Slogan: 不让黑客越雷池半步。 什么是 WAF​ WAF 是 Web Application Firewall 的缩写,也被称为 Web 应用防火墙。 …...

看似 bug 又非 bug 的一个 bug

最近的一个项目中&#xff0c;对于 CSS 的一些属性一些选择符可以大胆使用&#xff0c;然后很意外得撞上一个 iOS 中 Safari 的一个解析问题。 <Component style{{height: "calc(100vh - 46px)"}}>一个组件</Component> 这样的一段代码很简单&#xff…...

mysql常见问题

批量导入SQL 数据库结构 数据时&#xff0c;如果数据是批量插入的话会报错&#xff1a;2006 - MySQL server has gone away。 解决办法&#xff1a;找到你的 mysql 目录下的 my.ini 配置文件&#xff0c;加入以下代码 max_allowed_packet500M wait_timeout288000 interactiv…...

QT上位机开发(串口界面设计)

【 声明&#xff1a;版权所有&#xff0c;欢迎转载&#xff0c;请勿用于商业用途。 联系信箱&#xff1a;feixiaoxing 163.com】 如果上位机要和嵌入式设备进行打交道的话&#xff0c;那么串口可能就是我们遇到的第一个硬件设备。串口的物理接线很简单&#xff0c;基本上就是收…...

k8s之pod

1、pod&#xff1a;k8s中最小的资源管理组件&#xff0c;最小化运行容器化应用的资源管理对象 &#xff08;1&#xff09;pod是一个抽象的概念&#xff0c;可以理解为一个或者多个容器化应用的集合 &#xff08;2&#xff09;一个pod中运行一个容器是最常用的方式 &#xff…...

第二百四十三回 再分享一个Json工具

文章目录 1. 概念介绍2. 分析与比较2.1 分析问题2.2 比较差异 3. 使用方法4. 内容总结 我们在上一章回中介绍了"分享三个使用TextField的细节"相关的内容&#xff0c;本章回中将再 分享一个Json插件.闲话休提&#xff0c;让我们一起Talk Flutter吧。 1. 概念介绍 我…...

electron自定义菜单

创建menu.js const { app, Menu } require("electron"); const createMenu () > {const menu [{label: "菜单",submenu: [{label: "新增",click: () > {},}, ],},{label: "关于",submenu: [{label: "新增",click:…...

变量和函数提升(js的问题)

• js解释执行 • 变量和函数提升 变量声明提前&#xff0c;函数声明提前 • 变量声明提前&#xff1a;值停留在本地 • 函数声明提前&#xff1a;整个函数体提前 如果是var赋值声明的函数&#xff0c;变量提前&#xff0c;函数体停留在本地 1、变量提…...

Excel 插件:ASAP Utilities Crack

ASAP Utilities是一款功能强大的 Excel 插件&#xff0c;填补了 Excel 的空白。在过去的 20 年里&#xff0c;我们的加载项已经发展成为世界上最受欢迎的 Microsoft Excel 加载项之一。 ASAP Utilities 中的功能数量&#xff08;300 多个&#xff09;可能看起来有点令人眼花缭乱…...

hyperf 十九 数据库 二 模型

教程&#xff1a;Hyperf 一、命令行 symfony/console-CSDN博客 hypery 十一、命令行-CSDN博客 hyperf console 执行-CSDN博客 根据之前应该能了解到命令行的基本实现&#xff0c;和hyperf中命令行的定义。 1.1 命令初始化 hyperf.php中系统初始化中通过ApplicationFacto…...

使用python快速开发与PDF文档对话的Gemini聊天机器人

检索增强生成(Retrieval-augmented generation&#xff0c;RAG)使得我们可以让大型语言模型(LLMs)访问外部知识库数据(如pdf,word、text等)&#xff0c;从而让人们可以更加方便的通过LLM来学习外部数据的知识。今天我们将利用之前学习到的RAG方法&#xff0c;谷歌Gemini模型和l…...

Spring Cloud Gateway集成Knife4j

1、前提 网关路由能够正常工作。 案例 基于 Spring Cloud Gateway Nacos 实现动态路由拓展的参考地址&#xff1a;Spring Cloud Gateway Nacos 实现动态路由 详细官网案例&#xff1a;https://doc.xiaominfo.com/docs/middleware-sources/spring-cloud-gateway/spring-gatewa…...

Hive10_窗口函数

窗口函数&#xff08;开窗函数&#xff09; 1 相关函数说明 普通的聚合函数聚合的行集是组,开窗函数聚合的行集是窗口。因此,普通的聚合函数每组(Group by)只返回一个值&#xff0c;而开窗函数则可为窗口中的每行都返回一个值。简单理解&#xff0c;就是对查询的结果多出一列…...

ipvsadm命令详解

ipvsadm命令详解 大家好&#xff0c;我是免费搭建查券返利机器人赚佣金就用微赚淘客系统3.0的小编&#xff0c;也是冬天不穿秋裤&#xff0c;天冷也要风度的程序猿&#xff01;今天我们将深入探讨一个在Linux系统网络管理中极具威力的命令——ipvsadm&#xff0c;通过详细解析…...

zabbix通过自动发现-配置监控项、触发器(小白教程)

自动发现配置参考链接&#xff08;不小白&#xff0c;不友好&#xff09; zabbix-get介绍 1配置 zabbix server&#xff1a;版本7&#xff08;不影响&#xff09;,IP地址&#xff1a;192.168.0.60zabbix agent&#xff1a;版本agent1&#xff08;不影响&#xff09;&#xff…...

Dockerfile文件介绍

0 Preface/Foreword 0.1 Docker docker用来自制镜像。 1 Introduction 1.1 Dockerfile Dockerfile是用于定义Docker镜像的构建过程&#xff0c;它包含一系列的指令用于安装 软件包、配置环境等操作。 Dockerfile文件的格式如下&#xff1a; FROM base_image RUN apt-get up…...

深入剖析AI大模型:大模型时代的 Prompt 工程全解析

今天聊的内容&#xff0c;我认为是AI开发里面非常重要的内容。它在AI开发里无处不在&#xff0c;当你对 AI 助手说 "用李白的风格写一首关于人工智能的诗"&#xff0c;或者让翻译模型 "将这段合同翻译成商务日语" 时&#xff0c;输入的这句话就是 Prompt。…...

docker详细操作--未完待续

docker介绍 docker官网: Docker&#xff1a;加速容器应用程序开发 harbor官网&#xff1a;Harbor - Harbor 中文 使用docker加速器: Docker镜像极速下载服务 - 毫秒镜像 是什么 Docker 是一种开源的容器化平台&#xff0c;用于将应用程序及其依赖项&#xff08;如库、运行时环…...

椭圆曲线密码学(ECC)

一、ECC算法概述 椭圆曲线密码学&#xff08;Elliptic Curve Cryptography&#xff09;是基于椭圆曲线数学理论的公钥密码系统&#xff0c;由Neal Koblitz和Victor Miller在1985年独立提出。相比RSA&#xff0c;ECC在相同安全强度下密钥更短&#xff08;256位ECC ≈ 3072位RSA…...

React Native 开发环境搭建(全平台详解)

React Native 开发环境搭建&#xff08;全平台详解&#xff09; 在开始使用 React Native 开发移动应用之前&#xff0c;正确设置开发环境是至关重要的一步。本文将为你提供一份全面的指南&#xff0c;涵盖 macOS 和 Windows 平台的配置步骤&#xff0c;如何在 Android 和 iOS…...

抖音增长新引擎:品融电商,一站式全案代运营领跑者

抖音增长新引擎&#xff1a;品融电商&#xff0c;一站式全案代运营领跑者 在抖音这个日活超7亿的流量汪洋中&#xff0c;品牌如何破浪前行&#xff1f;自建团队成本高、效果难控&#xff1b;碎片化运营又难成合力——这正是许多企业面临的增长困局。品融电商以「抖音全案代运营…...

Linux云原生安全:零信任架构与机密计算

Linux云原生安全&#xff1a;零信任架构与机密计算 构建坚不可摧的云原生防御体系 引言&#xff1a;云原生安全的范式革命 随着云原生技术的普及&#xff0c;安全边界正在从传统的网络边界向工作负载内部转移。Gartner预测&#xff0c;到2025年&#xff0c;零信任架构将成为超…...

多模态大语言模型arxiv论文略读(108)

CROME: Cross-Modal Adapters for Efficient Multimodal LLM ➡️ 论文标题&#xff1a;CROME: Cross-Modal Adapters for Efficient Multimodal LLM ➡️ 论文作者&#xff1a;Sayna Ebrahimi, Sercan O. Arik, Tejas Nama, Tomas Pfister ➡️ 研究机构: Google Cloud AI Re…...

以光量子为例,详解量子获取方式

光量子技术获取量子比特可在室温下进行。该方式有望通过与名为硅光子学&#xff08;silicon photonics&#xff09;的光波导&#xff08;optical waveguide&#xff09;芯片制造技术和光纤等光通信技术相结合来实现量子计算机。量子力学中&#xff0c;光既是波又是粒子。光子本…...

Linux C语言网络编程详细入门教程:如何一步步实现TCP服务端与客户端通信

文章目录 Linux C语言网络编程详细入门教程&#xff1a;如何一步步实现TCP服务端与客户端通信前言一、网络通信基础概念二、服务端与客户端的完整流程图解三、每一步的详细讲解和代码示例1. 创建Socket&#xff08;服务端和客户端都要&#xff09;2. 绑定本地地址和端口&#x…...

安宝特案例丨Vuzix AR智能眼镜集成专业软件,助力卢森堡医院药房转型,赢得辉瑞创新奖

在Vuzix M400 AR智能眼镜的助力下&#xff0c;卢森堡罗伯特舒曼医院&#xff08;the Robert Schuman Hospitals, HRS&#xff09;凭借在无菌制剂生产流程中引入增强现实技术&#xff08;AR&#xff09;创新项目&#xff0c;荣获了2024年6月7日由卢森堡医院药剂师协会&#xff0…...