PySpark 教程：从零开始上手 PySpark

什么是 PySpark？

PySpark 是 Apache Spark 的 Python 接口。使用 PySpark，您可以在分布式处理环境中用 Python 和类 SQL 的命令来处理和分析数据。数据科学家使用 PySpark 清洗处理数据、构建机器学习流水线并调优模型。

大多数数据科学家与分析师熟悉 Python，并用它来实现机器学习流程。PySpark 让他们可以在大规模分布式数据集上继续使用熟悉的语言。Apache Spark 也可与 R 等其他数据科学语言配合使用。如果您对此感兴趣，可以从 Introduction to Spark with sparklyr in R 课程入门。

PySpark 中的 RDD 与 DataFrame

PySpark 提供两种分布式数据表示方式。新手常遇到的第一个问题就是该用哪一种。

对于绝大多数工作——包括本教程中的全部内容——DataFrame 都是正确选择。Spark 的 Catalyst 查询优化器会自动将 DataFrame 操作改写为高效的执行计划，带来您用 RDD 需要手工实现的性能收益。仅当您需要直接控制分区，或处理无法适配表格架构的数据时，再使用 RDD。

为何使用 PySpark？

PySpark 是处理大数据的首选，因为它将 Python 的易用性与 Spark 的分布式计算能力结合在一起。与替代方案比较如下：

企业选择 PySpark 之类框架的原因在于其处理大数据的速度极快。它比 Pandas 与 Dask 更快，且能处理更大的数据量。例如，若需要处理 PB 级数据，Pandas 和 Dask 会力不从心，但 PySpark 则能轻松应对。

虽然也可以在像 Hadoop 这样的分布式系统之上编写 Python 代码，但许多组织选择改用 Spark，并采用 PySpark API，因为它更快且可处理实时数据。使用 PySpark，您可以编写代码从持续更新的数据源采集数据；而在 Hadoop 中数据只能以批处理模式处理。 

Apache Flink 也是分布式处理系统，拥有名为 PyFlink 的 Python API，性能上甚至比 Spark 更快。然而，Apache Spark 问世更早、社区支持更好，因此整体可靠性更高。 

PySpark 还提供容错性：如果某个节点在任务中途失败，Spark 会利用 RDD 的血统信息重建丢失数据。该框架支持内存中计算，并存储在随机存取存储器（RAM）中。它甚至可以运行在未安装硬盘或 SSD 的机器上。

如何安装 PySpark

以下是在本地或云环境中安装 PySpark 的方法。 

先决条件

在开始安装之前，请确保已安装以下组件：

• Java 11 或更高版本（推荐 Java 17）
• Python 3.7 或更高版本
• Jupyter Notebook 

注意：如果您使用 DataLab 或 Databricks 等云平台，可跳过本地安装，因为 PySpark 已预装。

安装 PySpark

在 Jupyter Notebook 中打开一个 Python 文件，在第一个单元格运行以下代码：

!pip install pyspark

或者，您也可以参考这份端到端的 PySpark 安装指南，在您的设备上完成安装。

端到端机器学习 PySpark 教程

现在 PySpark 已就绪，我们将向您展示如何用该库完成一个端到端的客户细分项目。 

客户细分是企业用于识别并归类表现出相似特征用户的一种营销技术。比如，如果您只在夏季去星巴克购买冷饮，您就可被划分为“季节性消费者”，并通过专为夏季定制的促销活动来吸引。

数据科学家通常会构建无监督学习算法，如 K-Means 聚类或层次聚类，来执行客户细分。这些模型擅长识别用户群体间相似的模式，而这些规律往往难以被人眼直接察觉。

本教程将使用 K-Means 聚类，在电商数据集上进行客户细分。

完成本教程后，您将熟悉以下概念：

• 使用 PySpark 读取 CSV 文件

• 使用 PySpark 进行探索性数据分析

• 分组与排序

• 执行算术运算

• 聚合数据集

• 使用 PySpark 进行数据预处理

• 处理日期时间值

• 类型转换

• 连接两个数据帧

• rank() 函数

• PySpark 机器学习

• 创建特征向量

• 数据标准化

• 构建 K-Means 聚类模型

• 模型解读

步骤 1：创建 SparkSession

SparkSession 是使用 Spark 的入口，如果您要在 PySpark 中构建 DataFrame，就必须创建它。运行下面的代码来初始化 SparkSession： 

from pyspark.sql import SparkSession  # add this import


spark = (
    SparkSession.builder
    .appName("DataCamp PySpark Tutorial")
    .config("spark.memory.offHeap.enabled", "true")
    .config("spark.memory.offHeap.size", "10g")
    .getOrCreate()
)

