cloud-fan commented on a change in pull request #32594:
URL: https://github.com/apache/spark/pull/32594#discussion_r686501621
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ShufflePartitionsUtil.scala
##########
@@ -21,16 +21,171 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.spark.MapOutputStatistics
import org.apache.spark.internal.Logging
-import org.apache.spark.sql.execution.{CoalescedPartitionSpec,
ShufflePartitionSpec}
+import org.apache.spark.sql.execution.{CoalescedPartitionSpec,
PartialReducerPartitionSpec, ShufflePartitionSpec}
object ShufflePartitionsUtil extends Logging {
final val SMALL_PARTITION_FACTOR = 0.2
final val MERGED_PARTITION_FACTOR = 1.2
/**
- * Coalesce the partitions from multiple shuffles. This method assumes that
all the shuffles
- * have the same number of partitions, and the partitions of same index will
be read together
- * by one task.
+ * Coalesce the partitions from multiple shuffles, either in their original
states, or applied
+ * with skew handling partition specs. If called on partitions containing
skew partition specs,
+ * this method will keep the skew partition specs intact and only coalesce
the partitions outside
+ * the skew sections.
+ *
+ * This method will return an empty result if the shuffles have been
coalesced already, or if
+ * they do not have the same number of partitions, or if the coalesced
result is the same as the
+ * input partition layout.
+ *
+ * @return A sequence of sequence of [[ShufflePartitionSpec]]s, which each
inner sequence as the
+ * new partition specs for its corresponding shuffle after
coalescing. If Nil is returned,
+ * then no coalescing is applied.
+ */
+ def coalescePartitions(
+ mapOutputStatistics: Seq[Option[MapOutputStatistics]],
+ inputPartitionSpecs: Seq[Option[Seq[ShufflePartitionSpec]]],
+ advisoryTargetSize: Long,
+ minNumPartitions: Int): Seq[Seq[ShufflePartitionSpec]] = {
+ assert(mapOutputStatistics.length == inputPartitionSpecs.length)
+
+ if (mapOutputStatistics.isEmpty) {
+ return Seq.empty
+ }
+
+ // If `minNumPartitions` is very large, it is possible that we need to use
a value less than
+ // `advisoryTargetSize` as the target size of a coalesced task.
+ val totalPostShuffleInputSize =
mapOutputStatistics.flatMap(_.map(_.bytesByPartitionId.sum)).sum
Review comment:
We need a better algorithm to do partition coalescing, which considers
different operators (Union vs Join for example)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
