我有一个带有4个节点的体系结构和一个带有4000行的RDD,我需要在节点上同样地重新分配此RDD。结果是:
node 1 -> 1000 rows
node 2 -> 1000 rows
node 3 -> 1000 rows
node 4 -> 1000 rows.
我该如何在python中执行此操作?
我尝试实现 @sramalingam24已使用pyspark所提出的解决方案。
from collections import Counter
data = [(i,j) for i,j in zip([i/1000 for i in range(0, 4000, 1)], range(500, 4500, 1))]
rdd = sc.parallelize(data).map(lambda x : (x[0], x[1]))
df = sqlContext.createDataFrame(rdd, ['key', 'values'])
df = df.repartition('key')
检查结果:
Counter(df.select(spark_partition_id()).collect())
Out[*]: Counter({Row(SPARK_PARTITION_ID()=5): 1000, Row(SPARK_PARTITION_ID()=128): 1000, Row(SPARK_PARTITION_ID()=107): 1000, Row(SPARK_PARTITION_ID()=69): 1000})
如果您使用文本文件来构建RDD,则可以使用:
scala> val rdd = sc.textFile("hdfs://.../input.txt", 4)
或您可以使用:
scala> rdd = rdd.repartition(4)
rdd.repartition(n)做一个shuffle以拆分数据以匹配n个分区。