alexeykudinkin commented on code in PR #7825:
URL: https://github.com/apache/hudi/pull/7825#discussion_r1094810752
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieDatasetBulkInsertHelper.scala:
##########
@@ -220,4 +215,39 @@ object HoodieDatasetBulkInsertHelper
val keyGenerator = ReflectionUtils.loadClass(keyGeneratorClassName, new
TypedProperties(config.getProps)).asInstanceOf[BuiltinKeyGenerator]
keyGenerator.getPartitionPathFields.asScala
}
+
+ /**
+ * We use custom Spark [[Partitioner]] that is aware of the target table's
partitioning
+ * so that during inevitable shuffling required for de-duplication, we also
assign records
+ * into individual Spark partitions in a way affine with target table's
physical partitioning
+ * (ie records from the same table's partition will be co-located in the
same Spark's partition)
+ *
+ * This would allow us to
+ * <ul>
+ * <li>Save on additional shuffling subsequently (by
[[BulkInsertPartitioner]])</li>
+ * <li>Avoid "small files explosion" entailed by random (hash)
partitioning stemming
+ * from the fact that every Spark partition hosts records from many
table's partitions
+ * resulting into every Spark task writing into their own files in these
partitions (in
+ * case no subsequent re-partitioning is performed)
+ * </li>
+ * <ul>
+ *
+ * For more details check out HUDI-5685
+ */
+ private case class TablePartitioningAwarePartitioner(override val
numPartitions: Int,
+ val isPartitioned:
Boolean) extends Partitioner {
Review Comment:
Good point
--
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]
