Overwrite specific partitions in spark dataframe write method

Question

I want to overwrite specific partitions instead of all in spark. I am trying the following command:

df.write.orc('maprfs:///hdfs-base-path','overwrite',partitionBy='col4')

where df is dataframe having the incremental data to be overwritten.

hdfs-base-path contains the master data.

When I try the above command, it deletes all the partitions, and inserts those present in df at the hdfs path.

What my requirement is to overwrite only those partitions present in df at the specified hdfs path. Can someone please help me in this?

Solution 1

This is a common problem. The only solution with Spark up to 2.0 is to write directly into the partition directory, e.g.,

df.write.mode(SaveMode.Overwrite).save("/root/path/to/data/partition_col=value")

If you are using Spark prior to 2.0, you'll need to stop Spark from emitting metadata files (because they will break automatic partition discovery) using:

sc.hadoopConfiguration.set("parquet.enable.summary-metadata", "false")

If you are using Spark prior to 1.6.2, you will also need to delete the _SUCCESS file in /root/path/to/data/partition_col=value or its presence will break automatic partition discovery. (I strongly recommend using 1.6.2 or later.)

You can get a few more details about how to manage large partitioned tables from my Spark Summit talk on Bulletproof Jobs.

Solution 2

Finally! This is now a feature in Spark 2.3.0: SPARK-20236

To use it, you need to set the spark.sql.sources.partitionOverwriteMode setting to dynamic, the dataset needs to be partitioned, and the write mode overwrite. Example:

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.

Solution 3

spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
data.toDF().write.mode("overwrite").format("parquet").partitionBy("date", "name").save("s3://path/to/somewhere")

This works for me on AWS Glue ETL jobs (Glue 1.0 - Spark 2.4 - Python 2)

Solution 4

Adding 'overwrite=True' parameter in the insertInto statement solves this:

hiveContext.setConf("hive.exec.dynamic.partition", "true")
hiveContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")

df.write.mode("overwrite").insertInto("database_name.partioned_table", overwrite=True)

By default overwrite=False. Changing it to True allows us to overwrite specific partitions contained in df and in the partioned_table. This helps us avoid overwriting the entire contents of the partioned_table with df.

Solution 5

Using Spark 1.6...

The HiveContext can simplify this process greatly. The key is that you must create the table in Hive first using a CREATE EXTERNAL TABLE statement with partitioning defined. For example:

# Hive SQL
CREATE EXTERNAL TABLE test
(name STRING)
PARTITIONED BY
(age INT)
STORED AS PARQUET
LOCATION 'hdfs:///tmp/tables/test'

From here, let's say you have a Dataframe with new records in it for a specific partition (or multiple partitions). You can use a HiveContext SQL statement to perform an INSERT OVERWRITE using this Dataframe, which will overwrite the table for only the partitions contained in the Dataframe:

# PySpark
hiveContext = HiveContext(sc)
update_dataframe.registerTempTable('update_dataframe')

hiveContext.sql("""INSERT OVERWRITE TABLE test PARTITION (age)
                   SELECT name, age
                   FROM update_dataframe""")

Note: update_dataframe in this example has a schema that matches that of the target test table.

One easy mistake to make with this approach is to skip the CREATE EXTERNAL TABLE step in Hive and just make the table using the Dataframe API's write methods. For Parquet-based tables in particular, the table will not be defined appropriately to support Hive's INSERT OVERWRITE... PARTITION function.

Hope this helps.

View more solutions

Comments

[Admin]

Thanks a lot Sim for answering. Just few doubts more, if suppose initial dataframe has data for around 100 partitions, then do I have to split this dataframe into another 100 dataframes with the respective partition value and insert directly into the partition directory. Can saving these 100 partitions be done in parallel? Also I am using Spark 1.6.1 If I am using orc file format, how can I stop emitting metadata files for that, is it same which you have mentioned for parquet?

[Sim]

Re: metadata, no, ORC is a different format and I don't think it produces non-data files. With 1.6.1 you need only ORC files in the subdirectories of the partition tree. You'll therefore have to delete _SUCCESS by hand. You can write in parallel to more than one partition but not from the same job. Start multiple jobs based on your platform capabilities, e.g., using REST API.

[David H]

