200个默认分区的spark.sql.shuffle.partitions难题
在许多帖子中，由于一些关于洗牌，分区，due to JOIN，AGGR，等等的问题，有这样的声明-如下面以某种形式显示的：
..。通常，无论何时进行spark sql聚合或连接，这都会导致的分区数= 200。这由spark.sql.shuffle.partitions设置。..。
所以，我的问题是：
- 我们的意思是，如果我们将DF的分区设置为765，例如
- ，则处理发生在765个分区上，但输出将标准化地合并/重新分区为200 -这里指的是word resulting？
- ，或者它是否在合并/重新分区到JOINing，AGGR?
之前的200个分区后使用200个分区进行处理
我问，因为我从来没有看到一个明确的观点。
我做了以下测试：
// genned ad DS of some 20M short rows
df0.count
val ds1 = df0.repartition(765)
ds1.count
val ds2 = df0.repartition(765)
ds2.count
sqlContext.setConf("spark.sql.shuffle.partitions", "765")
// The above not included on 1st run, the above included on 2nd run.
ds1.rdd.partitions.size
ds2.rdd.partitions.size
val joined = ds1.join(ds2, ds1("time_asc") === ds2("time_asc"), "outer")
joined.rdd.partitions.size
joined.count
joined.rdd.partitions.size在第一个测试中-没有定义分区"765")，得到的processing和num分区数是200。尽管如此，post 45704156声明它可能不适用于DFs -这是DS。
在第二个测试-定义sqlContext.setConf("spark.sql.shuffle.partitions"，分区“765”的上，得到的和num分区数是765。尽管如此，post 45704156声明它可能不适用于DFs -这是DS。
回答 2
Stack Overflow用户
发布于 2018-08-21 22:50:06
Spark.sql.shuffle.partitions是一个参数，它决定了在进行连接或聚合等混洗操作时的分区数量，也就是说，数据在节点间移动的位置。另一部分spark.default.parallelism将根据您的数据大小和最大块大小进行计算，在HDFS中是128mb。因此，如果您的作业不做任何混洗，它将考虑默认的并行度值，或者，如果您使用rdd，您可以自己设置它。当洗牌发生时，它将需要200次。
Val df = sc.parallelize(List(1,2,3,4,5),4).toDF() df.count() //这将使用4个分区
Val df1 = df df1除(Df).count //将生成具有2个阶段的200个分区
Stack Overflow用户
发布于 2018-11-07 18:06:51
这是你两种猜测的结合。
假设您有一组具有M个分区的输入数据，并将shuffle partitions设置为N。
在执行连接时，spark读取所有M个分区中的输入数据，并根据键将数据重新洗牌到N个分区。想象一个简单的散列分区程序，应用于键上的散列函数看起来很像A= hashcode( key ) % N，然后这些数据被重新分配给负责处理第A个分区的节点。每个节点可以负责处理多个分区。
在混洗之后，节点将在它们负责的分区中聚合数据。因为这里不需要做额外的混洗，所以节点可以直接产生输出。
因此，总之，您的输出将合并到N个分区，但是它之所以合并是因为它是在N个分区中处理的，而不是因为spark应用了一个额外的混洗阶段来专门将您的输出数据重新划分到N个分区。
https://stackoverflow.com/questions/51950062
复制相似问题
腾讯云开发者
