PySpark 使用 PySpark 加载 CSV 文件

在本文中，我们将介绍如何使用 PySpark 加载和处理 CSV 文件。

CSV 文件是一种常见的结构化数据格式，它以逗号分隔不同的字段。PySpark 提供了强大的工具和函数，可以轻松地加载和处理 CSV 文件。

阅读更多：PySpark 教程

1. 安装 PySpark

在开始之前，我们需要先安装 PySpark。可以使用以下命令来安装 PySpark：

pip install pyspark

2. 导入必要的模块

安装完成后，我们需要导入一些必要的 PySpark 模块，包括 SparkSession 和 functions 。

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

3. 创建 SparkSession

在加载和处理 CSV 文件之前，我们需要创建一个 SparkSession 对象。SparkSession 对象是与 Spark 集群进行通信的入口点。

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

4. 加载 CSV 文件

PySpark 提供了一个 read 方法，可以用来加载 CSV 文件。我们可以使用 option 方法来指定一些加载参数，例如文件路径、分隔符和是否包含标题等。

下面是一个加载 CSV 文件的示例：

df = spark.read \
    .option("header", "true") \
    .option("delimiter", ",") \
    .csv("path/to/csv/file.csv")

在这个示例中，我们加载了一个名为 file.csv 的 CSV 文件，并且指定了分隔符为逗号，并且指定了文件的第一行为标题行。

5. 数据处理

一旦我们加载了 CSV 文件，我们就可以对数据进行各种处理操作。PySpark 提供了丰富的函数和方法来处理和转换数据。

例如，我们可以使用 select 方法选择特定的列：

df.select("column1", "column2")

我们也可以使用 filter 方法对数据进行筛选：

df.filter(F.col("column1") > 100)

另外，我们还可以使用 withColumn 方法添加新的列或修改现有的列：

df.withColumn("new_column", F.col("column1") * F.col("column2"))

6. 数据写出

在完成数据处理后，我们可以使用 PySpark 将数据写出到其他的文件格式。

例如，我们可以将数据写出为 Parquet 格式：

df.write.parquet("path/to/parquet/file.parquet")

总结

PySpark 提供了灵活而强大的工具和函数来加载和处理 CSV 文件。在本文中，我们介绍了如何使用 PySpark 加载 CSV 文件，并展示了一些数据处理和写出的示例。希望这些内容对你在使用 PySpark 进行数据处理和分析时有所帮助。