Any update about that? Does saveToTable() will overwrite just specific partitions? Does spark smart enough to figure out which partitions were overwritten?

[Shankar]

I tried the above approach, i'm getting the error like Dynamic partition strict mode requires at least one static partition column. To turn this off set hive.exec.dynamic.partition.mode=nonstrict

[Shankar]

i dont have any static partition columns

[Carlos Verdes]

This remove the previous partitions if they are not in the current dataframe.

[Madhava Carrillo]

It was hard for me to find the setting to use this, so leaving here the reference: stackoverflow.com/questions/50006526/…

[OneCricketeer]

Can you please edit the answer to show example code from the JIRA?

[Neeraj Bhadani]

How to update the data if table is partitioned based on multiple columns say year, month and I only want to overwrite based on year?

[Neeraj Bhadani]

Also I am getting error : AnalysisException: u"insertInto() can't be used together with partitionBy(). Partition columns have already be defined for the table. It is not necessary to use partitionBy().;"

[Neeraj Bhadani]

without partitionBy I am getting duplicate data inserted even with mode("overwrite")

[pavel_orekhov]

Doesn't work. The new data that is not yet in HDFS is not written to it.

[pavel_orekhov]

Please take a look here stackoverflow.com/questions/54246038/… maybe you can help me

[y2k-shubham]

If I'm overwriting a single partition and I know name of that partition apriori, is there a way to specify that to spark like we can do in Hive? I'm asking this because that would give me a lot of assurance and kind of work as sanity check, plus I believe there would be some performance benefit too (since runtime resolution of partition for every record wouldn't be required)

[Tetlanesh]

@y2k-shubham yes, use spark.sql('insert overwrite table TABLE_NAME partition(PARTITION_NAME=PARTITION_VALUE) YOUR SELECT STATEMENT) This works for 2.2 at least, not suere if earlier versions support this.

[enneppi]

where do you specify the partitions?

[Davos]

How does this approach behave with the job bookmark? Say you have an existing partition (e.g. day) which only has the first 12 hours of data for the day, and new files have arrived in your source that are for the second 12 hours that should be added to the partition, I worry that the Glue job bookmark is pretty naive and it will end up only writing data from the new files for that second 12 hours. Or do you not use the job bookmark?

[Davos]

Why Python 2? Also this looks like Databricks specific, good to mention that for others not using that platform. I like idempotent but is this really? What if deleting the directory is successful but the append is not? How do you guarantee the df contains the deleted partition's data?

[Zach]

Great question! I had exactly the same concern. My use case is that I specifically ask Glue to re-process certain partitions and re-write the results (using the above two lines). With the job bookmark enabled, it refuses to re-process the "old" data.

[Davos]

So you don't use the bookmark? That was pretty much the only reason I could see for bothering with the glueContext over just sticking with Spark. I don't want to manage the processed state, but I'm finding the bookmark is flaky, relying on file modified timestamps and no way to sync it apart from a brutal reset. Why Python 2 and not 3?

[Zach]

Yeah, the job bookmark has been bothering me for a while now. It's good for some low profile day-to-day work. But once you have a little bit "off-road" actions, that thing is less than useless. Regarding the Python version, when upgrading from Glue 0.9, looking at the two options (Python 2 vs 3), I just didn't want to break anything since the code was written in Python 2 era ^_^

[Davos]

"less than useless", noted. Apart from print is a function, unicode done properly and the literal long not necessary there's not much in going 2->3. The Pyspark DSL syntax seems identical. Python 2 is officially unsupported in 2020, time to abandon it.

[Miroslaw]

Works good for me. I do not use hive table: someDf.write.format("orc").mode(SaveMode.Overwrite).partitio‌​nBy("partitioned_col‌​umn").save(path_to_w‌​rite_orc)

[user2552108]

This still overwrites the entire table.

[thebluephantom]

Seems to have changed this approach.

[sg1234]

This worked for me, although it wouldn't accept the "overwrite=True" parameter in Spark 2.3

[Costin Aldea]

This is partially correct. See Surya Murali comment for the additional settings I needed to add in order for it to work. At least in my case worked (spark 1.6, scala)

[Camilo Velasquez]

Thanks, it seems to work. I am also updating the partitions in Athena and it seems to work well. Thanks!

[kmes40505]

insertInto doesn't support clusterby...