【Python】PySpark

前言

Apache Spark是用于大规模数据（large-scala data）处理的统一（unified）分析引擎。

简单来说，Spark是一款分布式的计算框架，用于调度成百上千的服务器集群，计算TB、PB乃至EB级别的海量数据。

Spark对Python语言的支持，重点体现在Python第三方库：PySpark

PySpark是由Spark官方开发的Python语言第三方库。

Python开发者可以使用pip程序快速的安装PySpark并像其它第三方库那样直接使用。

在这里插入图片描述

基础准备

安装

同其它的Python第三方库一样，PySpark同样可以使用pip程序进行安装。

pip install pyspark或使用国内代理镜像网站（清华大学源）
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark

构建PySpark执行环境入口对象

想要使用PySpark库完成数据处理，首先需要构建一个执行环境入口对象。

PySpark的执行环境入口对象是：类SparkContext的类对象

# 导包
from pyspark import SparkConf, SparkContext# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)# 打印PySpark的运行版本
print(sc.version)# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

运行需要Java环境，推荐jdk8

PySpark的编程模型

SparkContext类对象，是PySpark编程中一切功能的入口。

PySpark的编程，主要分为如下三大步骤：

在这里插入图片描述

数据输入

PySpark支持多种数据的输入，在输入完成后，都会得到一个：RDD类的对象

RDD全称为：弹性分布式数据集（Resilient Distributed Datasets）

PySpark针对数据的处理，都是以RDD对象作为载体，即：

  • 数据存储在RDD内
  • 各类数据的计算方法，也都是RDD的成员方法
  • RDD的数据计算方法，返回值依旧是RDD对象

在这里插入图片描述

Python数据容器转RDD对象

PySpark支持通过SparkContext对象的parallelize成员方法，将list/tuple/set/dict/str转换为PySpark的RDD对象

# 导包
from pyspark import SparkConf, SparkContext# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd1 = sc.parallelize([1, 2, 3])    
rdd2 = sc.parallelize((1, 2, 3))    
rdd3 = sc.parallelize({1, 2, 3})    
rdd4 = sc.parallelize({'key1': 'value1', 'key2': 'value2'}) 
rdd5 = sc.parallelize('hello')  # 输出RDD的内容，需要使用collect()
print(rdd1.collect())   # [1, 2, 3]
print(rdd2.collect())   # [1, 2, 3]
print(rdd3.collect())   # [1, 2, 3]
print(rdd4.collect())   # ['key1', 'key2']
print(rdd5.collect())   # ['h', 'e', 'l', 'l', 'o']# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

注意：

  • 字符串会被拆分出一个个的字符，存入RDD对象
  • 字典仅有key会被存入RDD对象

读取文件转RDD对象

PySpark也支持通过SparkContext入口对象来读取文件，构建出RDD对象。

先提前预备一个txt文件

hello
python
day

# 导包
from pyspark import SparkConf, SparkContext# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.textFile('E:\\code\\py-space\\8.27\\hello.txt')# 输出RDD的内容，需要使用collect()
print(rdd.collect())    # ['hello', 'python', 'day']# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

数据计算

RDD对象内置丰富的：成员方法（算子）

map算子

将RDD的数据一条条处理（处理的逻辑基于map算子中接收的处理函数），返回新的RDD

rdd.map(func)
# func: f:(T) -> U
# f: 表示这是一个函数
# (T) -> U 表示的是方法的定义：()表示无需传入参数，(T)表示传入1个参数
# T是泛型的代称，在这里表示 任意类型
# U是泛型的代称，在这里表示 任意类型# (T) -> U : 这是一个函数，该函数接收1个参数，传入参数类型不限，返回一个返回值，返回值类型不限
# (A) -> A : 这是一个函数，该函数接收1个参数，传入参数类型不限，返回一个返回值，返回值类型和传入参数类型一致

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([1, 2, 3, 4, 5, 6])# 通过map方法将全部数据乘以10，传入参数为函数
rdd2 = rdd.map(lambda x: x * 10)# 输出RDD的内容，需要使用collect()
print(rdd2.collect())   # [10, 20, 30, 40, 50, 60]# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

由于map()的返回值还是RDD对象，可以继续在尾部进行链式调用

rdd3 = rdd.map(lambda x: x * 10).map(lambda x: x + 9)

flatMap算子

对RDD执行map操作，然后进行解除嵌套操作。

