spark的共享变量
因为RDD在spark中是分布式存储
1、python中定义的变量仅仅在driver中运行,在excutor中是获取不到值的——广播变量
2、若定义了一个变量进行累加,先分别在driver和excutor中进行累加,但是结果是不会主动返回给driver的——累加器
Broadcast Variables广播变量
-
driver中存放python变量广播到别的excutor中
-
若不使用,就会每个task存放一个
-
不能修改,只能读
-
通过value使用该变量
if __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'D:/Program Files/Java/jdk1.8.0_271'# 配置Hadoop的路径,就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1/hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 获取 conf 对象# setMaster 按照什么模式运行,local bigdata01:7077 yarn# local[2] 使用2核CPU * 你本地资源有多少核就用多少核# appName 任务的名字conf = SparkConf().setMaster("local[*]").setAppName("第一个Spark程序")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件,得到一个SC对象,第一个conf 是 形参的名字,第二个conf 是实参的名字sc = SparkContext(conf=conf)fileRdd = sc.textFile("../datas/user.tsv",2)city_dict = {1: "北京",2: "上海",3: "广州",4: "深圳",5: "苏州",6: "无锡",7: "重庆",8: "厦门",9: "大理",10: "成都"}# 将一个变量广播出去,广播到executor中,不是task中city_dict_broad = sc.broadcast(city_dict)# 广播变量# <class 'pyspark.broadcast.Broadcast'>print(type(city_dict_broad ))# <class 'dict'>print(type(city_dict_broad.value))def getLine(line):list01 = line.split(" ")#cityName = city_dict.get(int(list01[3]))# 使用广播变量的变量获取数据cityName = city_dict_broad.value.get(int(list01[3]))# print(cityName)return line + " " + cityNamemapRdd = fileRdd.map(getLine)mapRdd.foreach(print)# 释放广播变量city_dict_broad.unpersist()# 使用完后,记得关闭sc.stop()
累加器
将所有的excutor中的变量返回到driver中,进行汇总。
否则变量是放在excutor中的,而打印的是driver中,变量值不会改变。
用于修改——汇总
import os
import reimport jieba
# 导入pyspark模块
from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel"""
------------------------------------------Description : TODO:SourceFile : _06SouGou案例Author : yangeDate : 2024/10/31 星期四
-------------------------------------------
"""
if __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_241'# 配置Hadoop的路径,就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 获取 conf 对象# setMaster 按照什么模式运行,local bigdata01:7077 yarn# local[2] 使用2核CPU * 你本地资源有多少核就用多少核# appName 任务的名字conf = SparkConf().setMaster("local[*]").setAppName("搜索热词案例")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件,得到一个SC对象,第一个conf 是 形参的名字,第二个conf 是实参的名字sc = SparkContext(conf=conf)mapRdd = sc.textFile("../../datas/zuoye/sogou.tsv",minPartitions=8) \.filter(lambda line:len(re.split("\s+",line)) == 6) \.map(lambda line:(re.split("\s+",line)[0],re.split("\s+",line)[1],re.split("\s+",line)[2][1:-1])).persist(StorageLevel.MEMORY_AND_DISK_2)# 统计一天每小时点击量并按照点击量降序排序_sum = 0def sumTotalLine(tuple1):global _sum # 把_sum 设置为全局变量timeStr = tuple1[0] # 10:19:18if timeStr[0:2] == '10':_sum += 1mapRdd.foreach(lambda tuple1:sumTotalLine(tuple1))print(_sum) # 结果是0# 使用完后,记得关闭sc.stop()
上面程序最终结果是:0,因为 sum=0 是在 Driver 端的内存中的,executor 中程序再累加也是无法改变 Driver 端的结果的。下面的则为正确的
import os
import reimport jieba
# 导入pyspark模块
from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel"""
------------------------------------------Description : TODO:SourceFile : _06SouGou案例Author : yangeDate : 2024/10/31 星期四
-------------------------------------------
"""
if __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_241'# 配置Hadoop的路径,就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 获取 conf 对象# setMaster 按照什么模式运行,local bigdata01:7077 yarn# local[2] 使用2核CPU * 你本地资源有多少核就用多少核# appName 任务的名字conf = SparkConf().setMaster("local[*]").setAppName("搜索热词案例")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件,得到一个SC对象,第一个conf 是 形参的名字,第二个conf 是实参的名字sc = SparkContext(conf=conf)accCounter = sc.accumulator(0)mapRdd = sc.textFile("../../datas/zuoye/sogou.tsv",minPartitions=8) \.filter(lambda line:len(re.split("\s+",line)) == 6) \.map(lambda line:(re.split("\s+",line)[0],re.split("\s+",line)[1],re.split("\s+",line)[2][1:-1])).persist(StorageLevel.MEMORY_AND_DISK_2)# 统计一天每小时点击量并按照点击量降序排序#_sum = 0def sumTotalLine(tuple1):#global _sum # 把_sum 设置为全局变量timeStr = tuple1[0] # 10:19:18if timeStr[0:2] == '10':accCounter.add(1)mapRdd.foreach(lambda tuple1:sumTotalLine(tuple1))print(accCounter.value) # 104694# 假如我不知道累加器这个操作,这个题目怎么做?print(mapRdd.filter(lambda tuple1: tuple1[0][0:2] == '10').count())# 使用完后,记得关闭sc.stop()
相关文章:
spark的共享变量
因为RDD在spark中是分布式存储 1、python中定义的变量仅仅在driver中运行,在excutor中是获取不到值的——广播变量 2、若定义了一个变量进行累加,先分别在driver和excutor中进行累加,但是结果是不会主动返回给driver的——累加器 Broadcas…...

