PySpark 本地开发环境搭建与实践

ops/2024/11/2 19:18:08/

目录

一、PySpark 本地开发环境搭建

（一）Windows 本地 JDK 和 Hadoop 的安装

conda-toc" style="margin-left:80px;">（二）Windows 安装 Anaconda

conda%20%E4%B8%AD%E5%AE%89%E8%A3%85%20PySpark-toc" style="margin-left:80px;">（三）Anaconda 中安装 PySpark

（四）Pycharm 中创建工程

二、编写代码

（一）编写环境变量的代码

（二）获取 SparkContext 对象

（三）将代码模板化

完整的模板：记得给模板起个名字pyspark_local_script

三、本地开发案例

（一）WordCount 案例

（二）使用正则解决特殊分隔符问题

（三）本地开发 - 读取 hdfs 上的数据

（四）本地开发 - 获取外部的变量

传递数据

参数设置界面

四、Spark 程序的监控

4040 界面的使用

五、local 和结果文件的数量问题

（一）local 模式并行度

（二）结果文件数量与 local 模式的关系

六、总结

        在大数据处理领域，PySpark 作为一个强大的工具，为数据科学家和开发人员提供了便捷的方式来处理大规模数据。本文将详细介绍如何在 Windows 环境下搭建 PySpark 本地开发环境，并深入探讨在这个环境下的代码编写、案例实践、程序监控以及一些常见问题的处理，帮助读者快速上手 PySpark 本地开发。

一、PySpark 本地开发环境搭建

（一）Windows 本地 JDK 和 Hadoop 的安装

        JDK（Java Development Kit）是 Java 开发的基础，而 Hadoop 是处理大数据的重要框架。在 Windows 上安装它们是后续搭建 PySpark 环境的第一步。安装过程需要注意选择合适的版本，并按照安装向导进行操作，确保安装路径等设置正确。

JDK安装配置教程_jdk64位安装-CSDN博客

Windows 系统安装 Hadoop 详细教程-CSDN博客

conda">（二）Windows 安装 Anaconda

         Anaconda 是一个开源的 Python 发行版本，它包含了 conda、Python 等 180 多个科学包及其依赖项。右键以管理员身份运行安装程序，默认安装到了 ProgramData 文件夹（这是一个非空文件夹）。在安装过程中，要注意各种安装选项，确保安装顺利进行。Anaconda 的安装为后续在其环境中安装 PySpark 等相关包提供了基础。

通过网盘分享Miniconda3的：Miniconda3-py38_4.11.0-Windows-x86_64.exe

conda%20%E4%B8%AD%E5%AE%89%E8%A3%85%20PySpark">（三）Anaconda 中安装 PySpark

         在命令提示符（cmd）中进行操作。在安装过程中，如果遇到需要输入 y 或者 n 的情况，输入 y。安装完成后，可以通过 conda list 或者 pip list 检查是否包含 py4j 和 pyspark 两个包。PySpark 的安装路径在 $ANACONDA_HOME/Lib/site - packages。这里需要强调的是，这实际上是在本地安装一个 Spark 软件，如果没有 Spark 环境，仅仅安装了 PySpark 是无法运行 Spark 代码的。

（四）Pycharm 中创建工程

  1. 选择 Conda：在 Pycharm 中创建工程时，选择 Conda。直接点确定即可。因为 Anaconda 包含了 Python 并且可以安装各种环境，比如 pyspark，通过这种关联，Pycharm 可以使用 Anaconda 中的工具。
  2. 解决识别问题：如果 Anaconda 没有安装在 C 盘，可能会出现识别不了的情况。此时需要手动选择。
  3. 检查安装包中是否有相关软件，并验证选择的解释器是否正确。
  4. 创建文件夹，为后续代码编写做好准备。

main ：用于存放每天开发的一些代码文件
resources ：用于存放程序中需要用到的配置文件
datas ：用于存放每天用到的一些数据文件
test ：用于存放测试时的一些代码文件

二、编写代码

（一）编写环境变量的代码

        环境变量的设置对于 PySpark 程序的运行至关重要。它确保程序能够找到所需的资源和配置。在代码中，要正确地设置与 Spark 相关的环境变量，包括 Spark 的安装路径、配置文件路径等。

import osif __name__ == '__main__':# 你自己的JDK路径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'

