Overwrite only some partitions in a partitioned spark Dataset

Question

How can we overwrite a partitioned dataset, but only the partitions we are going to change? For example, recomputing last week daily job, and only overwriting last week of data.

Default Spark behaviour is to overwrite the whole table, even if only some partitions are going to be written.

Before Spark 2.3.0 there is a JIRA created for this. In 2.3.0 this is fixed. https://issues.apache.org/jira/browse/SPARK-20236 — wandermonk, Apr 24 '18 at 18:31

score 67 · Accepted Answer · edited Nov 12 '19 at 21:55

67

Since Spark 2.3.0 this is an option when overwriting a table. To overwrite it, you need to set the new spark.sql.sources.partitionOverwriteMode setting to dynamic, the dataset needs to be partitioned, and the write mode overwrite. Example in scala:

spark.conf.set(
  "spark.sql.sources.partitionOverwriteMode", "dynamic"
)
data.write.mode("overwrite").insertInto("partitioned_table")

I recommend doing a repartition based on your partition column before writing, so you won't end up with 400 files per folder.

Before Spark 2.3.0, the best solution would be to launch SQL statements to delete those partitions and then write them with mode append.

edited Nov 12 '19 at 21:55

wiesiu_p

558
7
6

answered Apr 24 '18 at 16:20

Madhava Carrillo

3,998
3
18
24

2

Hi, I tried this but didn't work for me. I had to pass in the overwrite=True as an argument in the insertInto method https://spark.apache.org/docs/2.3.0/api/python/pyspark.sql.html#pyspark.sql.DataFrameWriter.insertInto But thanks for the clue, this rocks – suhprano Dec 10 '18 at 23:16
Doesn't work with `insertInto` but works perfectly with `save.` – cph_sto Aug 21 '19 at 12:29
@MadhavaCarrillo Hi, are you aware of a setting to only delete the partition, instead of replacing it? Thanks – cph_sto Aug 22 '19 at 10:37
@cph_sto not really, no. You can use the drop partition from hive. `alter table t1 drop if exists partition (p1=1);` or even using comparators like `alter table t drop partition (PARTITION_COL>1);` – Madhava Carrillo Aug 22 '19 at 13:18
@neerajbhadani you can use those in hive, in a hive CLI. – Madhava Carrillo Aug 22 '19 at 13:21
@suhprano I experienced that in python, not in Scala. – Madhava Carrillo Aug 22 '19 at 13:22
@MadhavaCarrillo Hi, Many thanks for confirming it. Actually, I has posted a question [here](https://stackoverflow.com/questions/57520227/how-to-delete-a-particular-month-from-a-parquet-file-partitioned-by-month) and someone answered with a similar answer, but I am using `external tables` and not `internal tables `. In addition to that, I am not on HIVE, instead I am using `Jupyter-PySpark`. I don't know how to do it there. – cph_sto Aug 22 '19 at 15:04
Note that your target table format has to support dynamic partitioning. If it does not and you apply the steps above, be aware that your table might be deleted! Learned that the hard way, even though I luckily made a backup before – Markus Jan 07 '22 at 14:02

score 23 · Answer 2 · edited Jul 18 '18 at 12:58

Just FYI, for PySpark users make sure to set overwrite=True in the insertInto otherwise the mode would be changed to append

from the source code:

def insertInto(self, tableName, overwrite=False):
    self._jwrite.mode(
        "overwrite" if overwrite else "append"
    ).insertInto(tableName)

this how to use it:

spark.conf.set("spark.sql.sources.partitionOverwriteMode","DYNAMIC")
data.write.insertInto("partitioned_table", overwrite=True)

or in the SQL version works fine.

INSERT OVERWRITE TABLE [db_name.]table_name [PARTITION part_spec] select_statement

for doc look at here

score 1 · Answer 3 · edited Oct 24 '22 at 12:58

1

This works also for me, since is easier and straighforward

df.write.partitionBy('dt').mode('overwrite').format('parquet').option(
"partitionOverwriteMode", "dynamic").save(path)

Source: https://kontext.tech/article/1067/spark-dynamic-and-static-partition-overwrite

edited Oct 24 '22 at 12:58

Dharman

30,962
25
85
135

answered Oct 24 '22 at 05:25

David Sánchez

544
8
21

Overwrite only some partitions in a partitioned spark Dataset

3 Answers3

Linked

Related