Is there a way to overwrite existing data using pandas to_parquet with partitions?

user_13015551 0
Viewed 319

I'm using pandas to write a parquet file using the to_parquet function with partitions. Example:

df.to_parquet('gs://bucket/path', partition_cols=['key'])

The issue is that every time I run the code. It adds a new parquet file in the partition and when you read data, you get all the data from each time the script was run. Essentially, the data appends each time.

Is there a way to overwrite the data every time you write using pandas?

parquet pandas python
1 Answers

I have found dask to be helpful reading and writing parquet. It defaults the file name on write (which you can alter) and will replace the parquet file if you use the same name, which I believe is what you are looking for. You can append data to the partition by setting 'append' to True, which is more intuitive to me, or you can set 'overwrite' to True which will remove all files in the partition/folder prior to writing the file. Reading parquet works well as well by including partition columns in the dataframe on read.

https://docs.dask.org/en/stable/generated/dask.dataframe.to_parquet.html

See below some code I used to satisfy myself of the behaviour of dask.dataframe.to_parquet:

import pandas as pd
from dask import dataframe as dd
import numpy as np

dates = pd.date_range("2015-01-01", "2022-06-30")
df_len = len(dates)
df_1 = pd.DataFrame(np.random.randint(0, 1000, size=(df_len, 1)), columns=["value"])
df_2 = pd.DataFrame(np.random.randint(0, 1000, size=(df_len, 1)), columns=["value"])

df_1["date"] = dates
df_1["YEAR"] = df_1["date"].dt.year
df_1["MONTH"] = df_1["date"].dt.month

df_2["date"] = dates
df_2["YEAR"] = df_2["date"].dt.year
df_2["MONTH"] = df_2["date"].dt.month

ddf_1 = dd.from_pandas(df_1, npartitions=1)
ddf_2 = dd.from_pandas(df_2, npartitions=1)

name_function = lambda x: f"monthly_data_{x}.parquet"

ddf_1.to_parquet(
    "dask_test_folder",
    name_function=name_function,
    partition_on=["YEAR", "MONTH"],
    write_index=False,
)

print(ddf_1.head())
ddf_first_write = dd.read_parquet("dask_test_folder/YEAR=2015/MONTH=1")
print(ddf_first_write.head())

ddf_2.to_parquet(
    "dask_test_folder",
    name_function=name_function,
    partition_on=["YEAR", "MONTH"],
    write_index=False,
)

print(ddf_2.head())
ddf_second_write = dd.read_parquet("dask_test_folder/YEAR=2015/MONTH=1")
print(ddf_second_write.head())
user_127668130
Related