Crash-resumable Parquet streaming to S3 from Python. One multipart upload per file, a checkpoint any process can finish, parallel reads back.

Install · Quickstart · How it works · Performance · API

Highlights

• Resumes after a crash: a checkpoint is a small blob; any process turns it into a valid Parquet file without re-uploading a byte.
• About 2× pyarrow on writes (219 vs 116 MiB/s), with parallel reads (benchmarks).
• Rust core, Arrow in and out without copies: PyO3 bindings, GIL released around every network call.
• No conditional writes needed: works on S3-compatible stores that ignore If-Match.

writer = arroyo.ParquetStreamWriter(config, "events/2026-10-28.parquet", batch.schema)
writer.write(batch)
state = writer.checkpoint()        # small, serializable, survives the process
...                                # the process dies here
arroyo.finish(config, state)        # any other process: a valid Parquet file, no re-upload

Why

S3 objects cannot be appended to, and a streaming job that writes one file across many checkpoints has to keep a multipart upload open. Existing Python writers (pyarrow's S3FileSystem, obstore) stream a multipart upload but cannot resume one: when the process dies, the file is gone. arroyo implements the method from Arroyo's filesystem sink (Streaming to S3 is hard) so that a checkpoint carries enough to complete the file later, from anywhere, with nothing already uploaded sent again. It needs no conditional writes, so it works on S3-compatible stores that accept and ignore If-Match.

Install

pip install arroyo-python

Python 3.12 or newer. Wheels ship for Linux (x86_64, aarch64), macOS (arm64, x86_64) and Windows. Building from source needs a Rust toolchain and maturin.

How it works

Write path

Batches are buffered until a row group is full. Every leaf column is then encoded on its own rayon task, the encoded bytes are sliced into fixed-size parts without copying, and the parts upload on a tokio runtime while the next row group is encoded. The GIL is released around every network call.

Checkpoint and recovery

checkpoint() returns a WriterState: the parts acknowledged so far and the trailing bytes, which are the bytes not yet uploaded plus a Parquet footer for the row groups written so far. finish(config, state) uploads the trailing bytes as the last part(s) and completes the upload. The result is a valid Parquet file holding every row up to that checkpoint. Calling it twice is safe.

close() writes the real footer as the last part and returns a PendingCommit; commit() completes the upload. That split is the second phase of a two-phase commit: record the pending state durably first, then commit.

Read path

read_parquet reads the footer with one range request, then fetches concurrency row groups ahead of the consumer with coalesced range GETs, decodes them on the blocking pool and hands batches to Python through the Arrow C stream interface, in file order.

Quickstart

import pyarrow as pa
import arroyo

config = arroyo.S3Config.from_env(bucket="events")   # AWS_ENDPOINT, AWS_ACCESS_KEY_ID, ...

writer = arroyo.ParquetStreamWriter(
    config,
    "events/part-0001.parquet",
    schema,                            # anything with __arrow_c_schema__
    part_size=16 * 1024 * 1024,        # 5 MiB .. 5 GiB
    row_group_size=128 * 1024 * 1024,  # buffered Arrow bytes that close a row group
    compression="zstd",
)
writer.write(batch)                    # RecordBatch, Table, RecordBatchReader, polars ...

state = writer.checkpoint()            # closes the row group, waits for parts in flight
blob = state.to_bytes()                # store this with your own checkpoint
writer.write(next_batch)               # same file, same upload

pending = writer.close()               # real footer uploaded, upload not yet completed
record(pending.to_bytes())             # your durable checkpoint
pending.commit()                       # completes the multipart upload

# Another process, after a crash, from the last recorded blob:
arroyo.finish(config, arroyo.WriterState.from_bytes(blob))

reader = arroyo.read_parquet(config, "events/part-0001.parquet", columns=["a", "b"], concurrency=8)
table = pa.RecordBatchReader.from_stream(reader).read_all()

S3Config pickles, so it can be handed to multiprocessing workers.

Performance

Apple M3 Pro, 12 cores, 20,000,000 rows (801 MiB of Arrow data) per file, pyarrow 26.0.0, obstore 0.12.1, arroyo 0.1.0. arroyo reads with concurrency=8; the obstore baseline reads through pyarrow's own multithreaded decoder.

The benchmark writes and reads one Parquet file of synthetic int64, float64 and string columns against a local S3-compatible server (RustFS in Docker) with every library at its defaults. Numbers are the mean of three runs; make bench and python benchmarks/plot.py reproduce the chart on your machine.

Rules for callers

• One writer per state. There is no compare-and-set. Stop the old writer before finish(state) runs in a new process. A writer that is still running can complete the upload with a different part list; one of the two wins and the other fails the size check with ArroyoError.
• Abandoned uploads cost storage. Configure an AbortIncompleteMultipartUpload lifecycle rule on the bucket, and call arroyo.abort(config, state) or writer.abort() for uploads you give up on. The writer aborts its upload when a with block exits with an exception. A writer that is dropped without close() or abort() leaves its upload in place, so a recorded checkpoint stays finishable.
• Keep files short-lived. Some providers expire upload IDs after days. Roll files by size or time. A file may hold at most 10,000 parts; the writer raises before it would send part 10,001.
• Every checkpoint() closes a row group, so frequent checkpoints mean small row groups.

API

Writer options

Reader options

Memory for a read is about concurrency row groups' worth of compressed bytes plus a couple of decoded batches per row group.

Development

uv venv && source .venv/bin/activate
uv pip install maturin pytest pyarrow "moto[server]" obstore numpy
make develop     # maturin develop
make ci          # cargo fmt --check, clippy -D warnings, cargo test
make test-py     # pytest against an in-process moto S3
make s3          # RustFS in Docker on 127.0.0.1:9000 (rustfsadmin / rustfsadmin)

Set ARROYO_TEST_ENDPOINT, ARROYO_TEST_BUCKET and the AWS_* variables to run pytest against RustFS or any other store instead of moto. The benchmark takes ARROYO_BENCH_ENDPOINT and ARROYO_BENCH_BUCKET:

python benchmarks/bench_throughput.py --rows 20000000 --repeat 3 --json results.json
python benchmarks/plot.py results.json benchmarks/results/throughput.svg

License

MIT OR Apache-2.0. The write method follows Arroyo (MIT OR Apache-2.0).