PySpark 读取 CSV：高效加载与处理大文件

Q: PySpark 能处理不同分隔符的 CSV 吗？

可以。您可以在 spark.read.csv() 中通过 delimiter 选项设置自定义分隔符。例如，对竖线分隔的值使用 delimiter="|"。

Q: 如何一次读取目录中的多个 CSV 文件？

您可以使用通配路径，比如 spark.read.csv("/data/*.csv", ...) ，一次性将多个文件加载到同一个 DataFrame 中。

CSV 文件是数据处理与分析的基石。几乎每位数据从业者——从中级数据工程师、数据科学家到使用 PySpark 的开发者——都会在某个阶段与 CSV 打交道。然而，当数据集从 MB 级增长到 GB、甚至 TB 及以上时，像 Pandas 或标准 Python 库这样的简单单机工具开始难以承载。这正是 Apache Spark 和 PySpark 在分布式计算环境中管理大规模 CSV 文件时发挥作用的地方。

本文将教您在使用 PySpark 读取大型 CSV 文件时需要了解的一切。

请先确保已安装 PySpark，并通过我们的 PySpark 入门教程熟悉基础知识。

在 PySpark 中读取 CSV 是什么？

PySpark 允许用户将 CSV 文件读取为分布式 DataFrame。PySpark 的 DataFrame 与 Pandas DataFrame 十分类似，提供熟悉的接口。但在底层，PySpark DataFrame 会将计算与存储分布到多个节点上，为海量数据集带来卓越性能。

对于大型 CSV 文件，PySpark 具有以下显著优势：

分布式数据加载，
稳健的空值处理，
灵活的模式定义，
以及管理多文件或压缩 CSV 的简便方法。

尽管如此，仍会遇到一些常见难题，包括高效处理表头、准确推断或指定模式、以及管理损坏或不一致的记录。继续阅读，了解如何应对这些挑战。

在 PySpark 中读取 CSV 是您拿下面试题必备的话题之一。我们的 2025 年最全 36 道 PySpark 面试问答从基础概念到高级技术与优化策略，提供了全面指南。

在 PySpark 中读取 CSV 的基础

在 PySpark 工作流中，读取 CSV 往往是最初且最关键的步骤之一，它为后续的转换、探索性分析和机器学习任务打下基础。把这一步做好，能带来更清洁的数据处理与更高的下游性能。

概念框架

PySpark 在 Spark 的分布式模型下读取 CSV 文件。Spark 并非在单机内存中一次性读取所有 CSV 数据，而是将大规模数据任务拆分并分配到集群中的多个节点。Spark 内置的 Catalyst 优化器还能通过高效执行 CSV 摄取过程中所需的底层操作来进一步提升性能。

核心读取语法

读取 CSV 的简单方式是使用 Spark 内置函数：

spark.read.csv("file_path", header=True, inferSchema=True)技术支持

或显式写法：

spark.read.format("csv").option("header", "True").load("file_path")技术支持

关键参数包括：

file_path - CSV 文件的位置。
header - 若为 True，则使用 CSV 表头设置列名。
inferSchema - 自动推断列的数据类型。
delimiter - 列分隔符；默认逗号。

我们的 2025 年从零学习 PySpark 教程更详尽地介绍了 PySpark 基础与学习路径。

读取 CSV：选项与配置

PySpark 提供了丰富选项，以便对 CSV 读取过程进行精细控制。

表头与模式推断

设置 header=True 可指示 Spark 使用第一行作为列名。

inferSchema=True 让 Spark 通过扫描数据自动猜测列类型：

spark.read.csv("customers.csv", header=True, inferSchema=True)技术支持

尽管模式推断在初期方便且有效，但对大型数据集会降低性能，因为 Spark 需要反复扫描数据来确定类型。

自定义模式规范

显式定义模式能显著提升性能，因为无需 Spark 多次扫描数据。定义好的模式可预先告知列名与类型。

