Spark RDD 操作实战

Spark RDD 基础

更多spark相关知识请查看官方接口文档

PySpark是Spark的PythonAPI,允许Python调用Spark编程模型。

配置spark环境

!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q www-us.apache.org/dist/spark/spark-2.4.8/spark-2.4.8-bin-hadoop2.7.tgz  
!tar xf spark-2.4.8-bin-hadoop2.7.tgz
!pip install -q findspark
import os
os.environ["JAVA_HOME"]="/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"]="/content/spark-2.4.8-bin-hadoop2.7"
import findspark
findspark.init()

初始化Spark

SparkContext

from pyspark import SparkContext
sc = SparkContext(master = 'local[2]')

SparkContext信息获取

sc.version   #获取SparkContext版本
'2.4.8'
sc.pythonVer   #获取Python版本
'3.7'
sc.master   #要连接的MasterURL
'local[2]'
str(sc.sparkHome)   #Spark在工作节点的安装路径
'None'
str(sc.sparkUser())   #获取SparkContext的Spark用户名
'root'
sc.appName   #返回应用名称
'pyspark-shell'
sc.applicationId   #获取应用程序ID
'local-1623220140497'
sc.defaultParallelism   #返回默认并行级别
2
sc.defaultMinPartitions   #RDD默认最小分区数
2

配置

from pyspark import SparkConf, SparkContextconf = (SparkConf() \.setMaster("local") \.setAppName("My app") \.set("spark.executor.memory", "1g"))sc = SparkContext.getOrCreate(conf = conf)

使用Shell

PySpark Shell已经为SparkContext创建了名为 sc 的变量。

$./bin/spark-shell --master local[2]        #命令行启动spark
$./bin/pyspark --master local[4] --py-files code.py     #命令行提交spark脚本任务

用 --master 参数设定 Context 连接到哪个Master 务器,通过传递逗号分隔列表至 --py-files 添加 Python.zip、.egg 或 .py文件到 Runtime 路径。

加载数据

并行集合

rdd = sc.parallelize([('a',7),('a',2),('b',2)])
rdd2 = sc.parallelize([('a',2),('d',1),('b',1)])
rdd3 = sc.parallelize(range(100))
rdd4 = sc.parallelize([("a",["x","y","z"]), ("b",["p", "r"])])

外部数据

使用textFile()函数从HDFS、本地文件或其它支持Hadoop的文件系统里读取文本文件,或使用wholeTextFiles()函数读取目录里文本文件。

textFile = sc.textFile("/my/directory/*.txt")
# 如果是在google colab中可以运行下方代码
# textFile = sc.textFile("sample_data/california_housing_train.csv")
textFile2 = sc.wholeTextFiles("/my/directory/")
# 如果是在google colab中可以运行下方代码
# textFile2 = sc.wholeTextFiles("sample_data/")

提取RDD信息

基础信息

rdd.getNumPartitions()   #列出分区数
2
rdd.count()   #计算RDD实例数量
3
rdd.countByKey()   #按键计算RDD实例数量
defaultdict(int, {'a': 2, 'b': 1})
rdd.countByValue()   #按值计算RDD实例数量
defaultdict(int, {('a', 2): 1, ('a', 7): 1, ('b', 2): 1})
rdd.collectAsMap()   #以字典形式返回键值
{'a': 2, 'b': 2}
rdd3.sum()   #RDD元素求和
4950
sc.parallelize([]).isEmpty()   #检查RDD是否为空
True

汇总

rdd3.max()   #RDD元素的最大值
99
rdd3.min()   #RDD元素的最小值
0
rdd3.mean()   #RDD元素的平均值
49.5
rdd3.stdev()   #RDD元素的标准差
28.86607004772212
rdd3.variance()   #RDD元素的方差
833.25
rdd3.histogram(3)   #分箱(Bin)生成直方图
([0, 33, 66, 99], [33, 33, 34])
rdd3.stats()   #综合统计包括:计数、平均值、标准差、最大值和最小值
(count: 100, mean: 49.5, stdev: 28.86607004772212, max: 99.0, min: 0.0)

应用函数

map与flatmap函数

