PySpark 如何重新分区 PySpark DataFrame

在本文中，我们将介绍如何使用 PySpark 对 DataFrame 进行重新分区。PySpark 是一个强大的分布式计算框架，可以处理大规模的数据集。

什么是重新分区？

重新分区是指改变数据集的分区方式，将数据重新划分到不同的分区中。在分布式计算中，数据的分区方式可以影响到计算任务的性能和效率。

在 PySpark 中，DataFrame 是一个逻辑上的数据表，它可以通过分区来划分数据。每个分区都是一个独立的数据块，可以在集群中的不同节点上并行计算。

如何重新分区 DataFrame？

在 PySpark 中，可以使用 repartition 方法来重新分区 DataFrame。repartition 方法可以接收一个整数参数，指定新的分区数。

例如，假设我们有一个包含 1000 行数据的 DataFrame，并且想要将它重新分区为 4 个分区：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 假设 df 是一个包含 1000 行数据的 DataFrame
df = spark.read.csv("data.csv", header=True)

# 将 DataFrame 重新分区为 4 个分区
df_repartitioned = df.repartition(4)

在上面的例子中，我们通过 repartition(4) 方法将 DataFrame df 重新分区为 4 个分区。

当然，也可以根据特定的列来重新分区 DataFrame。可以使用 repartition 方法的另一个重载形式，传递一个或多个列作为参数。例如，假设我们有一个包含城市和销售额数据的 DataFrame，并且想要根据城市列重新分区：

# 假设 df 是一个包含城市和销售额数据的 DataFrame
df = spark.read.csv("sales.csv", header=True)

# 根据城市列重新分区
df_repartitioned = df.repartition("city")

上面的代码将 DataFrame df 根据城市列重新分区。

如何控制分区的数量？

通过 repartition 方法默认会分配一个可用的默认分区器，它将不均匀地分配数据到不同的分区中。如果希望更精确地控制分区的数量和分配方式，可以使用 repartitionByRange 或 repartitionByHash 方法。

使用 repartitionByRange 方法

repartitionByRange 方法可以将数据按照指定的列和范围进行重新分区。

例如，假设我们有一个包含销售数据的 DataFrame，其中包含城市和销售额两列。我们希望根据销售额范围将数据重新分区为 4 个分区：

# 假设 df 是一个包含城市和销售额数据的 DataFrame
df = spark.read.csv("sales.csv", header=True)

# 根据销售额范围重新分区为 4 个分区
df_repartitioned = df.repartitionByRange(4, "sales")

上面的代码将 DataFrame df 根据销售额范围重新分区为 4 个分区。

使用 repartitionByHash 方法

repartitionByHash 方法可以将数据按照指定的列进行重新分区。

例如，假设我们有一个包含用户数据的 DataFrame，其中包含用户 ID 和用户名两列。我们希望根据用户 ID 将数据重新分区为 4 个分区：

# 假设 df 是一个包含用户 ID 和用户名数据的 DataFrame
df = spark.read.csv("users.csv", header=True)

# 根据用户 ID 重新分区为 4 个分区
df_repartitioned = df.repartitionByHash(4, "user_id")