上述代码创建了一个 Spark 会话并为应用程序设置了名称。随后，我们将数据缓存到堆外内存，避免直接写入磁盘，并手动指定了内存大小。

步骤 2：创建 DataFrame

现在可以读取数据集了。您可以从我们的 PySpark 读取 CSV 教程下载示例电商数据集，或使用您自己的 CSV 文件：

df = spark.read.csv("datacamp_ecommerce.csv", header=True, escape='"', inferSchema=True)

请注意，我们定义了转义字符，以避免在解析 .csv 文件时受到逗号的影响。

让我们用 show() 函数查看 DataFrame 的前几行：

df.show(5,0)

该 DataFrame 包含 8 个变量：

1. InvoiceNo：每张客户发票的唯一标识符。

2. StockCode：每个库存商品的唯一标识符。

3. Description：客户购买的商品。

4. Quantity：客户在单张发票中购买的每种商品数量。

5. InvoiceDate：购买日期。

6. UnitPrice：每件商品的单价。

7. CustomerID：分配给每位用户的唯一标识符。

8. Country：完成购买所在的国家。

步骤 3：探索性数据分析

既然我们已经了解了该数据集中的变量，接下来进行一些探索性分析，以进一步理解这些数据点：

1. 首先统计 DataFrame 的行数：

df.count()  # Answer: 2,500

1. DataFrame 中有多少唯一客户？

df.select('CustomerID').distinct().count() # Answer: 95

1. 大多数购买来自哪个国家？

要找出购买最多的国家，我们需要在 PySpark 中使用 groupBy() 子句：

from pyspark.sql.functions import *
from pyspark.sql.types import *

df.groupBy('Country').agg(countDistinct('CustomerID').alias('country_count')).show()

运行上述代码后，将呈现如下表格：

平台上的几乎所有购买都来自英国，只有少量来自德国、澳大利亚和法国等国家。 

请注意，上表中的数据并未按购买数量排序。要对该表排序，我们可以加入 orderBy() 子句：

df.groupBy('Country').agg(countDistinct('CustomerID').alias('country_count')).orderBy(desc('country_count')).show()

现在输出已按降序排序：

1. 电商平台上，客户最近的一笔购买是什么时候？

要找到最新购买时间，我们需要将 InvoiceDate 列转换为时间戳格式，并使用 PySpark 的 max() 函数：

df = df.withColumn(
    "date",
    coalesce(
        to_timestamp(col("InvoiceDate"), "yy/MM/dd HH:mm"),
        to_timestamp(col("InvoiceDate"), "yyyy-MM-dd HH:mm:ss"),
        to_timestamp(col("InvoiceDate"))  # best-effort fallback
    )
)
df.select(max("date")).show()

运行上述代码后，您应看到如下表格：

1. 电商平台上，客户最早的一笔购买是什么时候？

与上面类似，可以使用 min() 函数找到最早的购买日期时间：

df.select(min("date")).show()

可以看到，最近与最早的购买都发生在同一天，相隔仅数小时。这意味着我们下载的数据集只包含单日的购买信息。

步骤 4：数据预处理

在分析完数据集并更好地理解各数据点后，我们需要准备数据，以便输入到机器学习算法。

让我们再次查看数据帧的前几行，以便明确预处理的方式：

df.show(5,0)

基于上述数据集，我们需要根据每位用户的购买行为创建多个客户分群。 

该数据集中的变量尚不易直接被客户细分模型使用。单个特征并不能很好地揭示客户购买行为。

因此，我们将基于现有变量派生出三个新的信息性特征——新近性（Recency）、频率（Frequency）和金额（Monetary，简称 RFM）。

RFM 通常用于营销，以评估客户价值，基于他们的：

1. 新近性：每位客户最近一次购买发生在何时？
2. 频率：他们购买的频率如何？
3. 金额：他们每次购买平均花费多少？

接下来我们将对数据帧进行预处理，以创建上述变量。

新近性（Recency）

首先计算新近性——平台上最近一次购买的日期和时间。可分两步完成：

i) 为每位客户分配新近性得分

我们将数据帧中的每个日期减去最早日期。这将告诉我们客户在数据帧中被“看见”的最近程度。值为 0 表示新近性最低，因为它将分配给在最早日期被观察到有购买行为的人。

df = df.withColumn("from_date", to_timestamp(lit("12/1/10 08:26"), "yy/MM/dd HH:mm"))
df2 = df.withColumn("recency", col("date").cast("long") - col("from_date").cast("long"))


