spark的使用
spark的使用
spark是一款分布式的计算框架,用于调度成百上千的服务器集群。
安装pyspark
# os.environ['PYSPARK_PYTHON']='解析器路径' pyspark_python配置解析器路径
import os
os.environ['PYSPARK_PYTHON']="D:/dev/python/python3.11.4/python.exe"
pip install pyspark # 原始国外安装
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark #网址安装
java安装
前置安装软件java包
java官网下载地址
一键下一步安装,配置环境变量
首先创建一个JAVA_HOME的全局变量然后在path中通过%%引入执行下面的bin 路径%JAVA_HOME%\bin


执行成功
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()
PySpark的数据计算,都是基于RDD对象来进行的,RDD对象内置丰富的:成员方法(算子)
map算子
功能:map算子,是将RDD的数据一条条处理,处理的逻辑基于map算子中接收的处理函数,返回新的RDD语法:

# 简单执行map将数据乘以10返回,如果不引入python解析器的路径引入就会报错,
from pyspark import SparkConf, SparkContext
# 指定spark的python解析器路径
import os
os.environ['PYSPARK_PYTHON']="D:/dev/python/python3.11.4/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])def func(data):return data * 10# map传入一个参数有返回值,是函数或者是值
rdd2 = rdd.map(func)
print(rdd2.collect())

flatMap
flatMap跟map差不多就是在最后做了一个解除嵌套的功能
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON']="D:/dev/python/python3.11.4/python.exe"
# 创建sparkConf 类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
# 创建sparkConf类对象创建sparkContext对象
sc = SparkContext(conf=conf)rdd = sc.parallelize(['中石科技 时间还复活甲 如今房价','慰问金 咖啡机 姐夫哥','格很高 客服管家二恶烷 可归结为'])rdd2 = rdd.flatMap(lambda x:x.split(' '))print(rdd2.collect())

map的结果

reduceByKey
reduceByKey对数据进行分组可以两两计算
from pyspark import SparkConf, SparkContext
import osos.environ['PYSPARK_PYTHON'] = "D:/dev/python/python3.11.4/python.exe"
# 创建sparkConf 类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
# 创建sparkConf类对象创建sparkContext对象
sc = SparkContext(conf=conf)rdd = sc.parallelize([('男', 11), ('男', 22), ('女', 21), ('男', 31), ('女', 99)])
# 把男女进行分组value值进行计算
rdd2 = rdd.reduceByKey(lambda a, b:a+b)print(rdd2.collect()) # [('女', 120), ('男', 64)]
reduce
与reduce的区别就是没有进行分组
take
取出前几个数据
...
rdd = sc.parallelize([1,2,3,4,5]).take(3) # [1,2,3]
count
计算rdd中的数据个数
filter

from pyspark import SparkConf,SparkContext
import os
os.environ['PYSPARK_PYTHON']='D:/dev/python/python3.11.4/python.exe'conf=SparkConf().setMaster('local[*]').setAppName('test_spark')
sc=SparkContext(conf=conf)rdd=sc.parallelize([1,2,3,4,5])rdd2=rdd.filter(lambda a:a%2==0)
print(rdd2.collect()) # [2,4]
distinct
进行数据去重
from pyspark import SparkConf,SparkContext
import os
os.environ['PYSPARK_PYTHON']='D:/dev/python/python3.11.4/python.exe'conf=SparkConf().setMaster('local[*]').setAppName('test_spark')
sc=SparkContext(conf=conf)add= sc.parallelize([1,2,3,4,5,6,73,3,2,4,56,3,5])add2=add.distinct()
print(add2.collect()) # [56, 1, 73, 2, 3, 4, 5, 6]
sortBy排序
from pyspark import SparkConf, SparkContext
import osos.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python3.11.4/python.exe'conf = SparkConf().setMaster('local[*]').setAppName('test_spark')
sc = SparkContext(conf=conf)add = sc.textFile('D:/wordText.txt')word_rdd = add.flatMap(lambda x: x.split(' '))
word_with_rdd = word_rdd.map(lambda word: (word, 1))
result_rdd =word_with_rdd.reduceByKey(lambda a,b:a+b)
result_num=result_rdd.sortBy(lambda x:x[1],ascending=False,numPartitions=1) # 1.根据什么排序,2.True 升序 False降序 3.分布式分区
print(result_num.collect())
collect
将rdd内容变成list,从而就可以打印出来
spark写入文件
首先安装
- 下载Hadoop安装包Hadoop安装包
- 然后把hadoop.dll放入指定文件夹内

