PySpark – 将 CSV 文件读入 DataFrame

在 PySpark 中，DataFrame 是一种强大的数据结构，用于处理大规模数据集。CSV 是一种常见的数据格式，本文将介绍如何使用 PySpark 将 CSV 文件读入 DataFrame。

准备工作

在开始之前，你需要安装和配置好 PySpark 环境。可以通过 pip 安装 PySpark：

pip install pyspark

然后，你需要创建一个 SparkSession 对象，用于与 Spark 集群通信和执行操作：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("CSV to DataFrame") \
    .getOrCreate()

读取 CSV 文件

要读取 CSV 文件，可以使用 spark.read.csv() 方法。该方法接受以下参数：

• path：CSV 文件的路径。
• header：CSV 文件是否包含标题行，默认为 False。
• inferSchema：是否自动推断列的数据类型，默认为 False。

以下示例演示了如何将 CSV 文件读入 DataFrame：

df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

在这个例子中，我们假设我们的 CSV 文件具有标题行，且希望自动推断列的数据类型。

查看 DataFrame

一旦成功读取 CSV 文件，并将其转换为 DataFrame，你可以使用 show() 方法来查看 DataFrame 的内容：

df.show()

show() 方法默认显示前 20 行数据。如果你想显示更多行数，可以指定 n 参数：

df.show(n=50)

处理 DataFrame

读取 CSV 文件并转换为 DataFrame 后，你可以使用 DataFrame API 提供的方法来处理和操作数据。

以下示例展示了一些常用的 DataFrame 操作：

查看 DataFrame 的列名

df.columns

查看 DataFrame 的前几行

df.head()

查看 DataFrame 的统计摘要信息

df.describe().show()

选择特定的列

df.select("column_name")

过滤行

df.filter(df["column_name"] > 10)

以上只是一些常见的操作，你可以根据具体需求使用 DataFrame API 提供的更多操作方法。

结束会话

在完成对 DataFrame 的处理后，记得关闭 SparkSession 对象，释放资源：

spark.stop()

以上就是使用 PySpark 将 CSV 文件读入 DataFrame 的介绍。希望本文能帮助你快速上手使用 PySpark 处理 CSV 数据。