通过预分区文件避免 Spark 中的随机播放 (PySpark)答案

【问题标题】：Avoiding a shuffle in Spark by pre-partitioning files (PySpark)通过预分区文件避免 Spark 中的随机播放 (PySpark)
【发布时间】：2018-03-08 20:09:11
【问题描述】：

我有一个数据集 dataset，它按值 00-99 进行分区，并希望创建一个 RDD first_rdd 来读取数据。

然后我想计算单词“foo”在每个分区的第二个元素中出现的次数，并将每个分区的记录存储在一个列表中。我的输出是final_rdd，其中每条记录的格式为(partition_key, (count, record_list))。

def to_list(a):
    return [a]

def append(a, b):
    a.append(b)
    return a

def extend(a, b):
    a.extend(b)
    return a

first_rdd = sqlContext.sql("select * from dataset").rdd
kv_rdd = first_rdd.map(lambda x: (x[4], x)) # x[4] is the partition value
# Group each partition to (partition_key, [list_of_records])
grouped_rdd = kv_rdd.combineByKey(to_list, append, extend)

def count_foo(x):
    count = 0
    for record in x:
        if record[1] == "foo":
            count = count + 1
    return (count, x)

final_rdd = grouped_rdd.mapValues(count_foo)
print("Counted 'foo' for %s partitions" % (final_rdd.count))

由于数据集的每个分区在计算上完全相互独立，Spark 不需要洗牌，但是当我查看 SparkUI 时，我注意到 combineByKey 导致了非常大的洗牌。

我有正确数量的初始分区，并且还尝试从 HDFS 中的分区数据中读取。每次我尝试它，我仍然得到一个洗牌。我做错了什么？

【问题讨论】：

标签： apache-spark pyspark rdd pyspark-sql

【解决方案1】：

我已经通过使用 mapPartitions 函数解决了我的问题，并将我自己的 reduce 函数传递给它，以便它在每个节点上本地“减少”并且永远不会执行随机播放。

在各个分区之间数据隔离的场景下，完美运行。当多个分区上存在同一个 key 时，这就是需要进行 shuffle 的地方，但这种情况需要单独检测和处理。

【讨论】：