确保底层配置参数设置为true：

• spark.sql.adaptive.enabled个
• spark.sql.adaptive.coalescePartitions.enabled个

在Spark v3.3.0中，默认情况下应该是这样。
另外，请确保将以下配置参数设置为false：

• spark.sql.adaptive.coalescePartitions.parallelismFirst个

在Spark v3.3.0中，默认情况下不是。
关于spark.sql.adaptive.coalescePartitions.parallelismFirst的文档：
当为true时，Spark在合并连续的shuffle分区时忽略spark.sql.adaptive.advisoryPartitionSizeInBytes指定的目标大小（默认为64MB），并且只考虑spark.sql.adaptive.coalescePartitions.minPartitionSize指定的最小分区大小（默认为1MB），以最大化并行度。这是为了在启用自适应查询执行时避免性能下降。建议将此配置设置为false，并遵守spark.sql.adaptive.advisoryPartitionSizeInBytes指定的目标大小。
因此，基本上，在并行化（尝试使用所有内核）和限制分区大小之间存在权衡。将该参数设置为false将确保Spark将尝试合并分区以达到每个分区64MB。
如果您想更改64MB的值，可以通过更改spark.sql.adaptive.advisoryPartitionSizeInBytes来实现。

在您的情况下需要两个设置：spark.sql.adaptive.coalescePartitions.enabled到True和定义spark.sql.adaptive.coalescePartitions.minPartitionNum：这是允许您设置合并分区的最小数量的配置。它可以设置为优化您的用例的值。
我认为在您的情况下，在写入磁盘之前动态计算最佳min_partitions会很有用。建议的解决方案：

from pyspark.sql import DataFrame
from pyspark.rdd import RDD

def get_optimal_min_partitions(df: DataFrame, target_size_per_partition: int = 128 * 1024 * 1024) -> int:
    """
    Get the optimal minimum number of partitions based on dataframe size.
    :param df: The dataframe
    :param target_size_per_partition: The target size per partition in bytes. Default is 128MB.
    :return: The optimal minimum number of partitions.
    """
    # Get the underlying RDD
    rdd = df.rdd

    # Get the size of the data
    total_size = rdd._jrdd.partitions().size()

    # Calculate the optimal minimum number of partitions
    optimal_min_partitions = int(total_size / target_size_per_partition)
    optimal_min_partitions = max(1, optimal_min_partitions)

    return optimal_min_partitions

字符串
在您根据增量df的大小计算出最佳最小分区后，您可以在将最小分区设置为计算值后写入磁盘。

def calculate_and_write_delta(old_df: DataFrame,
                              new_df: DataFrame,
                              id_keys: List[str],
                              deletion_field_name: str,
                              output_path: str,
                              excluded_fields: List[str] = None) -> None:
    # calculate the delta
    delta_df = calculate_delta(old_df, new_df, id_keys, deletion_field_name, excluded_fields)
    
    # calculate optimal minimum partitions
    optimal_min_partitions = get_optimal_min_partitions(delta_df)
    spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", str(optimal_min_partitions))
    
    # write the data
    delta_df.write.parquet(output_path)

型