Delta: Repartition Before Merge Creating Big Files

[rubenssoto]

Hi Guys

I created a delta dataset with year-month partition, each partition has 1.5gb - 2.5gb, so start my streaming job with merge. I enabled the option spark.delta.merge.repartitionBeforeWrite like documentation recommend.
Problably delta is running repartition(partition_column) and creates bigger files, like 2gb, 2.5gb.

Is there any way to tells delta repartition in more files? It could be a new feature?

Thank you

[jaehc]

I have a similar problem with this. I would say that if Delta Lake has a feature like post-merge transformation such as sorting or bucketing, etc, we would get much more efficiency in writing and reading data which has an irregular distribution in size or is too large in terms of partitions.

[vikatskhay]

We have the same issue but in batch mode.

Our table is partitioned by a column X, the data is not evenly distributed among the partitions and there's a lot of data in general. If there was less data to merge and write, it'd work perfectly, but unfortunately it's not the case for us. And if we don't use the repartitionBeforeWrite feature, we get a lot of small files and it takes very long to write them out to S3.

Looking at the implementation: https://github.com/delta-io/delta/commit/6d9420539e9b00f8df8d5bc80ce6aa7a5a9f3b7d#diff-3176b09e3efb247c52b675007bcb1a8dR428-R442
it becomes clear why it happens:

df.repartition(partitionColumns.map(col): _*)

will shuffle data into spark.sql.shuffle.partitions but in such a way that all rows for each logical partition (partitionColumns) will be in the same rdd partition.

In our case some logical partitions have tens of GBs of data and it lands on S3 as very big objects and the write stage takes ages as some reducers have to spill heavily, whereas those that have tiny amounts of data (our dataset is not evenly distributed) have almost no work to do.

As I understand, what would help eliminate this issue is if MergeIntoCommand instead of calling df.repartition(partitionColumns.map(col): _*) did something like this:

df.repartitionByRange(repartitionByRangeBeforeMergeColumns.map(col): _*)

where repartitionByRangeBeforeMergeColumns would need to be partitionColumns + someIdColumn. someIdColumn would need to be some more or less unique identifier or hash that would allow data to be spread more or less evenly over spark.sql.shuffle.partitions. This way output files in big chunky logical partitions will be approx {total_dataframe_size} / {spark.sql.shuffle.partitions} in size.

We appreciate it's a lot to ask from project maintainers, but it would be super valuable for use cases where delta tables are partitioned by one or especially more columns. It will help speed up writes and get more optimized data layout for reads. We run compactions to optimize reads but there is not much we can do to help slow writes.

[wadejensen]

Ditto what @vikatskhay has said. Although I can understand why repartitionByRange might not be the default.
Perhaps the partition strategy could be exposed as a parameter?

For example: spark.delta.merge.repartitionerStrategy: "exact'' | "range" | "count"?

[windpiger]

turn on the Adaptive Query Execution, it can reduce small files when do Merge or do some execution which include Join operator

[vikatskhay]

@windpiger we did try that, unfortunately it didn't change the situation with many small files (with repartitionBeforeWrite disabled)

[ZoharMeir]

We are facing a similar issue.

I really liked @vikatskhay's suggestion, but was unable to figure out how to dynamically control the number of output files in each partition based on its size. So as a starting point, I just added the option to allow setting the number of output files per partition in the merge operation.

The change involves replacing df.repartition(partitionColumns.map(col): _*) in MergeIntoCommand.repartitionIfNeeded with the following code:

val numFilesPerPartition = spark.conf.get(DeltaSQLConf.MERGE_NUM_FILES_PER_PARTITION)
if (numFilesPerPartition == 1) {
  df.repartition(partitionColumns.map(col): _*)
} else {
  val partitionColumnsExt = partitionColumns.map(col) :+ ceil(rand() * numFilesPerPartition)
  df.repartitionByRange(partitionColumnsExt: _*)
}

And also requires the addition of the following configuration to DeltaSQLConf:

val MERGE_NUM_FILES_PER_PARTITION =
  buildConf("merge.repartitionBeforeWrite.numOfFiles")
    .internal()
    .doc(
      """
        |Determine the number of output files for each output partition when
        |repartitionBeforeWrite is enabled.
      """.stripMargin)
    .intConf
    .checkValue(_ >= 1, "numOfFiles must be positive.")
    .createWithDefault(1)

Hope this can help someone.

[ZoharMeir]

If anyone is interested, this is how our current solution looks like:

  protected def repartitionIfNeeded(
      spark: SparkSession,
      df: DataFrame,
      partitionColumns: Seq[String]): DataFrame = {
    if (partitionColumns.nonEmpty && spark.conf.get(DeltaSQLConf.MERGE_REPARTITION_BEFORE_WRITE)) {
      val numFilesPerPartition = spark.conf.get(DeltaSQLConf.MERGE_NUM_FILES_PER_PARTITION)
      val extraPartitionColumn = if (numFilesPerPartition == 1) None
      else {
        val nonPartitionColumns = df.columns.diff(partitionColumns)
        if (nonPartitionColumns.isEmpty) None
        else {
          val partitionSplitId = abs(hash(nonPartitionColumns.map(col): _*) % numFilesPerPartition)
          Some(partitionSplitId)
        }
      }
      val partitionExpressions = partitionColumns.map(col) ++ extraPartitionColumn
      df.repartition(partitionExpressions: _*)
    } else {
      df
    }
  }

This allows us to set spark.databricks.delta.merge.repartitionBeforeWrite.numOfFiles to our desired number of output files per partition