导致Shuffle的Spark转换有哪些？

Question

我在查找Spark文档时遇到了一些困难，想知道哪些操作会导致数据的“洗牌”（shuffle），哪些不会。在这个列表中，哪些操作会导致洗牌，哪些不会呢？

像map和filter这些操作是不会导致洗牌的。不过，对于其他的操作我就不太确定了。

map(func)
filter(func)
flatMap(func)
mapPartitions(func)
mapPartitionsWithIndex(func)
sample(withReplacement, fraction, seed)
union(otherDataset)
intersection(otherDataset)
distinct([numTasks]))
groupByKey([numTasks])
reduceByKey(func, [numTasks])
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])
sortByKey([ascending], [numTasks])
join(otherDataset, [numTasks])
cogroup(otherDataset, [numTasks])
cartesian(otherDataset)
pipe(command, [envVars])
coalesce(numPartitions)

Answer 1

这里是关于洗牌转换的一般性说明。

会导致洗牌的转换操作包括重新分区的操作，比如 repartition 和 coalesce，还有按键操作（除了计数）比如 groupByKey 和 reduceByKey，以及连接操作，比如 cogroup 和 join。

来源

Answer 2

这可能对你有帮助：

https://spark.apache.org/docs/latest/programming-guide.html#shuffle-operations

或者这个链接也不错：

http://www.slideshare.net/SparkSummit/dev-ops-training，从第208页开始看

在第209页上提到：

“像distinct这样的转换使用'numPartitions'，可能会进行数据洗牌。”

Answer 3

这里有一份可能会导致数据重新分配的操作列表：

cogroup

groupWith

join: 哈希分区

leftOuterJoin: 哈希分区

rightOuterJoin: 哈希分区

groupByKey: 哈希分区

reduceByKey: 哈希分区

combineByKey: 哈希分区

sortByKey: 范围分区

distinct

intersection: 哈希分区

repartition

coalesce

来源：使用Spark和Scala进行大数据分析，优化分区，Coursera

Answer 4

其实，想要了解这些内容非常简单，甚至不需要查文档。你只需要创建一个RDD，然后调用调试字符串，这里有一个例子，你可以自己尝试其他的。

scala> val a  = sc.parallelize(Array(1,2,3)).distinct
scala> a.toDebugString
MappedRDD[5] at distinct at <console>:12 (1 partitions)
  MapPartitionsRDD[4] at distinct at <console>:12 (1 partitions)
    **ShuffledRDD[3] at distinct at <console>:12 (1 partitions)**
      MapPartitionsRDD[2] at distinct at <console>:12 (1 partitions)
        MappedRDD[1] at distinct at <console>:12 (1 partitions)
          ParallelCollectionRDD[0] at parallelize at <console>:12 (1 partitions)

从上面的例子可以看出，distinct这个函数会引发一次数据洗牌。了解这一点比查文档更重要，因为有些情况下，某个函数是否需要洗牌是有区别的。比如，通常情况下，连接两个数据集（join）是需要洗牌的，但如果你连接的是两个来自同一个RDD的数据集，Spark有时可以省略这次洗牌。