目录
一、PySpark 本地开发环境搭建
(一)Windows 本地 JDK 和 Hadoop 的安装
(二)Windows 安装 Anaconda
(三)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博客
(二)Windows 安装 Anaconda
Anaconda 是一个开源的 Python 发行版本,它包含了 conda、Python 等 180 多个科学包及其依赖项。右键以管理员身份运行安装程序,默认安装到了 ProgramData 文件夹(这是一个非空文件夹)。在安装过程中,要注意各种安装选项,确保安装顺利进行。Anaconda 的安装为后续在其环境中安装 PySpark 等相关包提供了基础。
通过网盘分享Miniconda3的:Miniconda3-py38_4.11.0-Windows-x86_64.exe
(三)Anaconda 中安装 PySpark
在命令提示符(cmd)中进行操作。在安装过程中,如果遇到需要输入 y 或者 n 的情况,输入 y。安装完成后,可以通过 conda list
或者 pip list
检查是否包含 py4j
和 pyspark
两个包。PySpark 的安装路径在 $ANACONDA_HOME/Lib/site - packages。这里需要强调的是,这实际上是在本地安装一个 Spark 软件,如果没有 Spark 环境,仅仅安装了 PySpark 是无法运行 Spark 代码的。
(四)Pycharm 中创建工程
- 选择 Conda:在 Pycharm 中创建工程时,选择 Conda。直接点确定即可。因为 Anaconda 包含了 Python 并且可以安装各种环境,比如 pyspark,通过这种关联,Pycharm 可以使用 Anaconda 中的工具。
- 解决识别问题:如果 Anaconda 没有安装在 C 盘,可能会出现识别不了的情况。此时需要手动选择。
- 检查安装包中是否有相关软件,并验证选择的解释器是否正确。
- 创建文件夹,为后续代码编写做好准备。
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 界面。这个界面提供了丰富的信息:
- RDD 相关信息:每个黑点表示一个 RDD,每个矩形框中的 RDD 的转换都是在内存中完成的,曲线代表经过了 Shuffle,灰色代表没有执行(因为之前执行过)。
- 进程信息:显示当前这个程序的运行进程的信息。每个 Spark 程序都由两种进程组成:一个 Driver 和多个 Executors。Driver 进程负责解析程序,构建 DAG 图,构建 Stage,构建、调度、监控 Task 任务的运行;Executor 进程负责运行程序中的所有 Task 任务。
- 存储信息:Storage 部分显示当前这个程序在内存缓存的数据信息。
- 配置信息:Environment 显示当前这个程序所有的配置信息。
五、local 和结果文件的数量问题
(一)local 模式并行度
- local:使用本地模式,并行度是 1。
- local[3]:使用本地模式,并行度是 3,这个并行度最好和 CPU 的核数一致,一般并行度 <= CPU 的核数。
- 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 学习和实践的道路上提供有力的帮助,让读者能够更好地利用这个强大的工具来挖掘数据的价值。