（二）获取 SparkContext 对象

         SparkContext 是 Spark 中的核心类，任何一个 Spark 的程序都必须包含一个 SparkContext 类的对象。通过获取这个对象，我们可以进一步构建和执行 Spark 任务。例如：

import os
# 导入pyspark模块
from pyspark import SparkContext,SparkConfif __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 对象conf = SparkConf().setMaster("local[*]").setAppName("第一个Spark程序")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字sc = SparkContext(conf=conf)print(sc)# 使用完后，记得关闭sc.stop()

（三）将代码模板化

         创建一个名为 pyspark_local_script 的模板，并在其中添加必要的内容。模板化代码有助于提高代码的复用性和规范性。在模板中，可以将一些常用的代码结构和函数定义好，方便在不同的项目中使用。

完整的模板：记得给模板起个名字pyspark_local_script
import os
# 导入pyspark模块
from pyspark import SparkContext,SparkConf"""
------------------------------------------Description : TODO：SourceFile : ${NAME}Author  : ${USER}Date  : ${DATE}
-------------------------------------------
"""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 对象conf = SparkConf().setMaster("local[*]").setAppName("第一个Spark程序")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字sc = SparkContext(conf=conf)print(sc)# 使用完后，记得关闭sc.stop()

模板的使用：

三、本地开发案例

（一）WordCount 案例

代码编写：这是一个经典的大数据处理案例。通过读取文本文件，将其中的单词进行拆分、计数。代码实现如下：

import os
# 导入pyspark模块
from pyspark import SparkContext, SparkConfif __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/java/jdk1.8.0_181'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/Linux/hadoop/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)print(sc)fileRdd = sc.textFile("../../datas/WordCount/data.txt")rsRdd = fileRdd.filter(lambda x: len(x) > 0) \.flatMap(lambda line: line.strip().split()) \.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a+b)rsRdd.saveAsTextFile("../datas/WordCount/result2")# 使用完后，记得关闭sc.stop()# 注意，在cmd窗口中执行spark-submit命令时，需要将以下路径添加到环境变量中

建议安装一个工具psutil，如果不想看到就在 Python解释器中安装：pip install psutil

查看运行结果：运行代码后，可以在指定的输出路径中查看结果文件。结果文件中包含了每个单词及其出现的次数。

常见的其他错误：

（二）使用正则解决特殊分隔符问题

        在实际数据处理中，可能会遇到特殊的分隔符。这时可以使用正则表达式来改造 WordCount 代码。例如，如果数据是用特定的非空格字符分隔的，可以通过修改 flatMap 函数中的分隔逻辑，使用正则表达式来正确拆分单词。

import os
import re# 导入pyspark模块
from pyspark import SparkContext, SparkConfif __name__ == '__main__':
# 配置环境
os.environ['JAVA_HOME'] = 'C:/Program Files/java/jdk1.8.0_181'
# 配置Hadoop的路径，就是前面解压的那个路径
os.environ['HADOOP_HOME'] = 'D:/Linux/hadoop/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)
print(sc)fileRdd = sc.textFile("../../datas/WordCount/data.txt")
rsRdd = fileRdd.filter(lambda x: len(x) > 0) \
.flatMap(lambda line: re.split("\\s+", line.strip())) \
.map(lambda word: (word, 1)) \
.reduceByKey(lambda sum, num: sum+num)rsRdd.saveAsTextFile("../datas/WordCount/result3")# 使用完后，记得关闭
sc.stop()

（三）本地开发 - 读取 hdfs 上的数据

        在 Windows 环境下，用户通常没有权限访问 hdfs 文件系统。这需要进行一些额外的配置，比如配置 Hadoop 的相关权限，或者通过一些代理工具来实现访问。在代码中，要正确设置 Hadoop 的配置参数，以确保能够读取 hdfs 上的数据。

	fileRdd = sc.textFile("hdfs://bigdata01:9820/spark/wordcount/input/*")rsRdd = fileRdd.filter(lambda line: len(line.strip()) > 0 ).flatMap(lambda line: re.split("\\s+",line.strip())).map(lambda word: (word,1)).reduceByKey(lambda sum,num : sum+num)rsRdd.saveAsTextFile("hdfs://bigdata01:9820/spark/wordcount/output4")

