Writing speed in Delta tables significantly increases after copying it in Databricks

user_8588000 0
Viewed 1291

I am merging a PySpark dataframe into a Delta table. The output delta is partitioned by DATE. The following query takes 30s to run:

query = DeltaTable.forPath(spark, PATH_TO_THE_TABLE).alias(
            "actual"
        ).merge(
            spark_df.alias("sdf"),
            "actual.DATE >= current_date() - INTERVAL 1 DAYS 
             AND (actual.feat1 = sdf.feat1) 
             AND (actual.TIME = sdf.TIME) 
             AND (actual.feat2 = sdf.feat2) "
            ,
        ).whenNotMatchedInsertAll()

After copying the content of the Delta to another location, the above query becomes 60 times faster (i.e. it takes 0.5s on the same cluster), when using the NEW_PATH instead of PATH_TO_THE_TABLE. Here is the command to copy the delta:

(spark.read.format("delta").load(PATH_TO_THE_TABLE).write.format(
        "delta"
    )
.mode("overwrite").partitionBy(["DATE"]).save(NEW_PATH))

How can I make querying on the first delta as fast as on the new one? I understand that Delta has a versioning system and I suspect it is the reason it takes so much time. I tried to vacuum the Delta table (which lowered the query time to 20s) but I am still far from the 0.5s.

Stack:

  • Python 3.7;
  • Pyspark 3.0.1;
  • Databricks Runtime 7.3 LTS
delta-lake pyspark databricks sql performance
1 Answers

I see that you compare actual.DATE >= current_date() as this is most important part of your query please try to run periodically ZORDER sorting delta by thta filed:

OPTIMIZE actual ZORDER BY (actual.DATE)

You can also try totally vacuum delta:

VACUUM actual RETAIN 0 HOURS

to do that you need to set spark.databricks.delta.retentionDurationCheck.enabled false.

If you don't want benefits of delta (transaction, concurrent writes, timetravel history etc.) you can just use parquet.

user_109729590
Related