Ways to write fast millions of rows inside a new delta table

[jeremy98]

Hello everyone,
I am facing an issue with writing 100–500 million rows (partitioned by a column) into a newly created Delta table. I have set up a cluster with 256 GB of memory and 64 cores. However, the following code takes a considerable amount of time even when writing around 70 million rows:

            df.repartition(num_partitions*4, partition_col).write \
                .format("delta") \
                .mode("overwrite") \
                .partitionBy(partition_col) \
                .option("mergeSchema", "true") \
                .option("optimizeBucket", "true") \
                .option("maxRecordsPerFile", "1000000") \
                .option("autoOptimize.optimizeWrite", "true") \
                .option("autoOptimize.autoCompact", "true") \
                .option("autoOptimize.autoRepartition", "true") \
                .saveAsTable(f"{bronze_layer}.{table_name}")

 What Do i need to adjust to speed up this written step?

[jeremy98]

Someone can help me?

[radothede]

Hi @jeremy98 

This is what I would suggest to test:

1) remove repartition step or reduce number or partitions (start with number of cores and then try to increase it x2, x3)

repartition(num_partitions*4, partition_col)

I know repartitioning helps to divide data into equally smaller chunks and distribute tasks across the cores, however, on the other side, it triggers shuffle, which might be expensive. 

2) whats the cardinality of column You are partitioning the data in this step?

It triggers another shuffle.

.partitionBy(partition_col)

how many partitions does this one creates?

3) do You really need this?

.option("optimizeBucket", "true") \
.option("autoOptimize.optimizeWrite", "true") \
.option("autoOptimize.autoCompact", "true") \
.option("autoOptimize.autoRepartition", "true") \

It requires shuffle. 

Maybe You could schedule a job that optimize the table (once per day or whatever frequency you need it to run) and tune the files size, here is an example of 128 mbs: 

SET spark.databricks.delta.optimize.maxFileSize = 134217728;
OPTIMIZE my_delta_table;

4) whats the data structure under the table? partitioned? bucketed? 

5) while writing Your table You can investigate cluster metrics and check if all the cores are evenly loaded and the number of spark tasks being executed at the same time is what You would expect