在这里插入图片描述

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize(['a b c', 'd e f'])# 输出RDD的内容，需要使用collect()
print(rdd.map(lambda x: x.split(' ')).collect())    # [['a', 'b', 'c'], ['d', 'e', 'f']]
print(rdd.flatMap(lambda x:x.split(' ')).collect())   # ['a', 'b', 'c', 'd', 'e', 'f']# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

reduceByKey算子

针对KV型（二元元组）RDD，自动按照key分组，然后根据你提供的聚合逻辑，完成组内数据(value)的聚合操作

rdd.reduceByKey(func)
# func: (V, V) -> V
# 接收2个传入参数（类型要一致），返回一个返回值，返回值类型和传入参数类型要求一致

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([('a', 1), ('a', 1), ('b', 1), ('b', 1), ('b', 1)])# 输出RDD的内容，需要使用collect()
print(rdd.reduceByKey(lambda a, b: a+b).collect())  # [('b', 3), ('a', 2)]# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

reduceByKey中的聚合逻辑是：比如有[1,2,3,4,5]，然后聚合函数是：lambda a,b: a+b

在这里插入图片描述

注意：reduceByKey中接收的函数，只负责聚合，不理会分组；分组是自动by key来分组的

filter算子

过滤想要的数据进行保留。

rdd.filter(func)
# func: (T) -> bool
# 传入一个参数任意类型，返回值必须是True/False，返回是True的数据被保留，False的数据被丢弃

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([1, 2, 3, 4, 5, 6])# 输出RDD的内容，需要使用collect()
print(rdd.filter(lambda x: x % 2 == 0).collect())  # [2, 4, 6]# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

distinct算子

对RDD数据进行去重，返回新的RDD

rdd.distinct() # 无需传参

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([1, 2, 3, 3, 2, 6])# 输出RDD的内容，需要使用collect()
print(rdd.distinct().collect())  # [6, 1, 2, 3]# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

sortBy算子

对RDD数据进行排序，基于你指定的排序依据。

rdd.sortKey(func, ascending=False, numPartitions=1)
# func: (T) -> U：告知按照RDD中的哪个数据进行排序，比如lambda x: x[1]表示按照RDD中的第二列元素进行排序
# ascending：True升序，False降序
# numPartitions：用多少分区排序，全局排序需要设置为1

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([('Aiw', 9), ('Tom', 6), ('Jack', 8), ('Bolb', 5)])# 输出RDD的内容，需要使用collect()
print(rdd.sortBy(lambda x: x[1], ascending=False,numPartitions=1).collect())  # [('Aiw', 9), ('Jack', 8), ('Tom', 6), ('Bolb', 5)]# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

数据输出

collect算子

将RDD各个分区内的数据，统一收集到Driver中，形成一个List对象。

rdd.collect()
# 返回值是一个List

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([1, 2, 3])rdd_list: list = rdd.collect()print(rdd_list)   # [1, 2, 3]
print(type(rdd_list))   # <class 'list'># 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

reduce算子

对RDD数据集按照你传入的逻辑进行聚合

rdd.reduce(func)
# func：(T, T) -> T
# 传入2个参数，1个返回值，要求返回值和参数类型一致

在这里插入图片描述

示例：

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize(range(1, 10))print(rdd.reduce(lambda a, b: a+b))   # 45# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

take算子

取RDD的前N个元素，组合成List进行返回。

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize(range(1, 10))rdd_take: list = rdd.take(3)print(rdd_take)   # [1, 2, 3]
print(type(rdd_take))   # <class 'list'># 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

count算子

计算RDD有多少条数据，返回值是一个数字。

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize(range(1, 10))rdd_count: int = rdd.count()print(rdd_count)   # 9
print(type(rdd_count))   # <class 'int'># 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

saveAsTextFile算子

将RDD的数据写入文本文件中。支持本地写出、HDFS等文件系统。

注意事项：

在这里插入图片描述

# 导包
from pyspark import SparkConf, SparkContext, sql
import os# 设置环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/Python/python.exe'
os.environ['HADOOP_HOME'] = 'D:/Hadoop-3.0.0'# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)rdd = sc.parallelize(range(1, 10))rdd.saveAsTextFile('./8.27/output') # 运行之前确保输出文件夹不存在，否则报错# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

上述代码输出结果，输出文件夹内有多个分区文件

修改RDD分区为1个

方式一：SparkConf对象设置属性全局并行度为1：

# 创建SparkConf类对象
conf = SparkConf().setMaster('local[*]').setAppName('test_spark_app')
# 设置属性全局并行度为1
conf.set('spark.default.parallelism','1')
# 基于SparkConf类对象创建SparkContext类对象
sc = SparkContext(conf=conf)

