Spark 没有在减少时使用任何并行化答案

【问题标题】：Spark is not utlizing any parallization on reductionSpark 没有在减少时使用任何并行化
【发布时间】：2015-01-05 03:00:34
【问题描述】：

我是新手。我正在使用 python (pyspark) 编写我的程序。我使用groupByKey 函数将键值对转换为键（值列表）对。我在 64 核计算机上运行 spark，并尝试通过使用以下命令启动程序来利用所有 64 核。

spark-submit --master local[64] my_program.py

但是，我注意到在执行groupByKey 函数时，只使用了一个内核。数据相当大。那么，为什么 spark 不将其划分为 64 个分区并在 64 个不同的内核中进行归约/分组？

我是否遗漏了一些重要的并行化步骤？

代码的相关部分是这样的，

# Here input itself is a key-(list of values) pair. The mapPartitions
# function is used to return a key-value pair (variable x), from
# which another key-(list of values) pair is created (variable y)
x = input.mapPartitions(transFunc)
# x contains key value pair, such as [(k1, v1), (k1, v2), (k2, v3)]
y = x.groupByKey()
# y contains key-list of values pair such as [(k1, [v1, v2]), (k2, [v2])]

【问题讨论】：

你是如何加载数据的？
@maasg：我使用 mapPartitions。在 mapPartitions 之后，say 变量 x 中的结果数据是一个键值对，其中 key 是一个字符串，value 也是一个字符串。然后我使用 groupByKey 形成一个键到（值列表）对，其中键与 x 中的键相同，值列表是字符串值列表。
能否将代码添加到问题中？
@maasg：现在已经添加了代码。
仍然缺少加载数据的部分。

标签： python scala bigdata apache-spark

【解决方案1】：

Spark 中的默认并行级别由配置选项决定：spark.default.parallelism。默认值为：（*来自docs）

本地模式：本地机器上的核心数 Mesos 细粒度模式：8 其他：所有执行器节点上的核心总数或 2，以较大者为准

可以使用这些操作将 RDD 重新组合到更多或更少的分区中：

rdd.repartition(partitions: Int) // redistributes the RDD into the given nr of partitions
rdd.coalesce(partitions:Int) // reduces the number of partitions of the RDD to the given nr

需要内部 shuffle 的操作通常采用numPartitions 参数来指定目标分区的数量。在这样的操作之后，RDD 将拥有新的分区数。让我用一个例子来说明这一点：

给定：

val rdd = sc.textFile("localFile")  // default nr of partitions. Let's say 2

然后：

val moreParallelRdd = rdd.repartition(64) // 64 partitions
val onePartitionRdd = moreParallelRdd.coalesce(1) // 1 partition
val sortedRdd = onePartitionRdd.sortBy(x=> sortSelector(x), numPartitions=10) // 10 partitions

【讨论】：