How to write datafusion record batch to s3?

[dadepo]

Through a discussion on the project's Slack, I learned how to write results from a Datafusion query to a Delta table, which can then be stored in a Delta Lake location on the file system. The key part of this process is demonstrated in the following code:

    let res = deltalake::operations::write::WriteBuilder::default()
        .with_location("../data/out")
        .with_save_mode(SaveMode::Overwrite)
        .with_input_batches(batches.into_iter()).await?;

However, I am currently trying to write to an S3 location and have been unable to configure the with_location method with the necessary information for an S3 endpoint.

How do I go about this?

[roeap]

in order for S3 to work you would need an url like s3://<bucket>/<path>. In most cases, additional information is required, which is passed via with_storage_options to the builder. The available keys are documented in the object strore crate.

[dadepo]

I have been able to use WriteBuilder to write stuff to minio. I can confirm this because I see the buckets and object via the minio dashboard but on trying to read it back, I get

Error: ObjectStore { source: Generic { store: "DeltaS3ObjectStore", source: LockClientRequired } }

I searched the docs for delta-rs and could not find any instance of LockClientRequired.

Not sure what the LockClientRequired refers to but I noticed a difference between another delta lake folder I created via spark and this one that was created programmatically.

The one created via spark has what looks like normal version files under _delta_log that is 00000000000000000000.json etc, while the one created programmatically has files with names like _commit_e15ca9b5-d651-47a6-a12a-c3c0ab87aefe.json.tmp within the _delta_log path.

Any ideas what could be going on?

[roeap]

LockClientRequired refers to an error variant. More specifically, delta rs is build for multi writer scenarios. This requires atomic rename support, which S3 does not have. To mitigate a DynamoDB is used to communicate locks on the table. To avoid having to use a dynamo db you can set AWS_S3_ALLOW_UNSAFE_RENAME to "TRUE" in the environment.

the commit file you se is an artifact from a failed commit to the table. i.e. data is written to a tmp file first, and then renamed into the commit json file. If the commit fails, the rename does not happen.