Hello everyone, I am running Daft/Ray on AWS with two r7g.16xlarge workers. I have encountered a "reduce your request rate" S3 error while executing the following command:

python
df.write_parquet(ROOT_PATH, io_config=io_config, partition_cols=[daft.col(SOME_COL)])

In this case,

SOME_COL

has 256 distinct values. The error message is as follows:

When completing multiple part upload for AWS
Error SLOW_DOWN during CompleteMultipartUpload operation:
Please reduce your request rate. /arrow/cpp/src/arrow/io/interfaces.cc:288: Error ignored when destroying file of type N5arrow2fs12_GLOBAL__N_118 ObjectOutputStream: IO Error: When uploading part for key XXXXX

The likely cause is that each worker is writing multiple small Parquet files under the same HIVE partition (with the same

SOME_COL

value). It seems that after many retries, the write operation eventually succeeds. Increasing the

retry_initial_backoff_ms

might help, but could you suggest a more effective solution? For example, is there a way to configure Daft workers to avoid spilling small Parquet files under the same partitions and instead keep them larger? As a comparison, it seems that DuckDB exports partitioned Parquets in this manner. Thank you in advance!

Hey @Igor Perchersky! Yeah we currently leverage Pyarrow’s writer for writing files and it seems that some folks have been encountering issues with the way they interact with S3. @Sammy Sidhu @Colin Ho for thoughts?

Hi @jay, in this specific case it would be jolly good if each Daft worker will act as "independent duckdb", making its best to reduce the fragmentation of its own parquet files under any given partition...

Yes so we already do default to a good

parquet_target_filesize

(defaults to 512MB) The issue is that in distributed execution, each “independent worker” (think of it as an independent duckdb) received one partition at a time, and performs a write on that partition. That partition is likely not large enough to saturate the

parquet_target_filesize

. (e.g. if the Daft partition is 1GB in size, and you’re writing into 10 logical partitions on S3, then you’ll get files of roughly 100MB in size each) There is a solution we have in the works here which is to perform a streaming write — essentially we would want to not just write per-Daft-partition, but to perform a write across Daft partitions. For now, a potential workaround is to use

.into_partitions(N)

to reduce the number of partitions (hence increasing the number of rows available in each partition to be piped into compacted files). You can try that and see if it fixes some of the issues you’re facing!

Thank you @jay It seems that repartition is an overkill I have found that in my specific case smth. like

df. with_column("id_bucket",df["id"].partitioning.iceberg_bucket(N)).write_parquet(..., partition_cols=[daft.col("id_bucket")])

is good enough

Great to hear!