I have a partitioned delta table with about 10 million records, with the largest partition's parquet file coming in at ~170MB. This table gets rebuilt daily on a 15 node AWS glue cluster (spark 3.3.0, delta version 2.2.0) and takes about 10 minutes. Now I'm trying to merge intraday updates into this table using the same glue cluster. There are about 80k updated records being created, and I repartition them using the same partition column as the final delta table before attempting to merge. The problem is that the merge operation is hanging, and I can't figure out why. I've allowed the merge stage to run for 45 minutes before killing it. I've increased the driver memory to 20GB hoping that would solve the issue, but no luck. Has anybody seen this behavior, or have any suggestions on what I can try to make this merge work?

It might be an issue related to repartitionBeforeRewrite, check this => https://github.com/delta-io/delta/issues/1474

thank you, I'll turn off that option in config and test again and report back

Can you share the code you are running and the executor spark config?

Here is the relevant snippet of my code. 'final_table' is attempting to merge into the delta table when the job hangs. The dataframe has already been cached by this point and repartitioned to match the partitoning of the target delta table. Normally rewriting the full table of 10M rows takes seconds. I've also attached the spark config. I appreciate the feedback.

final_table = spark.sql(query_main4).repartition('orderid_group')
final_table.createOrReplaceTempView("final_table")
final_table.cache()
x=final_table.count()
print(f"final_table: {x}; {datetime.now().strftime('%H:%M:%S')}")
orders_part3.unpersist()


#WRITE OUT TO DELTA TABLE
if run_type=='intraday':
    #THIS BLOCK HANGS
    #'final_table' HAS 80k ROWS
    #'dt' IS PARTITIONED BY 'orderid_group'; 'orderid' is a bigint
    dt = DeltaTable.forPath(spark, f"s3a://{transformed_delta_path}/{final_table_name}")
    (dt.alias("a")
       .merge(source=final_table.alias("b"), condition="a.orderid_group=b.orderid_group and a.orderid=b.orderid")
       .whenMatchedUpdateAll()
       .whenNotMatchedInsertAll()
       .execute()
    )
else:
    #THIS BLOCK RUNS DAILY WITHOUT ISSUE
    (final_table
        .write
        .format("delta")
        .mode('overwrite')
        .option('overwriteSchema', "true")
        .partitionBy(partition_columns)
        .save(f"s3a://{transformed_delta_path}/{final_table_name}")
    )

config looks good, what is the size of this final table (source)

final_table is about 80,000 rows. I'll try changing the broadcast threshold, but would that have any other performance implications or would it only affect this merge?

check if it is broadcast join then try setting

spark.sql.autoBroadcastJoinThreshold=-1

thank you, I'll try that shortly and report back

👍 delta io

disabling the broadcast join had no effect. What I ended up doing was creating a dataframe from the delta table, removing the rows that exist in the source table, unioning the source dataframe to this filtered dataframe, and finally overwriting the delta table with this new dataframe. It now takes the same amount of time as my daily job that recreates the full table but I should be able to run it on a smaller cluster.