PySpark的学习

作者：小蓝xlanll | 2024-04-01 14:44:14

踩

PySpark的学习

一. 什么是PySpark

使用过的bin/pyspark 程序 , 要注意 , 这个只是一个应用程序 , 提供一个 Python 解释器执行环境来运行 Spark 任务

现在说的 PySpark, 指的是 Python 的运行类库 , 是可以在 Python 代码中 :import pyspark PySpark 是 Spark 官方提供的一个 Python 类库 , 内置了完全的 Spark API, 可以通过 PySpark 类库来编写 Spark 应用程序 ,并将其提交到Spark 集群中运行 . 下图是,PySpark 类库和标准 Spark 框架的简单对比

Anaconda是Python语言的一个发行版. 内置了非常多的数据科学相关的Python类库, 同时可以提供虚拟环境来供不同的程序使用.

我们写spark的时候在windows上开发不可避免的会用到部分hadoop功能

为了避免在windows上报错, 我们给windows打补丁.

二.PyCharm配置Python解释器

1.配置本地解释器

如果没有找到conda虚拟环境的解释器,可以：

2. 配置远程 SSH Linux 解释器

刚刚,配置了本地的Python(基于conda虚拟环境)的解释器, 现在我们来配置Linux远程的解释器.

PySpark 支持在Windows 上执行,但是会有性能问题以及一些小bug , 在Linux上执行是完美和高效的.

所以, 我们也可以配置好Linux上的远程解释器, 来运行Python Spark代码

1) 设置远程SSH python pySpark 环境

2) 添加新的远程连接

3) 设置虚拟机Python环境路径

三.应用入口：SparkContext

Spark Application程序入口为： SparkContext ，任何一个应用首先需要构建SparkContext 对象，如下两步构建：

第一步、创建SparkConf对象

设置Spark Application基本信息，比如应用的名称AppName和应用运行Master

第二步、基于SparkConf对象，创建SparkContext对象

四.单词计数代码实践


from pyspark import SparkContext, SparkConf
import os
# 这里可以选择本地PySpark环境执行Spark代码，也可以使用虚拟机中PySpark环境，通过os可以配置
os.environ['SPARK_HOME'] = '/export/servers/spark'
# PYSPARK_PYTHON = "/root/anaconda3/envs/pyspark_env/bin/python"
# 当存在多个版本时，不指定很可能会导致出错
# os.environ["PYSPARK_PYTHON"] = PYSPARK_PYTHON
# os.environ["PYSPARK_DRIVER_PYTHON"] = PYSPARK_PYTHON
if __name__ == '__main__':
print('PySpark First Program')
# TODO: 当应用运行在集群上的时候，MAIN函数就是Driver Program，必须创建SparkContext对象
# 创建SparkConf对象，设置应用的配置信息，比如应用名称和应用运行模式
conf = SparkConf().setAppName("miniProject").setMaster("local[*]")
# TODO: 构建SparkContext上下文实例对象，读取数据和调度Job执行
sc = SparkContext(conf=conf)
# 第一步、读取本地数据 封装到RDD集合，认为列表List
wordsRDD = sc.textFile("file:///export/pyfolder1/pyspark-chapter01_3.8/data/word.txt")
# 第二步、处理数据 调用RDD中函数，认为调用列表中的函数
# a. 每行数据分割为单词
flatMapRDD = wordsRDD.flatMap(lambda line: line.split(" "))
# b. 转换为二元组，表示每个单词出现一次
mapRDD = flatMapRDD.map(lambda x: (x, 1))
# c. 按照Key分组聚合
resultRDD = mapRDD.reduceByKey(lambda a, b: a + b)
# 第三步、输出数据
res_rdd_col2 = resultRDD.collect()
# 输出到控制台
for line in res_rdd_col2:
print(line)
# 输出到本地文件中
resultRDD.saveAsTextFile("file:///export/pyfolder1/pyspark-chapter01_3.8/data/output1/")
print('停止 PySpark SparkSession 对象')
# 关闭SparkContext
sc.stop()