以上这个说明，windows用户没有权限访问hdfs文件系统

# 申明当前以root用户的身份来执行操作
os.environ['HADOOP_USER_NAME'] = 'root'

完整代码

import os
import re# 导入pyspark模块
from pyspark import SparkContext, SparkConfif __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/java/jdk1.8.0_181'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/Linux/hadoop/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'# 申明当前以root用户的身份来执行操作os.environ['HADOOP_USER_NAME'] = 'root'conf = SparkConf().setMaster("local[*]").setAppName("")sc = SparkContext(conf=conf)fileRdd = sc.textFile("hdfs://bigdata01:9820/spark/wordcount/input/*")rsRdd = fileRdd.filter(lambda line: len(line.strip()) > 0) \.flatMap(lambda line: re.split("\\s+", line.strip())) \.map(lambda word: (word, 1))\.reduceByKey(lambda sum, num: sum + num)rsRdd.saveAsTextFile("hdfs://bigdata01:9820/spark/wordcount/output4")# 使用完后，记得关闭sc.stop()

运行

（四）本地开发 - 获取外部的变量

        类似于 Java 中的 String[] args，在 PySpark 中也可以获取外部变量。可以通过命令行参数传递的方式来实现。例如，在运行 pyspark 脚本时，可以使用 spark - sumit xxxxx.py 参数 1, 参数 2 的形式传递参数。在代码中，需要对这些参数进行解析和使用。

import os
import re
import sys# 导入pyspark模块
from pyspark import SparkContext, SparkConfif __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/java/jdk1.8.0_181'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/Linux/hadoop/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'# 申明当前以root用户的身份来执行操作os.environ['HADOOP_USER_NAME'] = 'root'# 获取 conf 对象conf = SparkConf().setMaster("local[*]").setAppName("")# 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字sc = SparkContext(conf=conf)print(sc)# 获取第一个参数fileRdd = sc.textFile(sys.argv[1])rsRdd = fileRdd.filter(lambda line: len(line.strip()) > 0) \.flatMap(lambda line: re.split("\\s+", line.strip())) \.map(lambda word: (word, 1)) \.reduceByKey(lambda sum, num: sum + num)# 获取第二个参数rsRdd.saveAsTextFile(sys.argv[2])# 使用完后，记得关闭sc.stop()
传递数据

参数一： hdfs://bigdata01:9820/spark/wordcount/input/*
参数二： hdfs://bigdata01:9820/spark/wordcount/output4

参数设置界面

四、Spark 程序的监控

4040 界面的使用

        因为是本地的程序，所以可以通过访问地址 http://localhost:4040 来监控程序。每个 Spark 程序都有一个对应的 4040 界面。这个界面提供了丰富的信息：

  1. RDD 相关信息：每个黑点表示一个 RDD，每个矩形框中的 RDD 的转换都是在内存中完成的，曲线代表经过了 Shuffle，灰色代表没有执行（因为之前执行过）。
  2. 进程信息：显示当前这个程序的运行进程的信息。每个 Spark 程序都由两种进程组成：一个 Driver 和多个 Executors。Driver 进程负责解析程序，构建 DAG 图，构建 Stage，构建、调度、监控 Task 任务的运行；Executor 进程负责运行程序中的所有 Task 任务。
  3. 存储信息：Storage 部分显示当前这个程序在内存缓存的数据信息。
  4. 配置信息：Environment 显示当前这个程序所有的配置信息。

五、local 和结果文件的数量问题

（一）local 模式并行度

  1. local：使用本地模式，并行度是 1。
  2. local[3]：使用本地模式，并行度是 3，这个并行度最好和 CPU 的核数一致，一般并行度 <= CPU 的核数。
  3. local[*]：并行度 = CPU 的核数。

（二）结果文件数量与 local 模式的关系

        文件的结果经常是 2 个文件，这跟分区数有关系，跟 local = N 也有一定的关系。其规律是 min(N,2)，例如如果是 local [1]，最后的文件数量就是 1。

        如果在 local 模式下，想要结果文件是 10，可以使用 sc.textFile("../datas/wordcount/data.txt",10) 的方式来设置分区数。