Scrapy与MongoDB
Scrapy可以在非常短的时间里获取大量的数据。这些数据无论是直接保存为纯文本文件还是CSV文件,都是不可取的。爬取一个小时就可以让这些文件大到无法打开。这个时候,就需要使用数据库来保存数据了。 MongoDB由于其出色的性能,已经成为爬虫的首…...
爬虫基础与实践
爬虫技术基础与实践 在当今数字化的时代,数据成为了宝贵的资源。爬虫技术作为获取数据的重要手段,受到了广泛的关注和应用。本文将介绍爬虫的基本概念、工作原理以及一些常用的技术和工具。 一、爬虫的基本概念 爬虫,也称为网络蜘蛛或网络机器…...
快速上手Serverless架构与FastAPI结合实现自动化移动应用后端
快速上手Serverless架构与FastAPI结合实现自动化移动应用后端 引言 随着云计算技术的发展,Serverless架构已经成为构建现代应用的一种流行选择。它允许开发者将更多精力集中在核心业务逻辑上,而无需管理底层基础设施。本文将以AWS Lambda和API Gateway…...

ansible自动化运维(二)playbook模式详解
一.Ansible中的playbook模式 Playbook不同于使用单个模块操作远程服务器,Playbook的功能更加强大。如果说单个模块执行类似于Linux系统中的命令,那么Playbook就类似于shell脚本,将多个模块组合起来实现一组的操作。 Playbook还是会用到ad-h…...

