【Python】PySpark 数据计算 ① ( RDD#map 方法 | RDD#map 语法 | 传入普通函数 | 传入 lambda 匿名函数 | 链式调用 )

2023-07-30 21:53:18

文章目录





一、RDD#map 方法



1、RDD#map 方法引入


在 PySpark 中 RDD 对象 提供了一种 数据计算方法 RDD#map 方法 ;

该 RDD#map 函数 可以对 RDD 数据中的每个元素应用一个函数 , 该 被应用的函数 ,

  • 可以将每个元素转换为另一种类型 ,
  • 也可以针对 RDD 数据的 原始元素进行 指定操作 ;

计算完毕后 , 会返回一个新的 RDD 对象 ;


2、RDD#map 语法


map 方法 , 又称为 map 算子 , 可以将 RDD 中的数据元素 逐个进行处理 , 处理的逻辑 需要用外部 通过 参数传入 map 函数 ;


RDD#map 语法 :

rdd.map(fun)

传入的 fun 是一个函数 , 其函数类型为 :

(T) -> U

上述 函数 类型 前面的 小括号 及其中的内容 , 表示 函数 的参数类型 ,

  • () 表示不传入参数 ;
  • (T) 表示传入 1 个参数 ;

同时 T 类型是 泛型 , 表示任意类型 , 也就是说 该函数的 参数 可以是任意类型的 ;


上述 函数 类型 右箭头 后面的 U , -> U 表示的是 函数 返回值类型 ,

  • (T) -> U 表示 参数 类型为 T , 返回值类型为 U , T 和 U 类型都是任意类型 , 可以是一个类型 , 也可以是不同的类型 ;
  • (T) -> T 函数类型中 , T 可以是任意类型 , 但是如果确定了参数 , 那么返回值必须也是相同的类型 ;

U 类型也是 泛型 , 表示任意类型 , 也就是说 该函数的 参数 可以是任意类型的 ;


3、RDD#map 用法


RDD#map 方法 , 接收一个 函数 作为参数 , 计算时 , 该 函数参数 会被应用于 RDD 数据中的每个元素 ;

下面的 代码 , 传入一个 lambda 匿名函数 , 将 RDD 对象中的元素都乘以 10 ;

# 将 RDD 对象中的元素都乘以 10
rdd.map(lambda x: x * 10)

4、代码示例 - RDD#map 数值计算 ( 传入普通函数 )


在下面的代码中 ,

首先 , 创建了一个包含整数的 RDD ,

# 创建一个包含整数的 RDD
rdd = sparkContext.parallelize([1, 2, 3, 4, 5])

然后 , 使用 map() 方法将每个元素乘以 10 ;

# 为每个元素执行的函数
def func(element):
    return element * 10


# 应用 map 操作，将每个元素乘以 10
rdd2 = rdd.map(func)

最后 , 打印新的 RDD 中的内容 ;

# 打印新的 RDD 中的内容
print(rdd2.collect())

代码示例 :

"""
PySpark 数据处理
"""

# 导入 PySpark 相关包
from pyspark import SparkConf, SparkContext
# 为 PySpark 配置 Python 解释器
import os
os.environ['PYSPARK_PYTHON'] = "Y:/002_WorkSpace/PycharmProjects/pythonProject/venv/Scripts/python.exe"

# 创建 SparkConf 实例对象 , 该对象用于配置 Spark 任务
# setMaster("local[*]") 表示在单机模式下 本机运行
# setAppName("hello_spark") 是给 Spark 程序起一个名字
sparkConf = SparkConf() \
    .setMaster("local[*]") \
    .setAppName("hello_spark")

# 创建 PySpark 执行环境 入口对象
sparkContext = SparkContext(conf=sparkConf)

# 打印 PySpark 版本号
print("PySpark 版本号 : ", sparkContext.version)

# 创建一个包含整数的 RDD
rdd = sparkContext.parallelize([1, 2, 3, 4, 5])


# 为每个元素执行的函数
def func(element):
    return element * 10


# 应用 map 操作，将每个元素乘以 10
rdd2 = rdd.map(func)

# 打印新的 RDD 中的内容
print(rdd2.collect())

# 停止 PySpark 程序
sparkContext.stop()

执行结果 :

Y:\002_WorkSpace\PycharmProjects\pythonProject\venv\Scripts\python.exe Y:/002_WorkSpace/PycharmProjects/HelloPython/hello.py
23/07/30 21:39:59 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/07/30 21:39:59 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PySpark 版本号 :  3.4.1
[10, 20, 30, 40, 50]

Process finished with exit code 0

在这里插入图片描述


5、代码示例 - RDD#map 数值计算 ( 传入 lambda 匿名函数 )


在下面的代码中 ,

首先 , 创建了一个包含整数的 RDD ,

# 创建一个包含整数的 RDD
rdd = sparkContext.parallelize([1, 2, 3, 4, 5])

然后 , 使用 map() 方法将每个元素乘以 10 , 这里传入了 lambda 函数作为参数 , 该函数接受一个整数参数 element , 并返回 element * 10 ;

# 应用 map 操作，将每个元素乘以 10
rdd2 = rdd.map(lambda element: element * 10)

最后 , 打印新的 RDD 中的内容 ;

# 打印新的 RDD 中的内容
print(rdd2.collect())

代码示例 :

"""
PySpark 数据处理
"""

# 导入 PySpark 相关包
from pyspark import SparkConf, SparkContext
# 为 PySpark 配置 Python 解释器
import os
os.environ['PYSPARK_PYTHON'] = "Y:/002_WorkSpace/PycharmProjects/pythonProject/venv/Scripts/python.exe"

# 创建 SparkConf 实例对象 , 该对象用于配置 Spark 任务
# setMaster("local[*]") 表示在单机模式下 本机运行
# setAppName("hello_spark") 是给 Spark 程序起一个名字
sparkConf = SparkConf() \
    .setMaster("local[*]") \
    .setAppName("hello_spark")

# 创建 PySpark 执行环境 入口对象
sparkContext = SparkContext(conf=sparkConf)

# 打印 PySpark 版本号
print("PySpark 版本号 : ", sparkContext.version)

# 创建一个包含整数的 RDD
rdd = sparkContext.parallelize([1, 2, 3, 4, 5])

# 应用 map 操作，将每个元素乘以 10
rdd2 = rdd.map(lambda element: element * 10)

# 打印新的 RDD 中的内容
print(rdd2.collect())

# 停止 PySpark 程序
sparkContext.stop()

执行结果 :

Y:\002_WorkSpace\PycharmProjects\pythonProject\venv\Scripts\python.exe Y:/002_WorkSpace/PycharmProjects/HelloPython/hello.py
23/07/30 21:46:53 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/07/30 21:46:54 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PySpark 版本号 :  3.4.1
[10, 20, 30, 40, 50]

Process finished with exit code 0

在这里插入图片描述


6、代码示例 - RDD#map 数值计算 ( 链式调用 )


在下面的代码中 , 先对 RDD 对象中的每个元素数据都乘以 10 , 然后再对计算后的数据每个元素加上 5 , 最后对最新的计算数据每个元素除以 2 , 整个过程通过函数式编程 , 链式调用完成 ;

核心代码如下 :

# 创建一个包含整数的 RDD
rdd = sparkContext.parallelize([1, 2, 3, 4, 5])

# 应用 map 操作，将每个元素乘以 10
rdd2 = rdd.map(lambda element: element * 10)\
    .map(lambda element: element + 5)\
    .map(lambda element: element / 2)

# 打印新的 RDD 中的内容
print(rdd2.collect())

代码示例 :

"""
PySpark 数据处理
"""

# 导入 PySpark 相关包
from pyspark import SparkConf, SparkContext
# 为 PySpark 配置 Python 解释器
import os
os.environ['PYSPARK_PYTHON'] = "Y:/002_WorkSpace/PycharmProjects/pythonProject/venv/Scripts/python.exe"

# 创建 SparkConf 实例对象 , 该对象用于配置 Spark 任务
# setMaster("local[*]") 表示在单机模式下 本机运行
# setAppName("hello_spark") 是给 Spark 程序起一个名字
sparkConf = SparkConf() \
    .setMaster("local[*]") \
    .setAppName("hello_spark")

# 创建 PySpark 执行环境 入口对象
sparkContext = SparkContext(conf=sparkConf)

# 打印 PySpark 版本号
print("PySpark 版本号 : ", sparkContext.version)

# 创建一个包含整数的 RDD
rdd = sparkContext.parallelize([1, 2, 3, 4, 5])

# 应用 map 操作，将每个元素乘以 10
rdd2 = rdd.map(lambda element: element * 10)\
    .map(lambda element: element + 5)\
    .map(lambda element: element / 2)

# 打印新的 RDD 中的内容
print(rdd2.collect())

# 停止 PySpark 程序
sparkContext.stop()

执行结果 :

Y:\002_WorkSpace\PycharmProjects\pythonProject\venv\Scripts\python.exe Y:/002_WorkSpace/PycharmProjects/HelloPython/hello.py
23/07/30 21:50:29 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/07/30 21:50:29 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PySpark 版本号 :  3.4.1
[7.5, 12.5, 17.5, 22.5, 27.5]

Process finished with exit code 0

在这里插入图片描述

更多推荐

基于TensorFlow+CNN+协同过滤算法的智能电影推荐系统——深度学习算法应用(含微信小程序、ipynb工程源码)+MovieLens数据集（一）

目录前言总体设计系统整体结构图系统流程图运行环境Python环境TensorFlow环境方法一方法二后端服务器Django环境配置微信小程序环境相关其它博客工程源代码下载其它资料下载前言本项目专注于MovieLens数据集，并采用TensorFlow中的2D文本卷积网络模型。它结合了协同过滤算法来计算电影之间的余弦相似

人工智能安全-5-网络入侵检测

0提纲概述数据集数据预处理特征工程天池AI上的实践棘手问题1概述入侵检测是网络安全中的经典问题，入侵是指攻击者违反系统安全策略，试图破坏计算资源的完整性、机密性或可用性的任何行为。不管是哪种类型的入侵检测系统（IDS），其工作过程大体是相同的，可以分为三个主要的环节，即信息收集、分类检测和决策，其中，分类检测和决策环节

Git操作

前期准备：1、安装2、身份认证gitconfig--globaluser.name"your_username"gitconfig--globaluser.emailyour_email@domain.comgitconfig--list查看所有配置常见的Git命令gitstatus.命令用于显示工作目录和暂存区的状态

React useRequest解读

源码结构：可以看到虽然是一个hooks（具有一定功能且具备状态的单一函数）但是各种文件功能分得也是很细的，方便抽离和复用useRequest.ts抽离的原则还是单一功能原则可以看出真正的hooks实现是在Implement里对于类型type的引入我们需要显示声明这是importtypeuseRequestImpleme

华清 Qt day5 9月21

QT+=coreguisqlnetwork/*****************************************************************/#ifndefWIDGET_H#defineWIDGET_H#include<QWidget>#include<QWidget>#include

为何学linux及用处

目前企业使用的操作系统无非就是国产类的，windows和linux类。我们要提升自己的技能，需要学习这两款。我记得在大学时期，学习过windows以及linux，但当时觉得又不常用，就学的模棱两可。毕业之后，你会发现，其实这两种操作系统是很主流的。为什么学？下面就是一些工作中遇到的例子分享一下。我记得在企业中有次遇到数

【python第7课 实例，类】

文章目录一、实例1.1实例的变量1.2实例方法1.3构造方法1.4析构函数1.4预置实例属性：二，类1.1类变量1.2类方法1.3静态方法1.4类属性的增删改查一、实例1.1实例的变量使用示例classdog:def__init__(self,k,c,a):self.kinds=kself.color=cself.ag

【Hash表】两数之和-力扣 1 题

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。推荐:kuan的首页,持续学习,不断总结,共同进步,活到老学到老导航檀越剑指大厂系列:全面总结java核心技术点,如集合,jvm,并发编程redis,kaf

Docker笔记

安装卸载旧版本以及相关的依赖项sudoyumremovedocker\docker-client\docker-client-latest\docker-common\docker-latest\docker-latest-logrotate\docker-logrotate\docker-engine安装所需的软件包

leetcode分类刷题：二叉树（一、简单的层序遍历）

二叉树的深度优先遍历题目是让我有点晕，先把简单的层序遍历总结下吧：配合队列进行的层序遍历在逻辑思维上自然直观，不容易出错102.二叉树的层序遍历本题是二叉树的层序遍历模板：每次循环将一层节点出队，再将一层节点入队，也是所有可用层序遍历解二叉树题目的模板，只需要在模板里稍加改动即可解题fromtypingimportLi

Docker学习大纲

Docker是一个用于自动部署应用程序在轻量级容器中的平台。下面列出一些Docker的基础和必知概念。1.容器（Containers）容器是独立的应用程序运行环境。命令：dockerrunhello-world解析：该命令会从DockerHub下载一个叫做“hello-world”的镜像，并运行一个容器。2.镜像（Im

热文推荐
  1. API网关是如何提升API接口安全管控能力的

    API安全的重要性近几年，越来越多的企业开始数字化转型之路。数字化转型的核心是将企业的服务、资产和能力打包成服务（服务的形式通常为API，API又称接口，下文中提到的API和接口意思相同），从而让资源之间形成更强的连接和互动关系，释放原有资产的价值，提升企业的服务能力。企业数字化转型使得基于API的业务系统剧增，随之而

  2. RK3568笔记分享之“如何挂载SPI FRAM铁电存储芯片”——飞凌嵌入式

    对于做快速存储采集数据类产品的用户来说，在处理突发掉电情况时需要保存现有数据并避免数据丢失，这种情况下有很多种解决方案，铁电存储器(FRAM)就是个很好的选择。FRAM是一种具有快速写入速度的非易失性存储器，既可以进行非易失性数据存储，又可以像RAM一样操作。本文将借助飞凌嵌入式OK3568-C开发板来为大家介绍一种采

  3. 周界警戒AI算法+视频智能分析在安全生产场景中的应用

    长期以来，周界防范安防系统在大型园区、工厂、社区、机场、火车站站台、重点单位等领域应用较为广泛和常见。随着AI人工智能等新兴技术的快速发展与落地应用，通过AI智能检测与视频智能分析技术，现代化的周界安防系统可以做到全天候快速、准确地发现入侵等异常事件，并及时报警遏制。今天我们来介绍下旭帆科技安全生产周界警戒AI算法的具

  4. [2023.09.18]: Rust中类型转换在错误处理中的应用解析

    随着项目的进展，关于Rust的故事又翻开了新的一页，今天来到了服务器端的开发场景，发现错误处理中的错误类型转换有必要分享一下。Rust抽象出来了Result<T,E>，T是返回值的类型，E是错误类型。只要函数的返回值的类型被定义为Resut<T,E>，那么作为开发人员就有责任来处理调用这个函数可能发生的错误。通过Res

  5. CIIS 2023丨聚焦文档图像处理前沿领域，合合信息AI助力图像处理与内容安全保障

    近日，2023第十二届中国智能产业高峰论坛（CIIS2023）在江西南昌顺利举行。大会由中国人工智能学会、江西省科学技术厅、南昌市人民政府主办，南昌市科学技术局、中国工程科技发展战略江西研究院承办。本次大会重点关注AI大模型、生成式AI、无人系统、智能制造、数字安全等领域，汇集了来自中国工程院、国际欧亚科学院、国际核能

  6. 聚观早报｜高德发布安全出行大模型；小鹏G9焕新上市

    【聚观365】9月21日消息高德发布安全出行大模型小鹏G9焕新上市妙鸭相机上线免费版RedmiNote13Pro+支持IP68Neuralink将进行首次人体临床试验高德发布安全出行大模型高德发布了安全出行大模型。据介绍，安全出行大模型基于高德的地图大数据、位置大数据、导航大数据、智能决策系统等能力，从风险识别、风险预

  7. 悬崖边：企业如何应对网络安全漏洞趋势

    在本文中，我们将讨论企业在处理漏洞时面临的挑战，解释安全漏洞是如何引发网络攻击的，以及为什么它会导致不可接受的事件。我们还将分享我们在识别趋势性漏洞方面的经验。现代信息安全方法正在成为企业的工作流程。例如，不久前，整个IT行业都在向容器化发展，而对云环境的安全和保护机制的研究还是个新鲜事物。现在，几乎每家公司在产品架构

  8. 研究报告：周界警戒AI算法+视频智能分析在安全生产场景中的应用

    长期以来，周界防范安防系统在大型园区、工厂、社区、机场、火车站站台、重点单位等领域应用较为广泛和常见。随着AI人工智能等新兴技术的快速发展与落地应用，通过AI智能检测与视频智能分析技术，现代化的周界安防系统可以做到全天候快速、准确地发现入侵等异常事件，并及时报警遏制。今天我们来介绍下旭帆科技安全生产周界警戒AI算法的具

  9. Qt使用I.MX6U开发板上的按键（原理：将电脑键盘方向键↓在Qt中的枚举值与开发板中按键定义的枚举值一致，这样电脑端测试效果就与开发板的一致）

    在上篇介绍了Qt点亮I.MX6U开发板的一个LED，对于Qt控制I.MX6U开发板的一个蜂鸣器原理也是一样的，就不做详细介绍，具体可参考Qt控制I.MX6U开发板的一个蜂鸣器，本篇介绍Qt使用I.MX6U开发板上的按键的相关内容。文章目录1.开发板硬件图及板卡按键在电脑键盘中的对应原理2.出厂内核设备树中注册的按键3.