w = Window.partitionBy("CustomerID").orderBy(desc("recency"))
df2 = df2.withColumn("rn", row_number().over(w)).filter(col("rn") == 1).drop("rn")

ii) 选择最近的一次购买

同一位客户可能在不同时间多次购买。我们需要只选择他们最后一次被观察到购买商品的时间，因为这能指示最近一次购买发生的时间： 

df2 = df2.join(df2.groupBy('CustomerID').agg(max('recency').alias('recency')),on='recency',how='leftsemi')

查看新数据帧的前几行。可以看到它新增了名为“recency”的变量：

df2.show(5,0)

在 PySpark DataFrame 中更便捷地查看所有变量的方法是使用 printSchema() 函数。这相当于 Pandas 中的 info()：

df2.printSchema()

输出应类似如下：

频率（Frequency）

现在计算频率——客户在平台上购买的频繁程度。为此，只需按 CustomerID 分组并统计他们购买的商品数量。更多高级分组技巧，请参阅我们的 PySpark groupBy 教程：

df_freq = df2.groupBy('CustomerID').agg(count('InvoiceDate').alias('frequency'))

查看我们刚创建的新 DataFrame 的前几行：

df_freq.show(5,0)

每位客户都对应了一个频率值。该新 DataFrame 只有两列，我们需要将其与先前的数据进行连接。想了解更多连接类型，请参阅我们的 PySpark 连接教程：

df3 = df2.join(df_freq,on='CustomerID',how='inner')

打印该 DataFrame 的 schema：

df3.printSchema()

金额（Monetary Value）

最后计算金额——每位客户在 DataFrame 中的总消费额。分两步完成：

i) 计算每笔购买的总金额：

每个 CustomerID 的单笔购买都包含 Quantity 与 UnitPrice 两个变量：

要获得每位客户在单笔购买中的总花费，只需将 Quantity 与 UnitPrice 相乘：

m_val = df3.withColumn(
    "TotalAmount",
    col("Quantity").cast("double") * col("UnitPrice").cast("double")
)

ii) 计算每位客户的总消费：

要找出每位客户的总体花费，只需按 CustomerID 分组，并对总金额求和：

m_val = m_val.groupBy('CustomerID').agg(sum('TotalAmount').alias('monetary_value'))

将该 DataFrame 与其他变量合并：

finaldf = m_val.join(df3,on='CustomerID',how='inner')

现在我们已经创建了构建模型所需的全部变量，运行以下代码仅选择所需列，并去重：

finaldf = finaldf.select(['recency','frequency','monetary_value','CustomerID']).distinct()

查看最终 DataFrame 的前几行，确认预处理已准确完成：

标准化

在构建客户细分模型之前，让我们对 DataFrame 进行标准化，确保所有变量的量纲处于相近水平：

from pyspark.ml.feature import VectorAssembler, StandardScaler


assemble = VectorAssembler(
    inputCols=["recency", "frequency", "monetary_value"],
    outputCol="features"
)
assembled_data = assemble.transform(finaldf)


scale = StandardScaler(inputCol="features", outputCol="standardized")
data_scale = scale.fit(assembled_data)
data_scale_output = data_scale.transform(assembled_data)

运行以下代码查看标准化后的特征向量：

data_scale_output.select('standardized').show(2,truncate=False)

这些即是将要输入聚类算法的缩放特征。

如果您想进一步学习 PySpark 的数据准备，可在 DataCamp 上学习这门特征工程课程。

步骤 5：构建机器学习模型

完成数据分析与准备后，让我们开始构建 K-Means 聚类模型。 

该算法将使用 PySpark 的 机器学习 API 创建。

i) 确定聚类数量

在构建 K-Means 聚类模型前，我们首先需要确定算法返回的簇（或群组）数量。例如，如果选择 3 个簇，就会得到 3 个客户分群。

最常用的确定 K-Means 聚类数目的方法是“肘部法”。

具体做法是对一系列不同的聚类数运行 K-Means 算法，并将每种聚类数对应的模型结果可视化。图中会出现一个类似“手肘”的拐点，我们就选择该拐点对应的聚类数。

阅读这篇 DataCamp K-Means 聚类教程，了解算法的工作原理。

让我们运行以下代码，从 2 到 10 个簇构建 K-Means 聚类算法：

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
import numpy as np

cost = np.zeros(10)

evaluator = ClusteringEvaluator(
    predictionCol="prediction",
    featuresCol="standardized",
    metricName="silhouette",
    distanceMeasure="squaredEuclidean"
)


ks = range(2, 10)
cost = np.zeros(len(ks))