方式二：创建RDD的时候设置（parallelize方法传入numSlices参数为1）

rdd = sc.parallelize(range(1, 10), numSlices=1)
rdd = sc.parallelize(range(1, 10), 1)

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

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

相关文章

大集合按照指定长度进行分割成多个小集合，用于批量多次处理数据

大集合按照指定长度进行分割成多个小集合，用于批量多次处理数据

&#x1f4da;目录 拆分案例拆分的核心代码 通常我们对集合的更新或者保存都需要用集合来承载通过插入的效率&#xff0c;但是这个会遇到一个问题就是你不知道那天那个集合的数量可能就超了&#xff0c;虽然我们连接数据库进行批量提交会在配置上配置allowMultiQueriestrue,但是…
阅读更多...
AMBEO 双声道空间音频现已迈进直播制作领域

AMBEO 双声道空间音频现已迈进直播制作领域

图片来源&#xff1a;Unsplash&#xff0c;作者&#xff1a;Bence Balla-Schottner AMBEO 双声道空间音频现已迈进直播制作领域 为所有观众解锁更加身临其境的听觉体验 森海塞尔将功能强大的 AMBEO 双声道空间音频技术引入了广播电视直播应用领域&#xff0c;对所有体育赛事广…
阅读更多...
【Flutter】使用Android Studio 创建第一个flutter应用。

【Flutter】使用Android Studio 创建第一个flutter应用。

前言 首先下载好 flutter sdk和 Android Studio。 FlutterSDK下载 Android Studio官网 配置 我的是 windows。 where.exe flutter dart查看flutter安装环境。 如果没有&#xff0c;自己在环境变量的path添加下flutter安装路径。 在将 Path 变量更新后&#xff0c;打开一个…
阅读更多...
运维Shell脚本小试牛刀(二)

运维Shell脚本小试牛刀(二)