rdd.map(lambda x: x+(x[1],x[0])).collect()   #对每个RDD元素执行函数
[('a', 7, 7, 'a'), ('a', 2, 2, 'a'), ('b', 2, 2, 'b')]
rdd5=rdd.flatMap(lambda x: x+(x[1],x[0]))   #对每个RDD元素执行函数,并拉平结果
rdd5.collect()
['a', 7, 7, 'a', 'a', 2, 2, 'a', 'b', 2, 2, 'b']
rdd4.flatMapValues(lambda x: x).collect()   #不改变键,对rdd4的每个键值对执行flatMap函数
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

选择数据

获取

rdd.collect()   #返回包含所有RDD元素的列表
[('a', 7), ('a', 2), ('b', 2)]
rdd.filter(lambda x: "a" in x) .collect()   #提取前两个RDD元素
[('a', 7), ('a', 2)]
rdd.first()   #提取第一个RDD元素
('a', 7)
rdd5.distinct().collect()   #提取前两个RDD元素
[2, 'b', 'a', 7]

抽样

rdd3.sample(False, 0.15, 81).collect()   #返回rdd3的采样子集
[3, 4, 27, 28, 35, 41, 43, 49, 51, 55, 64, 65, 66, 67, 85, 88, 89, 92]

筛选

 rdd.filter(lambda x: "a" in x) .collect()   #筛选RDD
[('a', 7), ('a', 2)]
rdd5.distinct().collect()   #返回RDD里的唯一值
[2, 'b', 'a', 7]
rdd.keys().collect()   #返回RDD键值对里的键
['a', 'a', 'b']

迭代

foreach函数迭代

def g(x):print(x)
rdd.foreach(g)   #为所有RDD应用函数

改变数据形状

Reduce操作

rdd.reduceByKey(lambda x,y : x+y).collect()   #合并每个键的RDD值
[('b', 2), ('a', 9)]
rdd.reduce(lambda a, b: a + b)   #合并RDD的值
('a', 7, 'a', 2, 'b', 2)

分组

# rdd3.groupBy(lambda x: x % 2).mapValues(list).collect()   #返回RDD的分组值
rdd.groupByKey().mapValues(list).collect()   #按键分组RDD
[('b', [2]), ('a', [7, 2])]

聚合

seqOp = (lambda x,y: (x[0]+y,x[1]+1))
combOp = (lambda x,y:(x[0]+y[0],x[1]+y[1]))
add = (lambda x,y:x+y)
rdd3.aggregate((0,0), seqOp, combOp)   #汇总每个分区里的RDD元素,并输出结果
(4950, 100)
rdd.aggregateByKey((0,0), seqOp, combOp).collect()   #汇总每个RDD的键的值
[('b', (2, 1)), ('a', (9, 2))]
rdd3.fold(0, add)   #汇总每个分区里的RDD元素,并输出结果
4950
rdd.foldByKey(0, add).collect()   #合并每个键的值
[('b', 2), ('a', 9)]
# rdd3.keyBy(lambda x: x+x).collect()   #通过执行函数,创建RDD元素的元组

数学运算

RDD运算

rdd.subtract(rdd2).collect()   #返回在rdd2里没有匹配键的rdd键值对
[('b', 2), ('a', 7)]
rdd2.subtractByKey(rdd).collect()   #返回rdd2里的每个(键,值)对,rdd中没有匹配的键
[('d', 1)]
rdd.cartesian(rdd2).collect()   #返回rdd和rdd2的笛卡尔积
[(('a', 7), ('a', 2)),(('a', 7), ('d', 1)),(('a', 7), ('b', 1)),(('a', 2), ('a', 2)),(('b', 2), ('a', 2)),(('a', 2), ('d', 1)),(('a', 2), ('b', 1)),(('b', 2), ('d', 1)),(('b', 2), ('b', 1))]

排序

RDD排序

rdd2.sortBy(lambda x: x[1]).collect()   #按给定函数排序
[('d', 1), ('b', 1), ('a', 2)]
rdd2.sortByKey() .collect()   #RDD按键排序RDD的键值对
[('a', 2), ('b', 1), ('d', 1)]

重分区

repartition函数