以下是在 PySpark 中定义自定义模式的方法：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
schema = StructType([
    StructField("user_id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("score", DoubleType(), True),
])
df = spark.read.csv("customers.csv", schema=schema, header=True)技术支持

接下来，看看在读取 CSV 时 PySpark 如何处理分隔符。

处理分隔符与特殊字符

许多 CSV 使用逗号以外的分隔符，如竖线或制表符。PySpark 允许显式指定分隔符：

spark.read.csv("customers.csv", header=True, delimiter="|")技术支持

此外，还可以配置转义与引号字符，以处理特殊字符场景：

spark.read.csv("data.csv", header=True, escape='\"', quote='"')技术支持

管理空值与缺失值

现实世界的 CSV 数据常包含不一致或不完整的记录。PySpark 可通过将自定义占位符转换为空值，简化空值处理：

spark.read.csv("customers.csv", header=True, schema=schema, nullValue="NA")技术支持

这能明确空值语义，大幅减少后续手动清洗。

我们再看看处理空值的其他策略。

您可以选择过滤掉空值：

# Filter rows where Age is not null
df_filtered = df.filter(df["Age"].isNotNull())
df_filtered.show()技术支持

这会将 DataFrame 过滤为仅包含 Age 列非空的行。输出类似于：

+---+-----+---+------+
| ID| Name|Age|Salary|
+---+-----+---+------+
|  1| John| 25| 50000|
|  3|  Bob| 30|  NULL|
|  4|Carol| 28| 55000|
+---+-----+---+------+技术支持

另一种策略是填充空值：

# Replace null values in Age and Salary with default values
df_filled = df.na.fill({"Age": 0, "Salary": 0})
df_filled.show()技术支持

输出如下：

+---+-----+---+------+
| ID| Name|Age|Salary|
+---+-----+---+------+
|  1| John| 25| 50000|
|  2|Alice|  0| 60000|
|  3|  Bob| 30|     0|
|  4|Carol| 28| 55000|
|  5|David|  0| 48000|
+---+-----+---+------+技术支持

读取多个文件与目录

PySpark 擅长管理大型多文件数据集。无需手动顺序加载并合并目录中的文件，PySpark 支持通配模式以实现快速、高效的批量加载：

spark.read.csv("/data/sales/*.csv", header=True, schema=schema)技术支持

这能将大量 CSV 文件一次性聚合为一个 DataFrame，流程简洁高效。

数据加载到 PySpark 后，后续步骤包括数据整理、特征工程与构建机器学习模型。我们的 PySpark 特征工程课程将深入讲解这些概念。

高效读取 CSV 的优化技巧

在处理大规模 CSV 数据时，利用 PySpark 的优化策略至关重要。

分区策略

分区会通过将数据均匀分布到集群节点，对性能产生重要影响。Spark 允许在数据摄取时显式控制分区大小与数量，以加速后续操作：

df = spark.read.csv("data.csv", header=True, schema=schema).repartition(20)技术支持

.repartition(20) 会将 DataFrame 在 Spark 集群上拆分为 20 个分区。由于 Spark 以块为单位处理数据，更多分区可以：

提升并行度
平衡集群中的负载
加速转换与写出

如果您运行在拥有大量内核的集群上，这有助于充分利用资源。但若分区数设置过多（例如在小数据集上设 1000 个分区），反而可能拖慢性能。

通过我们的 PySpark 速查表：Python 中的 Spark 了解更多如 repartition 等 Spark 函数。内容涵盖在 Python 中初始化 Spark、加载数据、排序与重分区等细节。

缓存与持久化

若您的工作流在摄取后需要反复访问同一数据集，将 DataFrame 缓存在内存或磁盘上可显著提升性能：

df.cache()技术支持

但需注意缓存占用系统资源；务必在内存使用与性能收益之间做好权衡。

惰性求值与触发 action

PySpark 采用惰性求值模型：DataFrame 操作会变换执行计划，而非立刻执行。实际读取仅在必要时触发，例如 show()、count() 或 collect() 命令：

# no reading yet
df = spark.read.csv("data.csv", header=True, schema=schema)   
# actual read triggered here
df.show(5)  技术支持

高级用例与注意事项

下面探讨在读取 CSV 文件时可能遇到的更复杂场景：

读取压缩的 CSV 文件

Spark 可高效处理 gz 或 .bz2 等压缩 CSV 文件，无需额外配置即可透明读取：

spark.read.csv("logs.csv.gz", header=True, schema=schema)技术支持

处理格式错误的记录

CSV 数据集中可能包含格式错误的行。PySpark 提供多种选项，帮助您优雅地管理错误或格式不正确的记录：

mode="PERMISSIVE"（默认）：包含格式错误的行，并以空值填充列。
mode="DROPMALFORMED：静默跳过格式错误记录。该模式在 CSV 内置函数中不受支持。
mode="FAILFAST"：一旦遇到任何格式错误记录即抛出异常。

spark.read.csv("data.csv", header=True, schema=schema, mode="FAILFAST")技术支持

区域设置与编码

有时 CSV 数据会使用非标准编码。PySpark 可通过 encoding 参数轻松处理不同编码：

spark.read.csv("data_utf8.csv", header=True, encoding="UTF-8")技术支持

支持的编码包括 US-ASCII、ISO-8859-1、UTF-8、UTF-16BE、UTF-16LE 和 UTF-16。

最佳实践与常见陷阱

使用 Spark 读取 CSV 时，请牢记以下最佳实践：

在已知数据集结构时显式指定模式。
控制分区以高效分配工作负载。
对高频访问的 DataFrame 进行有策略的缓存。
有意识地触发读取操作，注意惰性求值。

务必避免：

对包含表头行的 CSV 设置 header=False。
在大型或被反复访问的数据集中仅依赖 inferSchema。
忽略关键的分隔符或编码设置。

结语

充分理解并利用 PySpark 强大的 CSV 摄取能力，是实现高效大数据处理的关键。通过清晰的模式定义、处理自定义空值格式、采用高效的分区策略，以及应对压缩文件或多文件场景，您的工作流将更顺畅且更高效。

请记住，尽管 PySpark 在大规模数据任务上优势明显，但对小型数据集，Pandas 等简单工具仍然足够。当数据规模超出单机能力时再使用 PySpark，并充分考虑其分布式计算优势。

想进一步探索 PySpark，请查看我们的深度 Spark 课程，包括：

PySpark 读取 CSV 常见问答

在 PySpark 中读取大型 CSV 的最佳方式是什么？

最佳实践是使用 StructType 指定自定义模式，而不是依赖 inferSchema。这样可避免重复数据扫描，从而提升性能。

PySpark 能处理不同分隔符的 CSV 吗？

可以。您可以在 spark.read.csv() 中通过 delimiter 选项设置自定义分隔符。例如，对竖线分隔的值使用 delimiter="|"。

如何一次读取目录中的多个 CSV 文件？

您可以使用通配路径，比如 spark.read.csv("/data/*.csv", ...)，一次性将多个文件加载到同一个 DataFrame 中。

读取 CSV 时 `mode="DROPMALFORMED"` 有什么作用？

它会告诉 Spark 跳过并忽略 CSV 文件中的格式错误记录，而不是将其包含或导致失败。

PySpark 能读取如 `.gz` 或 `.bz2` 的压缩 CSV 吗？

可以。只要提供正确的文件路径，PySpark 会自动解压并读取压缩的 CSV 文件，无需额外配置。

作者

Derrick Mwiti

主题

PySpark Python

主题

技术