Apache Spark是一个为大数据处理而设计的分布式计算系统，它可以运行在Hadoop集群之上，也可以独立部署。而PySpark是Spark的Python API，提供了易用性和灵活性，是进行数据处理和分析的优秀选择。
一、环境搭建
在开始学习PySpark之前，需要安装Python以及Spark，这里提供两种安装方式。第一种是使用Anaconda，它是一个开源的Python发行版，可以安装Python以及众多常用库。第二种是手动安装Python和Spark。这里需要注意，PySpark需要跟Spark版本匹配。
下面是Anaconda环境下安装PySpark指南：
conda create -n pyspark python=3.7
conda activate pyspark
conda install pyspark手动安装Python和Spark，需要先下载好对应版本的Python和Spark。然后按照以下步骤执行：
# 安装Python
tar -zxvf Python-3.7.10.tgz
cd Python-3.7.10
./configure --prefix=/usr/local/python3.7
make && make install
# 安装Spark
tar -zxvf spark-3.1.2-bin-hadoop3.2.tgz
mv spark-3.1.2-bin-hadoop3.2 /usr/local/spark注意在安装完Spark之后，还需要配置环境变量：
export SPARK_HOME=/usr/local/spark
export PATH=$PATH:$SPARK_HOME/bin二、RDD基础
Resilient Distributed Dataset（RDD）是Spark的核心数据结构。它具有高度的容错性和可靠性，并且可以在内存中缓存数据，提高处理效率。
在PySpark中创建一个RDD，需要指定一个集合或者文件作为数据源。例如，下面的代码将一个文本文件中的行读入并创建一个RDD：
from pyspark import SparkContext
sc = SparkContext("local", "WordCount")
lines = sc.textFile("file:///usr/local/spark/README.md")上面代码指定了一个本地模式的SparkContext，并将文件/usr/local/spark/README.md中的行读入为RDD。在创建RDD后，可以使用如下的操作来操作它：
- map: 对RDD中每个元素执行一个函数，返回一个新的RDD。例如：
words = lines.flatMap(lambda line: line.split(" ")) - filter: 对RDD中每个元素执行一个函数，返回该函数返回值为True的元素组成的新RDD。例如：
filteredWords = words.filter(lambda word: len(word) > 5) - reduceByKey: 根据RDD中的Key对Value执行聚合操作。例如：
wordCount = filteredWords.map(lambda word: (word, 1)).reduceByKey(lambda x, y: x + y) - collect: 将RDD中的所有元素都收集到Driver端并返回一个列表。例如：
result = wordCount.collect()
三、DataFrame和SQL
除了RDD之外，在Spark中还有一种强大的数据结构：DataFrame。它是一种以列为基本操作对象的数据结构，提供了一系列的列转换和过滤操作。同时，Spark还提供了类SQL查询的API，可以通过SparkSession使用。
下面两段代码分别是如何创建和使用DataFrame和Spark SQL进行查询：
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("DataFrameExample").getOrCreate()
# 从CSV文件读入DataFrame
df = spark.read.csv("file:///usr/local/spark/examples/src/main/resources/people.csv", header=True, inferSchema=True)
# 显示前20行
df.show(20)
# 计算平均年龄
avgAge = df.select("age").groupBy().mean().collect()[0][0]
print("Average Age: {:.2f}".format(avgAge))# 注册为临时表
df.createOrReplaceTempView("people")
# SQL查询
result = spark.sql("SELECT name, age FROM people WHERE age > 30")
# 显示结果
result.show()四、机器学习
最后，我们来介绍一下PySpark中的机器学习库，它提供了多种常见的机器学习算法，包括分类、回归、聚类等。
下面是一个利用PySpark进行逻辑回归分类的例子：
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.sql import SparkSession
# 读入数据集
spark = SparkSession.builder.appName("LogisticRegressionExample").getOrCreate()
data = spark.read.format("csv").option("header", True).option("inferSchema", True).load("file:///usr/local/spark/examples/src/main/resources/student_scores.csv")
# 将特征转换为向量
assembler = VectorAssembler(inputCols=["math", "physics"], outputCol="features")
data = assembler.transform(data)
# 划分训练集和测试集
train, test = data.randomSplit([0.7, 0.3])
# 建立逻辑回归模型
lr = LogisticRegression()
# 设置调参参数
paramGrid = ParamGridBuilder().addGrid(lr.regParam, [0.01, 0.1]).build()
# 交叉验证
crossval = CrossValidator(estimator=lr,
estimatorParamMaps=paramGrid,
evaluator=BinaryClassificationEvaluator(),
numFolds=5)
model = crossval.fit(train)
# 预测
result = model.transform(test)
# 评估
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction")
print("AUC: {:.2f}".format(evaluator.evaluate(result)))五、总结
本篇文章主要介绍了PySpark的基本知识和常用操作。首先，我们介绍了环境搭建的两种方法，并给出了相关的代码。接着，我们讲述了RDD这一核心数据结构的相关操作，并给出了相应的代码。然后，我们介绍了DataFrame和Spark SQL的使用方法，并提供了相应的代码。最后，我们介绍了机器学习部分，包括建立模型、调参和评估等，并给出了相应的代码。掌握了这些知识后，读者可以利用PySpark进行大规模数据处理和机器学习。
原创文章，作者：YXDZL，如若转载，请注明出处：https://www.506064.com/n/335127.html
微信扫一扫 
支付宝扫一扫 