运维Shell脚本小试牛刀(一) 运维Shell脚本小试牛刀(二) 运维Shell脚本小试牛刀(三)::$(cd $(dirname $0)&#xff1b; pwd)命令详解 一: if---else.....fi 条件判断演示 [rootwww shelldic]# cat checkpass.sh #!/bin/bash - # # # # FILE: ch…
阅读更多...
npm和yarn的区别？

npm和yarn的区别？

文章目录 前言npm和yarn的作用和特点npm和yarn的安装的机制npm安装机制yarn安装机制检测包解析包获取包链接包构建包 总结后言 前言 这一期给大家讲解npm和yarn的一些区别 npm和yarn的作用和特点 包管理&#xff1a;npm 和 yarn 可以用于安装、更新和删除 JavaScript 包。它们提…
阅读更多...
哈希的应用——布隆过滤器

哈希的应用——布隆过滤器

✅<1>主页&#xff1a;&#xff1a;我的代码爱吃辣 &#x1f4c3;<2>知识讲解&#xff1a;数据结构——位图 ☂️<3>开发环境&#xff1a;Visual Studio 2022 &#x1f4ac;<4>前言&#xff1a;布隆过滤器是由布隆&#xff08;Burton Howard Bloom&…
阅读更多...
忘记密码-小米机型 其他安卓机型账号锁 设备锁的分析与刷写某第三方解锁包输入“注册码”

忘记密码-小米机型 其他安卓机型账号锁 设备锁的分析与刷写某第三方解锁包输入“注册码”

重要提示&#xff1a; 博文的主要目的是分析安卓机型账号锁的安全性和解决方法。操作仅限于自己的机型忘记密码 手机号不用过了保修期导致无法通过官方解锁的操作&#xff0c;请勿用于非法途径 在开始前。对于锁的认知可以参考这篇博文 安卓搞机玩机-什么是“锁 ” BL锁 屏幕锁…
阅读更多...
计算机竞赛 基于大数据的股票量化分析与股价预测系统

计算机竞赛 基于大数据的股票量化分析与股价预测系统

文章目录 0 前言1 课题背景2 实现效果3 设计原理QTChartsarma模型预测K-means聚类算法算法实现关键问题说明 4 部分核心代码5 最后 0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; 基于大数据的股票量化分析与股价预测系统 该项目较为新颖…
阅读更多...
六、事务-4.并发事务问题

六、事务-4.并发事务问题

一、脏读 事务A执行3个操作&#xff0c;第1个操作执行select语句&#xff0c;第2个操作执行update语句。 注意&#xff1a;事务没有执行完成的时候&#xff0c;事务是没有提交的。只有事务的3个操作完成之后&#xff0c;事务才会提交。 但事务A中第2个操作&#xff0c;会把表…
阅读更多...
【阻塞队列】

【阻塞队列】

文章目录 普通队列存在的问题单锁实现双锁实现 普通队列存在的问题 大部分场景要求分离向队列放入&#xff08;生产者&#xff09;、从队列拿出&#xff08;消费者&#xff09;两个角色、它们得由不同的线程来担当&#xff0c;而之前的实现根本没有考虑线程安全问题队列为空&a…
阅读更多...
【python爬虫】—豆瓣电影Top250

【python爬虫】—豆瓣电影Top250

豆瓣电影Top250 豆瓣榜单简介需求描述Python实现 豆瓣榜单简介 豆瓣电影 Top 250 榜单是豆瓣网站上列出的评分最高、受观众喜爱的电影作品。这个榜单包含了一系列优秀的影片&#xff0c;涵盖了各种类型、不同国家和时期的电影。 需求描述 使用python爬取top250电影&#xff…
阅读更多...
xss-labs靶场通关详解

xss-labs靶场通关详解

文章目录 前言level1level2level3level4level5level6level7level8level9level10level11level12level13level14level15level16level17level18level19&level20 前言 赶着假期结尾的时候&#xff0c;赶紧给自己找点任务做。现在对xss还是一知半解&#xff0c;只是了解个大概&a…
阅读更多...
向函数传递参数（传地址）

向函数传递参数（传地址）

过往课程 向函数传递参数&#xff08;传值、传引用、传const引用&#xff09; 传地址 向函数传地址&#xff0c;是指将变量的地址传递给函数。 函数通过声明参数为地址变量来接收一个变量的地址。 示例如下&#xff1a; #include <iostream> using namespace std;v…
阅读更多...
Nginx百科之gzip压缩、黑白名单、防盗链、零拷贝、跨域、双机热备

Nginx百科之gzip压缩、黑白名单、防盗链、零拷贝、跨域、双机热备

引言 早期的业务都是基于单体节点部署&#xff0c;由于前期访问流量不大&#xff0c;因此单体结构也可满足需求&#xff0c;但随着业务增长&#xff0c;流量也越来越大&#xff0c;那么最终单台服务器受到的访问压力也会逐步增高。时间一长&#xff0c;单台服务器性能无法跟上业…
阅读更多...
（JAVA）树——tree

（JAVA）树——tree

阅读更多...
嵌入式Linux开发实操（十五）：nand flash接口开发

嵌入式Linux开发实操（十五）：nand flash接口开发

# 前言 flash memory，分NAND和NOR： 如果说nor flash有个特点就是能执行代码，NOR并行接口具有地址和数据总线，spi flash更是主要用于存储代码，SPI（或QSPI）NOR代码可就地执行（XiP），一般系统要求flash闪存提供相对较高的频率和数据缓存的clocking。而nand flash主要用于…
阅读更多...
每天刷题五道RHCSA/6-10题（Radhat8.2）

每天刷题五道RHCSA/6-10题（Radhat8.2）

6.创建协作目录权限 mkdir /home/managers chown :sysmgrs /home/managers chmod 2770 /home/managers 测试&#xff1a; touch /home/managers/12345 ll /home/managers/12345 7.配置NTP systemctl status chronyd #查看状态 yum -y install chrony #如果没有安装&#xff0c…
阅读更多...
element ui-Pagination

element ui-Pagination

页面分为两个表格&#xff0c;当两边的表格数据量大时&#xff0c;分页样式就会受到影响&#xff0c;可以将跳转按钮的个数减少 页面分页代码如下 页面效果
阅读更多...
rabbitmq的优先级队列

rabbitmq的优先级队列

在我们系统中有一个 订单催付 的场景&#xff0c;我们的客户在天猫下的订单 , 淘宝会及时将订单推送给我们&#xff0c;如果在用户设定的时间内未付款那么就会给用户推送一条短信提醒&#xff0c;很简单的一个功能对吧&#xff0c;但是&#xff0c;tianmao商家对我们来说&#…
阅读更多...
HTML学习笔记02

HTML学习笔记02

HTML笔记02 页面结构分析 元素名描述header标题头部区域的内容&#xff08;用于页面或页面中的一块区域&#xff09;footer标记脚部区域的内容&#xff08;用于整个页面或页面的一块区域&#xff09;sectionWeb页面中的一块独立区域article独立的文章内容aside相关内容或应用…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023