rdd例子mapPartitionsWithIndex

mapPartitionsWithIndex()是一种在每个分区上应用的转换操作，它接受两个参数：分区索引和分区中的元素迭代器。它与mapPartitions()类似，但还提供了分区索引作为输入。

下面是一个示例，展示了如何使用mapPartitionsWithIndex()在RDD的每个分区上添加分区索引：

# 导入pyspark模块
from pyspark import SparkContext

# 创建SparkContext对象
sc = SparkContext("local", "mapPartitionsWithIndex example")

# 创建一个RDD
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
rdd = sc.parallelize(data, 3)

# 定义一个函数，在每个分区上添加分区索引
def add_partition_index(index, iterator):
    for x in iterator:
        yield (index, x)

# 使用mapPartitionsWithIndex()转换操作
result = rdd.mapPartitionsWithIndex(add_partition_index)

# 打印结果
for x in result.collect():
    print(x)

# 停止SparkContext对象
sc.stop()

输出结果如下：

(0, 1)
(0, 2)
(1, 3)
(1, 4)
(2, 5)
(2, 6)
(0, 7)
(1, 8)
(2, 9)
(2, 10)

在这个示例中，我们首先创建了一个包含10个元素的RDD，并将其分为3个分区。然后，我们定义了一个函数add_partition_index()，它接受分区索引和分区中的元素迭代器，并在每个元素前添加分区索引。最后，我们使用mapPartitionsWithIndex()转换操作将这个函数应用到RDD的每个分区上，得到了包含分区索引的新RDD。最后，我们使用collect()方法将结果收集到驱动程序并打印出来。

注意：在实际使用中，分区索引可能并不总是按照顺序递增的整数。这取决于RDD的分区方式和数据分布