基于Springboot社团管理系统【附源码】
基于Springboot社团管理系统 效果如下: 系统登录页面 用户管理页面 社团信息管理页面 社团活动管理页面 经费信息管理页面 新闻信息管理页面 系统主页面 社团信息页面 研究背景 在当今高校与社区环境中,学生社团蓬勃发展,成为学生课余生活…...
CSS:html中,.png的动态图,怎么只让它显示部分,比如只显示右上部分的,或右边中间部分
目录 背景 方法 1: 使用 background-image 和 background-position 示例代码 解释 方法 2: 使用 clip-path 裁剪图像 示例代码 解释 方法 3: 使用 object-fit 和 overflow 示例代码 解释 示例 总结 背景 在HTML中,如果你有一个 .png 的动态图(例如一个 GIF 动画或…...

解读CVPR2024-论文分享|RepViT: Revisiting Mobile CNN From ViT Perspective
论文标题 RepViT: Revisiting Mobile CNN From ViT Perspective 论文链接: https://arxiv.org/abs/2307.09283 论文作者 Ao Wang, Hui Chen, Zijia Lin, Jungong Han, Guiguang Ding 内容简介 这篇论文探讨了在资源受限的移动设备上,轻量级视觉变…...

linux部署安装wordpress
一、环境准备 首先我们先介绍下环境和实验中所需要的包 环境: 我使用的是centos7.6的系统 建议关掉selinux和影响到80端口的防火墙策略 selinux永久有效 修改 /etc/selinux/config 文件中的 SELINUX"" 为 disabled ,然后重启。 selinux即…...
[Java] 配置Powershell 的 Maven 环境变量
目录 前言单独为 Powershell 设置 Maven 环境变量 前言 安装使用 maven 的时候发现,明明已经配置好了环境变量。但是在 powershell 中还是无法识别 mvn 命令。原来这货需要另外配置。 单独为 Powershell 设置 Maven 环境变量 要在 PowerShell 中永久配置 Maven 环…...

Android -- [SelfView] 自定义弹窗式颜色选择器
Android – [SelfView] 自定义弹窗式颜色选择器 PS: 1. 弹框式显示; 2. 支持透明度设置; 3. 支持拖动控件选择颜色; 4. 支持 ARGB | HEX 数值填写预览颜色并返回; 5. 输出支持Hex 和 Int 两种格式;效果 使用方法&…...

vue-echarts高度缩小时autoresize失效
背景 项目中采用动态给x-vue-echarts style赋值width,height的方式实现echarts图表尺寸的改变 <v-chart...autoresize></v-chart>给v-chart添加autoresize后,在图表宽度变化,高度增加时无异常,高度减小时图表并未缩…...

rabbitMq的rabbitmqctl status报错
Error: unable to perform an operation on node rabbitASUS-PC. Please see diagnostics information and suggestions below. 遇到上图这个错大部分问题可能是由于 RabbitMQ CLI 工具的 Erlang Cookie 与服务器上的不匹配而导致连接问题。Erlang Cookie 在 RabbitMQ 节点之间…...
linux c++ uuid编译时的问题
linux c uuid编译时的问题 写在前面可能编译过和不能编译过的可以编译和link过的不能编译过的 写在前面 几次翻车与uuid相关,超出我认知。 所以,把一些遇到的相关问题写在这里。 可能编译过和不能编译过的 可以编译和link过的 cmake_minimum_require…...

【STM32】RTT-Studio中HAL库开发教程九:FLASH中的OPT
文章目录 一、概要二、内部FLASH排布三、内部FLASH主要特色四、OTP函数介绍五、测试验证 一、概要 STM32系列是一款强大而灵活的微控制器,它的片内Flash存储器可以用来存储有关代码和数据,在实际应用中,我们也需要对这个存储器进行读写操作。…...

[SWPUCTF 2021 新生赛]crypto9
[MoeCTF 2021]Web安全入门指北—GET 意思是GET传参,moeflag 就可以得到falg 输入?moeflag flag为: NSSCTF{ff26110b-8793-403c-990e-15c7f1820596} [SWPUCTF 2021 新生赛]crypto9 #gpt写的代码 from itertools import product letter_list ABCDEFG…...
vue中常用的指令
v - if 指令 功能详细解释 它是一种真正的条件渲染指令。在 Vue 实例初始化以及数据更新过程中,Vue.js 会对v - if指令中的表达式进行求值。这个表达式可以是简单的布尔变量,也可以是一个复杂的计算表达式,只要最终结果是布尔值就行。当表达式…...

Docker Compose实战三:轻松部署PHP
通过前面的文章(Docker Compose基础语法与MySQL部署),你已经掌握了Docker Compose的基本语法和常用指令,并成功部署了一个MySQL数据库服务器。今天,我们将继续深入探索Docker Compose的强大功能,介绍如何使…...

数据分析实战—房价特征关系
1.实战内容 (1) 读取房价特征关系表(house_price.npz)绘制离地铁站的距离与单位面积的房价的散点图,并对其进行分析; import pandas as pd import numpy as np import warnings warnings.filterwarnings(&…...

云和恩墨 zCloud 与华为云 GaussDB 完成兼容性互认证
近日,云和恩墨(北京)信息技术有限公司(以下简称:云和恩墨)的多元数据库智能管理平台 zCloud 与华为云计算技术有限公司(以下简称:华为云)的 GaussDB 数据库完成了兼容性互…...

深入浅出Asp.Net Core MVC应用开发系列-AspNetCore中的日志记录
ASP.NET Core 是一个跨平台的开源框架,用于在 Windows、macOS 或 Linux 上生成基于云的新式 Web 应用。 ASP.NET Core 中的日志记录 .NET 通过 ILogger API 支持高性能结构化日志记录,以帮助监视应用程序行为和诊断问题。 可以通过配置不同的记录提供程…...
云计算——弹性云计算器(ECS)
弹性云服务器:ECS 概述 云计算重构了ICT系统,云计算平台厂商推出使得厂家能够主要关注应用管理而非平台管理的云平台,包含如下主要概念。 ECS(Elastic Cloud Server):即弹性云服务器,是云计算…...
Java 语言特性(面试系列1)
一、面向对象编程 1. 封装(Encapsulation) 定义:将数据(属性)和操作数据的方法绑定在一起,通过访问控制符(private、protected、public)隐藏内部实现细节。示例: public …...

盘古信息PCB行业解决方案:以全域场景重构,激活智造新未来
一、破局:PCB行业的时代之问 在数字经济蓬勃发展的浪潮中,PCB(印制电路板)作为 “电子产品之母”,其重要性愈发凸显。随着 5G、人工智能等新兴技术的加速渗透,PCB行业面临着前所未有的挑战与机遇。产品迭代…...
MySQL 隔离级别:脏读、幻读及不可重复读的原理与示例
一、MySQL 隔离级别 MySQL 提供了四种隔离级别,用于控制事务之间的并发访问以及数据的可见性,不同隔离级别对脏读、幻读、不可重复读这几种并发数据问题有着不同的处理方式,具体如下: 隔离级别脏读不可重复读幻读性能特点及锁机制读未提交(READ UNCOMMITTED)允许出现允许…...
ssc377d修改flash分区大小
1、flash的分区默认分配16M、 / # df -h Filesystem Size Used Available Use% Mounted on /dev/root 1.9M 1.9M 0 100% / /dev/mtdblock4 3.0M...

理解 MCP 工作流:使用 Ollama 和 LangChain 构建本地 MCP 客户端
🌟 什么是 MCP? 模型控制协议 (MCP) 是一种创新的协议,旨在无缝连接 AI 模型与应用程序。 MCP 是一个开源协议,它标准化了我们的 LLM 应用程序连接所需工具和数据源并与之协作的方式。 可以把它想象成你的 AI 模型 和想要使用它…...
Auto-Coder使用GPT-4o完成:在用TabPFN这个模型构建一个预测未来3天涨跌的分类任务
通过akshare库,获取股票数据,并生成TabPFN这个模型 可以识别、处理的格式,写一个完整的预处理示例,并构建一个预测未来 3 天股价涨跌的分类任务 用TabPFN这个模型构建一个预测未来 3 天股价涨跌的分类任务,进行预测并输…...

Java-41 深入浅出 Spring - 声明式事务的支持 事务配置 XML模式 XML+注解模式
点一下关注吧!!!非常感谢!!持续更新!!! 🚀 AI篇持续更新中!(长期更新) 目前2025年06月05日更新到: AI炼丹日志-28 - Aud…...

《基于Apache Flink的流处理》笔记
思维导图 1-3 章 4-7章 8-11 章 参考资料 源码: https://github.com/streaming-with-flink 博客 https://flink.apache.org/bloghttps://www.ververica.com/blog 聚会及会议 https://flink-forward.orghttps://www.meetup.com/topics/apache-flink https://n…...