六、总结

        本文详细介绍了 PySpark 本地开发环境的搭建过程，包括 JDK、Hadoop、Anaconda、PySpark 的安装以及 Pycharm 工程的创建。同时，深入讲解了代码编写、本地开发案例（如 WordCount、处理特殊分隔符、读取 hdfs 数据、获取外部变量）、Spark 程序的监控和 local 模式下结果文件数量问题等内容。通过掌握这些知识和技能，读者可以在 Windows 本地环境中高效地进行 PySpark 开发，处理大规模数据，解决实际业务中的数据分析和处理问题。希望本文能为读者在 PySpark 学习和实践的道路上提供有力的帮助，让读者能够更好地利用这个强大的工具来挖掘数据的价值。


http://www.ppmy.cn/ops/130500.html

相关文章

交易所开发：开启数字金融新时代

交易所开发：开启数字金融新时代

当今数字化高速发展的时代&#xff0c;交易所作为金融市场的核心枢纽&#xff0c;发挥着至关重要的作用。而随着区块链技术的兴起&#xff0c;数字货币交易所的开发更是为金融领域带来了全新的变革与机遇。 一、数字货币交易所的重要性 数字货币交易所是连接数字货币世界与传统…
阅读更多...
微服务基础-Ribbon

微服务基础-Ribbon

客户端负载均衡器 1.1 Ribbon介绍： 1.2 Ribbon--默认负载均衡策略： @loadBalanced就是调用loadBalance服务：获取注册过的服务列表：
阅读更多...
GPT-Sovits-1-数据处理

GPT-Sovits-1-数据处理

1.1 切割音频 将音频切割为多个10s内的片段 1.2 降噪 这一步用的是modelscope的pipeline 如果要去除背景音&#xff0c;可以用傅立叶转为为频谱&#xff0c;去除低频部分后再转回来 1.3 提取音频特征 这里用到了 funasr 库 这一步目的是输出音频样本的《文本标签文件》&am…
阅读更多...
SQL-lab靶场less1-4

SQL-lab靶场less1-4

说明&#xff1a;部分内容来源于网络&#xff0c;如有侵权联系删除 前情提要&#xff1a;搭建sql-lab本地靶场的时候发现一些致命的报错&#xff1a; 这个程序只能在php 5.x上运行&#xff0c;在php 7及更高版本上&#xff0c;函数“mysql_query”和一些相关函数被删除&#xf…
阅读更多...
Python基于TensorFlow实现双向循环神经网络GRU加注意力机制分类模型(BiGRU-Attention分类算法)项目实战

Python基于TensorFlow实现双向循环神经网络GRU加注意力机制分类模型(BiGRU-Attention分类算法)项目实战

说明&#xff1a;这是一个机器学习实战项目&#xff08;附带数据代码文档视频讲解&#xff09;&#xff0c;如需数据代码文档视频讲解可以直接到文章最后关注获取。 1.项目背景 随着深度学习技术的发展&#xff0c;循环神经网络&#xff08;RNN&#xff09;及其变种如门控循环…
阅读更多...
人工智能基础 opencv

人工智能基础 opencv

OpenCV&#xff08;Open Source Computer Vision Library&#xff09;是一个开源的计算机视觉和机器学习软件库。它提供了丰富的图像处理和计算机视觉算法&#xff0c;广泛应用于各种领域&#xff0c;如图像处理、视频分析、对象检测、人脸识别、机器学习等。 环境安装&#xf…
阅读更多...
Rust精简核心笔记：第二波，语法精髓部分解锁

Rust精简核心笔记：第二波，语法精髓部分解锁

前面介绍了Rust精简比较第一波 Rust精简核心笔记&#xff1a;第一波&#xff0c;深入浅出语法精华-CSDN博客 把第二波整理如下&#xff0c;通过三波会把全部Rust核心、实用、最简练的语法整理出来&#xff0c;最高效掌握Rust。 Rust精简笔记(二) Rust核心笔记第二波总结整理&am…
阅读更多...
前端八股文第八篇

前端八股文第八篇

71. 请求参数如何防篡改 在前端开发中&#xff0c;通常需要发送请求到后端获取数据或执行操作。为了防止请求参数被篡改&#xff0c;可以采取以下措施&#xff1a; 使用 HTTPS 协议&#xff1a;使用 HTTPS 可以确保数据在传输过程中的加密和完整性验证&#xff0c;防止请求被中…
阅读更多...
最新文章

Copyright @ 2022~2023