什么是引起Shuffle的Spark转换？

Question

我很难在Spark文档中找到导致随机播放的操作，而不会导致随机播放的操作。在此列表中，哪些会导致洗牌，哪些不会导致洗牌？

地图和过滤器没有。但是，我不确定其他人。

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

没有文档，实际上很容易发现这一点。对于这些功能中的任何一个，只需创建一个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创建了一个洗牌。发现这种方式而不是docs尤其重要，因为在某些情况下某些功能将需要或不需要改组。例如，join通常需要进行改组，但是如果您将两个RDD合并，则来自同一RDD
Spark的分支有时可能会取消改组。