rdd.repartition(4)   #新建一个含4个分区的RDD
MapPartitionsRDD[104] at coalesce at NativeMethodAccessorImpl.java:0
rdd.coalesce(1)   #将RDD中的分区数缩减为1个
CoalescedRDD[105] at coalesce at NativeMethodAccessorImpl.java:0

保存

存储RDD到本地或HDFS

rdd.saveAsTextFile("rdd.txt")
rdd.saveAsHadoopFile("hdfs://namenodehost/parent/child", 'org.apache.hadoop.mapred.TextOutputFormat')

终止SparkContext

停止SparkContext

sc.stop()

执行脚本程序

提交脚本执行

$./bin/spark-submit examples/src/main/python/pi.py

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

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

相关文章

16、Go Gin 深入理解Gin中间件

中间件介绍 Gin框架允许开发者在处理请求的过程汇总,加入用户自己的钩子(Hook)函数 这个钩子函数就叫中间件,中间件适合处理一些公共的业务逻辑 比如登录认证,权限校验、数据分页、记录日志、耗时统计等 1、全局中…

STM32通用定时器的应用实例(基于STM32F103)

目录 概述 1 STM32Cube配置项目 1.1 准备环境 1.2 配置项目参数 1.3 生成Project 2 HAL函数 2.1 初始化函数:HAL_TIM_Base_Init 2.2 中断模式启动定时器函数:HAL_TIM_Base_Start 2.3 定时器回调函数: HAL_TIM_PeriodElapsedCallback…

ElasticSearch操作之重置密码脚本

ElasticSearch操作之重置密码脚本 #!/bin/bash # 使用样例 ./ES密码重置.sh 旧密码 新密码# 输入旧密码 es_old_password$1# 设置新的密码变量 es_password$2# 正确响应 es_reponse{"acknowledged":true}# 检查Elasticsearch是否在运行 if pgrep -f elasticsearch &g…

2024年5月计算机视觉论文推荐:包括扩散模型、视觉语言模型、图像编辑和生成、视频处理和生成以及图像识别等各个主题

我们今天总结下2024年5月发表的最重要的论文,重点介绍了计算机视觉领域的最新研究和进展,包括扩散模型、视觉语言模型、图像编辑和生成、视频处理和生成以及图像识别等各个主题。 Diffusion Models 1、Dual3D: Efficient and Consistent Text-to-3D Ge…

数据赋能(98)——概念:数据整理、数据整合、数据处理

此文为本人学习与提高能力的笔记。 数据整理、数据整合与数据处理这三个术语,尽管在不少情境下可能被看作是近义词或拥有类似的语义范畴,但为了确保术语使用的精准度和专业性,我们必须对其有更为深入的认知。这三个概念虽在某种程度上相互交…

【C++题解】1697. 请输出n~1之间所有的整数

问题:1697. 请输出n~1之间所有的整数 类型:循环 题目描述: 从键盘读入一个整数 n ,请输出 n∼1 之间所有的整数,每行输出 1 个。 比如,假设读入 n5 ,输出结果如下: 5 4 3 2 1 输入&#xff1…

php 亚马逊AWS-S3对象存储上传文件