os.environ['HADOOP_HOME']='D:/dev/hadoop/hadoopjob3.0'
conf = SparkConf().setMaster('local[*]').setAppName('test_spark')
sc = SparkContext(conf=conf)rdd2=sc.parallelize([[1,3,5],[6,7,9]])
rdd2.saveAsTextFile('D:/output1')

这样创建出来的文件就有16个分区,因为我的是16内核
如果想要在一个分区就要设置参数
import ...
os.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python3.11.4/python.exe'
os.environ['HADOOP_HOME']='D:/dev/hadoop/hadoopjob3.0'
conf = SparkConf().setMaster('local[*]').setAppName('test_spark')
# 第一种
conf.set("spark.default.parallelism",'1') # 设置一个分区
sc = SparkContext(conf=conf)# rdd2=sc.parallelize([[1,3,5],[6,7,9]])
# 第二种设置一个分区
rdd2=sc.parallelize([[1,3,5],[6,7,9]],1) # numSlices=1 参数可以不写直接传1
rdd2.saveAsTextFile('D:/output1')
相关文章:
spark的使用
spark的使用 spark是一款分布式的计算框架,用于调度成百上千的服务器集群。 安装pyspark # os.environ[PYSPARK_PYTHON]解析器路径 pyspark_python配置解析器路径 import os os.environ[PYSPARK_PYTHON]"D:/dev/python/python3.11.4/python.exe"pip inst…...
力扣:66. 加一(Python3)
题目: 给定一个由 整数 组成的 非空 数组所表示的非负整数,在该数的基础上加一。 最高位数字存放在数组的首位, 数组中每个元素只存储单个数字。 你可以假设除了整数 0 之外,这个整数不会以零开头。 来源:力扣…...
Go的标准库Context理解
作为一个才入门的菜鸟,还没写过真正的 go 项目,要理解这个 Context 还是有点难,不过还是要尝试一下。在 Go http包的Server中,每一个请求在都有一个对应的 goroutine 去处理。请求处理函数通常会启动额外的 goroutine 用来访问后端…...
Nuxt3_1_路由+页面+组件+资源+样式 使用及实例
1、 简介 1.1 开发必备 node版本 v16.10.0 我使用的是16.14.0编辑器推荐使用Volar Extension 的VS code插件Terminal 运行nuxt指令 1.2 环境搭建 安装项目: npx nuxilatest init [first_nuxt3]进入项目目录: cd [first_nuxt3]安装依赖:n…...
Kafka第三课
Flume 由三部分 Source Channel Sink 可以通过配置拦截器和Channel选择器,来实现对数据的分流, 可以通过对channel的2个存储容量的的设置,来实现对流速的控制 Kafka 同样由三大部分组成 生产者 服务器 消费者 生产者负责发送数据给服务器 服务器存储数据 消费者通过从服务器取…...
elasticsearch修改es集群的索引副本数量
前言 最近es集群进行调整,从2节点变成了单节点。所以需要将集群模式改为单点模式,并需要将es 集群的全部索引副本个数改为0,不然会有很多未分配的分片,导致集群状态为 yellow。 具体实践 1. 先将现有的index的副本数量为0个 此…...
【SpringCloud】Ribbon定制化配置
文章目录 使用Ribbon自带负载均衡算法添加负载均衡算法ConfigurationRestTemplate使用上面负载均衡算法 自定义负载均衡算法负载均衡算法实现RestTemplate在Controller中使用该负载均衡算法ServiceIInstance解释 使用Ribbon自带负载均衡算法 添加负载均衡算法Configuration /…...
Mac terminal 每次打开都要重新配置文件
1. 问题描述 每次打开 Terminal,base_profile文件中配置的内容就不生效,需要重新执行source ~/.bash_profile才可以使用。 2. 原因分析 zsh加载的是~/.zshrc文件,而.zshrc 文件中并没有定义任务环境变量。 3. 解决办法 在~/.zshrc文件末尾添…...
el-button实现按钮,鼠标移入显示,移出隐藏
2023.8.18今天我学习了 如何实现鼠标移入显示按钮,鼠标移出隐藏按钮。 效果如图: 鼠标移入时: 鼠标移出时: mouseover //鼠标移入事件 mouseleave //鼠标移出事件 原本我是想直接在el-button写入这两个方法,但是elem…...
uniapp+uview封装小程序请求
提要: uniapp项目引入uview库 此步骤不再阐述 1.创建环境文件 env.js: let BASE_URL;if (process.env.NODE_ENV development) {// 开发环境BASE_URL 请求地址; } else {// 生产环境BASE_URL 请求地址; }export default BASE_URL; 2.创建请求文件 该…...
idea常见错误大全之:解决全局搜索失效+搜索条件失效(条件为空)+F8失灵
问题一:全局搜索快捷键ctrlshiftf 突然失灵了,键盘敲烂了 都没反应,这是为什么呢? 肯定不是idea本身的原因,那么就是其它外在因素影响到了idea的快捷键,那么其它的快捷键为什么没失效呢,原因只有…...
【论文阅读】基于深度学习的时序预测——LTSF-Linear
系列文章链接 论文一:2020 Informer:长时序数据预测 论文二:2021 Autoformer:长序列数据预测 论文三:2022 FEDformer:长序列数据预测 论文四:2022 Non-Stationary Transformers:非平…...
02.FFMPEG的安装和添加硬件加速自编译
说一个极其郁闷的事情,就在昨天收到3399的一块板子后,往电脑上面一插,然后悲剧的事情就发生了,我的电脑蓝屏重启了,这下好了,我写到一半的帖子也不见了,我的SSH里面的记录全部消失了,…...
elementUI 的上传组件<el-upload>,自定义上传按钮样式
方法一: 原理:调用<el-upload>组件的方法唤起选择文件事件 效果: 页面代码: 1、选择图片按钮 <div class"flex_row_spacebetween btn" click"chooseImg"><span class"el-icon-plus ic…...
【卷积神经网络】卷积,池化,全连接
随着计算机硬件的升级与性能的提高,运算量已不再是阻碍深度学习发展的难题。卷积神经网络(Convolution Neural Network,CNN)是深度学习中一项代表性的工作,CNN 是受人脑对图像的理解过程启发而提出的模型,其…...
【SA8295P 源码分析】76 - Thermal 功耗 之 /dev/thermalmgr 相关调试命令汇总
【SA8295P 源码分析】76 - Thermal 功耗 之 /dev/thermalmgr 相关调试命令汇总 1、配置文件:/mnt/etc/system/config/thermal-engine.conf2、获取当前SOC所有温度传感器的温度:cat /dev/thermalmgr3、查看所有 Thermal 默认配置和自定义配置:echo query config > /dev/th…...
以太网(一):PoE供电
一、定义: PoE系统包括供电端设备(PSE)和受电端设备(PD)两部分PoE(Power over Ethernet):是一种可以在以太网中透过双绞线来传输电力与数据到设备上的技术PSE(Power S…...
骨传导耳机游泳能戴吗?骨传导游泳耳机哪个牌子好?
溽热的夏日,如果能够跳入水中畅游一番,那真的是再好不过了,既能强身健体,又能降温解暑。公共的游泳场馆人声鼎沸,像我这种“社恐”患者,如果在场馆中要待好几个小时,难免会觉得时间漫长…...
18万字应急管理局智慧矿山煤矿数字化矿山技术解决方案WORD
导读:原文《18万字应急管理局智慧矿山煤矿数字化矿山技术解决方案WORD》(获取来源见文尾),本文精选其中精华及架构部分,逻辑清晰、内容完整,为快速形成售前方案提供参考。 目 录 第一章 项目概述 1.1项目…...
【mysql】MySQL CUP过高如何排查?
文章目录 一. 问题锁定二. QPS激增会导致CPU飘高三. 慢SQL会导致CPU飘高四. 大量空闲连接会导致CPU飘高五. MySQL问题排查常用命令 一. 问题锁定 通过top命令查看服务器CPU资源使用情况,明确CPU占用率较高的是否是mysqld进程,如果是则可以明确CUP飘高的原…...
19c补丁后oracle属主变化,导致不能识别磁盘组
补丁后服务器重启,数据库再次无法启动 ORA01017: invalid username/password; logon denied Oracle 19c 在打上 19.23 或以上补丁版本后,存在与用户组权限相关的问题。具体表现为,Oracle 实例的运行用户(oracle)和集…...
java 实现excel文件转pdf | 无水印 | 无限制
文章目录 目录 文章目录 前言 1.项目远程仓库配置 2.pom文件引入相关依赖 3.代码破解 二、Excel转PDF 1.代码实现 2.Aspose.License.xml 授权文件 总结 前言 java处理excel转pdf一直没找到什么好用的免费jar包工具,自己手写的难度,恐怕高级程序员花费一年的事件,也…...
oracle与MySQL数据库之间数据同步的技术要点
Oracle与MySQL数据库之间的数据同步是一个涉及多个技术要点的复杂任务。由于Oracle和MySQL的架构差异,它们的数据同步要求既要保持数据的准确性和一致性,又要处理好性能问题。以下是一些主要的技术要点: 数据结构差异 数据类型差异ÿ…...
Fabric V2.5 通用溯源系统——增加图片上传与下载功能
fabric-trace项目在发布一年后,部署量已突破1000次,为支持更多场景,现新增支持图片信息上链,本文对图片上传、下载功能代码进行梳理,包含智能合约、后端、前端部分。 一、智能合约修改 为了增加图片信息上链溯源,需要对底层数据结构进行修改,在此对智能合约中的农产品数…...
安宝特案例丨Vuzix AR智能眼镜集成专业软件,助力卢森堡医院药房转型,赢得辉瑞创新奖
在Vuzix M400 AR智能眼镜的助力下,卢森堡罗伯特舒曼医院(the Robert Schuman Hospitals, HRS)凭借在无菌制剂生产流程中引入增强现实技术(AR)创新项目,荣获了2024年6月7日由卢森堡医院药剂师协会࿰…...
SQL慢可能是触发了ring buffer
简介 最近在进行 postgresql 性能排查的时候,发现 PG 在某一个时间并行执行的 SQL 变得特别慢。最后通过监控监观察到并行发起得时间 buffers_alloc 就急速上升,且低水位伴随在整个慢 SQL,一直是 buferIO 的等待事件,此时也没有其他会话的争抢。SQL 虽然不是高效 SQL ,但…...
沙箱虚拟化技术虚拟机容器之间的关系详解
问题 沙箱、虚拟化、容器三者分开一一介绍的话我知道他们各自都是什么东西,但是如果把三者放在一起,它们之间到底什么关系?又有什么联系呢?我不是很明白!!! 就比如说: 沙箱&#…...
Python 高级应用10:在python 大型项目中 FastAPI 和 Django 的相互配合
无论是python,或者java 的大型项目中,都会涉及到 自身平台微服务之间的相互调用,以及和第三发平台的 接口对接,那在python 中是怎么实现的呢? 在 Python Web 开发中,FastAPI 和 Django 是两个重要但定位不…...
路由基础-路由表
本篇将会向读者介绍路由的基本概念。 前言 在一个典型的数据通信网络中,往往存在多个不同的IP网段,数据在不同的IP网段之间交互是需要借助三层设备的,这些设备具备路由能力,能够实现数据的跨网段转发。 路由是数据通信网络中最基…...
英国云服务器上安装宝塔面板(BT Panel)
在英国云服务器上安装宝塔面板(BT Panel) 是完全可行的,尤其适合需要远程管理Linux服务器、快速部署网站、数据库、FTP、SSL证书等服务的用户。宝塔面板以其可视化操作界面和强大的功能广受国内用户欢迎,虽然官方主要面向中国大陆…...