for idx, k in enumerate(ks):
    km = KMeans(featuresCol="standardized", k=k)
    model = km.fit(data_scale_output)
    output = model.transform(data_scale_output)
    cost[idx] = model.summary.trainingCost   # WSSSE

以上代码成功构建并评估了 2 到 10 个簇的 K-Means 模型。结果已存入数组，可用折线图可视化：

import pandas as pd
import pylab as pl
df_cost = pd.DataFrame(cost)  # cost has 8 values, one per k in range(2, 10)
df_cost.columns = ["cost"]
new_col = range(2, 10)
df_cost.insert(0, 'cluster', new_col)
pl.plot(df_cost.cluster, df_cost.cost)
pl.xlabel('Number of Clusters')
pl.ylabel('Score')
pl.title('Elbow Curve')
pl.show()

上述代码会生成如下图表：

ii) 构建 K-Means 聚类模型

从上图可见，拐点大约出现在 4，因此我们以 4 个簇来构建 K-Means 算法：

KMeans_algo=KMeans(featuresCol='standardized', k=4)
KMeans_fit=KMeans_algo.fit(data_scale_output)

iii) 生成预测

使用我们创建的模型，为数据集中的每位客户分配簇标签：

preds=KMeans_fit.transform(data_scale_output)

preds.show(5,0)

注意该 DataFrame 中新增了“prediction”列，指明每个 CustomerID 属于哪个簇：

步骤 6：簇分析

本教程的最后一步是分析我们刚构建的客户分群。

运行以下代码，查看 DataFrame 中每个 CustomerID 的新近性、频率与金额：

import matplotlib.pyplot as plt
import seaborn as sns

df_viz = preds.select('recency','frequency','monetary_value','prediction')
df_viz = df_viz.toPandas()
avg_df = df_viz.groupby(['prediction'], as_index=False).mean()

rfm_columns = ['recency', 'frequency', 'monetary_value']

for metric in rfm_columns:
    sns.barplot(x='prediction', y=metric, data=avg_df)
    plt.show()

以上代码将生成如下图表：

各簇客户特征概览如下：

• 簇 0：该分群的客户在新近性、频率与金额上都较低。他们很少在平台购物，潜力较低，可能会停止与电商平台进行业务往来。
• 簇 1：该簇用户的新近性较高，但在平台上的花费不多，访问频率也不高。这表明他们可能是刚开始与公司进行业务往来的新客户。
• 簇 2：该分群客户的新近性与频率居中，但在平台上消费较高。这表明他们倾向于购买高价值商品或进行批量采购。
• 簇 3：最后一组用户的新近性高、购买频繁，但在平台上的花费不多，这可能意味着他们每次购买倾向于选择价格较低的商品。

若想深入超越本文涉及的预测建模概念，您可以在 DataCamp 学习 Machine Learning with PySpark 课程。

从零开始学习 PySpark：下一步

现在您已完成本教程，基于您的目标，推荐的下一步如下：

如果您完整跟进了本次 PySpark 教程，恭喜您！您已在本地设备上成功安装 PySpark，分析了一个电商数据集，并使用该框架构建了机器学习算法。

需要说明的一点是，上述分析仅基于单日收集的 2,500 行电商数据。如果我们拥有更多数据，分析结论会更稳健，因为像 RFM 建模这类技术通常应用于数月的历史数据。

不过，您可以将本文所学的原理应用到无监督学习领域中各种更大规模的数据集上。

查看这份 DataCamp 的速查表，了解更多 PySpark 的语法与模块。

最后，如果您希望超越本教程所涵盖的概念，系统学习 PySpark 编程基础，欢迎在 DataCamp 参加 Big Data with PySpark 学习路径。该路径包含一系列课程，将教您使用 PySpark 完成以下工作：

• 数据管理、分析与预处理
• 构建与调优机器学习流水线
• 大数据分析 
• 特征工程 
• 构建推荐引擎

结语

当您的数据超出单机所能处理的范围时，PySpark 就是合适的工具。本教程中的 RFM 客户细分项目演示了完整流程：加载数据、探索性分析、特征工程与机器学习。这些模式都可在生产环境中更大规模的数据集上复用。

但也要坦诚一点：这个示例使用的是单日 2,500 行交易记录。PySpark 处理它当然轻松。分布式架构的真正价值，体现在处理跨数月、数百万事件的交易历史时——那时内存中执行与容错机制才真正发挥作用。

如需继续进阶，我们的 Big Data with PySpark 学习路径以结构化序列覆盖数据工程流水线、推荐引擎与生产级机器学习。面向团队，DataCamp for Business 提供针对数据工程岗位定制的学习路径。申请演示了解更多。