PySpark partitionBy() 方法

在 Spark 中，常常需要对数据进行分区，以便能够更好地进行数据处理。Spark 提供了 partitionBy() 方法，可以用来按指定的键进行分区操作。

语法

rdd.partitionBy(numPartitions, partitionFunc)

参数说明：

• numPartitions：指定分区的数量。
• partitionFunc：指定分区的方式。

示例

假设有以下一组数据：

data = [("apple", 12), ("orange", 8), ("apple", 4), ("orange", 5)]

使用 parallelize() 方法创建一个 RDD，然后执行 partitionBy() 方法，按 key 进行分区操作：

from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName("partitionBy").setMaster("local")
sc = SparkContext(conf=conf)

data = [("apple", 12), ("orange", 8), ("apple", 4), ("orange", 5)]
rdd = sc.parallelize(data)

partitioned = rdd.partitionBy(2, lambda x: 0 if x[0] == "apple" else 1)

print(partitioned.glom().collect())

输出结果如下：

[[], [('apple', 12), ('apple', 4)], [('orange', 8), ('orange', 5)], []]

表示 RDD 被分成了四个分区，其中第二个分区和第三个分区是按 key 进行分区的，分别包含了所有 apple 类型和 orange 类型的数据。

注意事项

• 分区操作会造成数据的重新分配和网络传输，因此要避免过度分区；
• partitionFunc 函数的返回值应该为一个整数，用来表示数据所属的分区编号；
• 如果调用 partitionBy() 方法之前已经进行了 groupByKey、reduceByKey 或者 aggregateByKey 等聚合操作，那么该操作已经隐含了分区操作，因此不需要再次调用 partitionBy()。

以上就是 PySpark 的 partitionBy() 方法的介绍。