最近做国外项目的时候,需要把文件上传到AWS-S3对象存储空间里,下面整理一下上传方法,和碰到的问题 代码 /*** 亚马逊oss Aws上传* composer require aws/aws-sdk-php* param $filePath* param $ossPath* return array* author wzb* data 202…

Java进阶学习笔记4——Static应用知识:代码块

代码块: 代码块是类的五大成员之一(成员变量、构造器、方法、代码块、内部类)。 Java类生命周期:加载、验证、准备、初始化、卸载。 代码块分为两种: 静态代码块: 格式:static {} 特点&…

SpringBoot 集成 Nebula

工作需求&#xff0c;开始了解图数据库&#xff0c;经过工具选型&#xff0c;最终选择nebula graph&#xff0c;并集成到springboot&#xff0c;java 环境下如何对 Nebula Graph 进行操作&#xff0c;本文整理下过程。 1、首先引入 pom 依赖 <dependency><groupId&g…

如何理解栈的增长方向

1、概念 函数调用栈&#xff0c;简称栈。不管是函数的执行还是函数调用&#xff0c;栈都起着非常重要的作用 保存函数的局部变量向被调用函数传递参数返回函数的返回值保存函数的返回地址。返回地址是指从被调用函数返回后调用者应该继续执行的指令地址 该文章会帮你很好的了…

Flutter 中的 ClipPath 小部件:全面指南

Flutter 中的 ClipPath 小部件&#xff1a;全面指南 在Flutter中&#xff0c;ClipPath是一个强大的小部件&#xff0c;它允许你通过一个路径来裁剪其子组件。这意味着你可以将任何小部件裁剪成圆形、矩形、或者任何你能通过Path定义的形状。本文将提供一个全面的指南&#xff…

AWS数据库之MemoryDB for Redis

NoSQL&#xff08;非关系型数据库&#xff09;通常可以分为内存数据库、文档数据库、图数据库、时序数据库等等。 适用于 Redis 的 MemoryDB 是一项耐用的内存数据库服务&#xff0c;可提供超快性能。它专为采用微服务架构的现代应用程序而构建。 MemoryDB 与 Redis&#xff…

鸿蒙系统与OpenHarmony:中国科技行业的新动力与就业前景

背景 经历近年来的迅猛发展&#xff0c;鸿蒙原生应用数量已突破4000款&#xff0c;生态设备数量超过8亿台&#xff0c;开发者群体壮大至220万人。更为显著的是&#xff0c;鸿蒙系统在中国市场的份额已经超过了15%&#xff0c;稳居第三大操作系统&#xff0c;其生态之树已然枝繁…

人工智能时代,大模型如何崛起?

在人工智能时代&#xff0c;大模型的崛起可以归因于以下几个关键因素&#xff1a; 首先&#xff0c;强大的计算能力是基础。随着硬件技术的不断进步&#xff0c;如高性能的 GPU 集群等&#xff0c;能够为训练大模型提供足够的算力支持&#xff0c;使得大规模数据的处理和复杂模…

探秘机器学习经典:K-近邻算法(KNN)全解析

在浩瀚的机器学习宇宙中,K-近邻算法(K-Nearest Neighbors,简称KNN)如同一颗璀璨的明星,以其简洁直观的原理和广泛的应用范围,赢得了众多数据科学家的喜爱。今天,让我们一起揭开KNN的神秘面纱,深入探讨它的运作机制、优缺点、应用场景,以及如何在实际项目中灵活运用。 …

23. 【Java教程】接口

本小节我们将学习 Java 接口&#xff08;interface&#xff09;&#xff0c;通过本小节的学习&#xff0c;你将了解到什么是接口、为什么需要接口、如何定义和实现接口&#xff0c;以及接口的特点等内容。最后我们也将对比抽象类和接口的区别。 1. 概念 Java 接口是一系列方法的…

PostgreSQL事务基础理解

PostgreSQL事务 事务是数据库管理系统执行过程中的一个逻辑单位&#xff0c;由一个有限的数据库操作序列构成。数据库事务通常包含一个序列对数据库的读和写操作&#xff0c;主要是包含以下两个目的&#xff1a; 为数据库操作序列提供一个从失败中恢复到正常状态的方法&#…

事务管理控制

文章目录 1. 事务的基本概念2. 数据库的并发控制2.1 事务调度2.2 并发操作带来的问题2.3 并发调度的可串行性2.4 并发控制技术2.5 两段锁协议2.6 多粒度封锁协议 3. 数据库的备份与恢复3.1 数据库系统故障3.2 数据库的备份3.3 数据库的恢复 4. 数据库的安全性与完整性4.1 数据库…

VMware虚拟机桥接无线网卡上网(WIFI)

一、打开VM点击【编辑】-【虚拟网络编辑器】 二、点击【桥接模式】- 点击【自动设置】- 选择自己的无线网适配器 - 【确定】 三、开机之后会弹出提示连接网络&#xff0c;就能看见网络已经连上了

网络变压器原理与维修视频

如果你在寻找网络变压器原理与维修方面的视频教程&#xff0c;可以在网上找一些优质的资料进行学习。一些国内外知名的教育和科技资源网站&#xff0c;如B站、优酷、YouTube等&#xff0c;都有涉及网络变压器原理和维修的视频教程&#xff0c;你可以在这些网站上进行搜索。 通常…