faucet-source-s3

An AWS S3 source that lists objects under a prefix and parses them as JSON Lines, a JSON array, or raw text. Part of the faucet-stream ecosystem.

Built on the official aws-sdk-s3 client (built once, reused across every read) with concurrent object fetches via buffer_unordered, it pulls a whole bucket prefix into any faucet-stream sink as fast as the storage and your parallelism allow. Point it at MinIO / LocalStack / Cloudflare R2 with a custom endpoint_url, and let the standard AWS credential chain handle auth.

Feature highlights

• Three file formats — json_lines (one record per line), json_array (one record per array element), and raw_text (one {key, content} record per object).
• Parallel object reads — up to concurrency objects fetched at once via futures::buffer_unordered (default 10).
• True line-level streaming — for json_lines / raw_text, object bodies are decoded line-by-line via tokio::io::AsyncBufReadExt, so client memory is bounded at O(batch_size) regardless of file or scan size.
• Prefix listing with pagination — handles truncated ListObjectsV2 responses transparently and honours an optional max_objects cap.
• S3-compatible — custom endpoint_url for MinIO, LocalStack, R2, and other S3-API services.
• Optional compression — behind the compression feature: gzip / zstd / auto (suffix-detected per object key), so a single run can mix compressed and uncompressed objects.
• Read-integrity verification — every object's byte length is checked against the Content-Length S3 advertises (verify_length, default on), so a cleanly-truncated transfer is rejected instead of silently parsed as a complete object. Opt into full checksum verification (verify_checksum) for SHA-256 / CRC-32 / CRC-32C / CRC-64-NVME / ETag-MD5.
• Matrix-friendly — ${parent.field} tokens in prefix are substituted per parent record in parent/child pipelines.

Installation

# As a library:
cargo add faucet-source-s3
cargo add tokio --features full

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-s3

# With compression support:
cargo install faucet-cli --features "source-s3,compression"

Or via the umbrella crate:

cargo add faucet-stream --features source-s3

source-s3 is opt-in — it is not part of the CLI/umbrella default feature set.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
name: s3_to_postgres
pipeline:
  source:
    type: s3
    config:
      bucket: my-data-lake
      prefix: events/2026/
      region: us-east-1
      file_format: json_lines
      concurrency: 16
  sink:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost/warehouse
      table_name: events_raw
      column_mapping:
        type: jsonb
        column: payload

export AWS_ACCESS_KEY_ID=...    # standard AWS credential chain
export AWS_SECRET_ACCESS_KEY=...
faucet run pipeline.yaml

Configuration reference

Core

File formats (S3FileFormat)

Read-integrity verification

Object bodies are read through a verifying reader that validates the transfer at EOF, so a stream that ends early but cleanly (a truncated transfer that still yields a clean EOF) is rejected instead of being parsed and emitted as a complete object — silent data loss otherwise.

• Length (verify_length, default true) — counts the bytes read and compares them against the Content-Length S3 advertised. A mismatch (short or long) fails the page with FaucetError::Source. Cheap; runs over the bytes that are read anyway. If the store reports no Content-Length, the check is skipped with a debug log rather than failing.
• Checksum (verify_checksum, default false) — sets ChecksumMode::Enabled on each GetObject and verifies the body against the strongest checksum S3 has stored for the object, in order: SHA-256, CRC-64-NVME, CRC-32C, CRC-32, then the ETag as MD5 for a non-multipart upload. Costs a hash over the full body. If the object advertises no checksum this source can verify, a one-shot warning is logged and the length check still applies.

Both checks operate on the stored bytes (below decompression), so they work unchanged for compression-enabled .gz / .zst objects.

pipeline:
  source:
    type: s3
    config:
      bucket: my-bucket
      prefix: events/
      verify_checksum: true   # length check is already on by default

Authentication

This source uses the standard AWS SDK credential chain — there are no credential fields in the config. Credentials resolve automatically, in order:

1. Environment variables — AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN.
2. Shared AWS config files — ~/.aws/credentials, ~/.aws/config (honours AWS_PROFILE).
3. IAM roles — EC2 instance metadata, ECS task roles, or Lambda execution roles.

For MinIO / LocalStack, set endpoint_url and supply the service's access/secret keys via the same environment variables.

Examples

JSON array files from MinIO (S3-compatible)

version: 1
pipeline:
  source:
    type: s3
    config:
      bucket: local-bucket
      prefix: exports/
      region: us-east-1
      endpoint_url: http://localhost:9000
      file_format: json_array
  sink:
    type: jsonl
    config:
      path: ./exports.jsonl

Raw text files into a document store

version: 1
pipeline:
  source:
    type: s3
    config:
      bucket: documents-bucket
      prefix: reports/
      file_format: raw_text
      max_objects: 50
  sink:
    type: mongodb
    config:
      connection_url: mongodb://localhost:27017
      database: lake
      collection: reports

Each record is {"key": "reports/q1.txt", "content": "…"}.

Compressed JSONL with auto codec detection

version: 1
pipeline:
  source:
    type: s3
    config:
      bucket: my-data-lake
      prefix: logs/2026/
      file_format: json_lines
      concurrency: 32
      compression: auto    # .gz → gzip, .zst → zstd, otherwise none
  sink:
    type: stdout
    config:
      format: jsonl

auto resolves the codec per object key, so a prefix holding a mix of .jsonl, .jsonl.gz, and .jsonl.zst objects all read correctly in one run. Requires building with the compression feature.

One page per object for a load-job sink

version: 1
pipeline:
  source:
    type: s3
    config:
      bucket: my-data-lake
      prefix: staging/
      file_format: json_lines
      batch_size: 0      # one StreamPage per S3 object
  sink:
    type: bigquery
    config:
      project_id: my-project
      dataset_id: analytics
      table_id: events

Streaming & batching

S3Source implements Source::stream_pages: it lists matching objects, then streams their records into StreamPages without buffering the whole scan. The per-page record count comes from the config batch_size field (the trait-level batch_size argument is ignored so a pipeline hint can't silently override an explicit config value).

Behaviour by format:

• json_lines — the object body is decoded line-by-line via tokio::io::AsyncBufReadExt::lines, so client-side memory stays bounded at O(batch_size) lines regardless of file or scan size. Records flow across object boundaries — a single page may carry lines drawn from multiple objects.
• raw_text — each object contributes exactly one {key, content} record. The whole body is held in one String (the record is the file), then accumulated into the same batch_size buffer.
• json_array — the array can only be validated once the closing ] is observed, so each object is buffered fully, then its elements are chunked into pages of batch_size. This bounds page size but not peak per-object memory.

batch_size = 0 is the "no batching" sentinel: every emitted StreamPage corresponds to exactly one S3 object — no within-object chunking and no cross-object accumulation. Use it for small lookup files, or for downstream sinks (SQL COPY, BigQuery load jobs, Snowflake stage uploads) that prefer one large request per file to many small ones.

Memory ceiling — raw_text / json_array. Both formats hold one whole decoded object in memory at a time (inherent: raw_text's record is the file, and a JSON array isn't valid until its closing ]). Because objects are fetched concurrently, peak memory is roughly concurrency × (largest object's decoded size), not batch_size. For large raw_text / json_array objects, lower concurrency to cap peak memory, or re-emit the data as json_lines upstream so it streams line-by-line.

The S3 source has no incremental-replication mode today, so every emitted page carries bookmark: None. It does not implement resume/state, effectively-once, write modes, or a dead-letter queue (those are sink- or CDC-source-specific capabilities).

Compression

Behind the crate-local compression Cargo feature. Adds the compression config field with values none, gzip, zstd, or auto (the default — detects .gz / .zst from the object key).

type: s3
config:
  bucket: my-data-lake
  prefix: logs/
  compression: auto    # or 'gzip' | 'zstd' | 'none'

The codec resolves per object key at read time, so a single source can read a mix of compressed and uncompressed objects in one run. When an explicit codec disagrees with the key suffix, a one-time warning is logged. Decompression is streamed straight into the line reader — there is no separate raw + decompressed copy held at once.

Config loading

use faucet_core::config::{load_json, load_env_file};
use faucet_source_s3::S3SourceConfig;

# fn example() -> Result<(), Box<dyn std::error::Error>> {
let config: S3SourceConfig = load_json("config.json")?;
let config: S3SourceConfig = load_env_file(".env", "S3_SOURCE")?;
# Ok(()) }

Example .env:

S3_SOURCE_BUCKET=my-data-lake
S3_SOURCE_PREFIX=raw/events/
S3_SOURCE_REGION=us-east-1
S3_SOURCE_CONCURRENCY=10

Schema introspection

faucet schema source s3

Programmatically:

use faucet_core::Source;
# async fn example(source: faucet_source_s3::S3Source) -> Result<(), Box<dyn std::error::Error>> {
let schema = source.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);
# Ok(()) }

Library usage

use faucet_source_s3::{S3Source, S3SourceConfig, S3FileFormat};
use faucet_core::Source;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = S3SourceConfig::new("my-data-bucket")
        .prefix("exports/2026/")
        .region("us-west-2")
        .file_format(S3FileFormat::JsonLines)
        .concurrency(20)
        .with_batch_size(5000);

    let source = S3Source::new(config).await?;
    let records = source.fetch_all().await?;
    println!("Read {} records from S3", records.len());
    Ok(())
}

To run a full pipeline, wrap the source with any sink in faucet_core::Pipeline or drive faucet_core::run_stream for bounded-memory streaming.

How it works

• Client reuse — the aws-sdk-s3 Client is built once in S3Source::new() (resolving region and endpoint_url) and reused for every list and get. No per-object client construction.
• Paginated listing — list_objects_v2 follows next_continuation_token until the response is no longer truncated, stopping early once max_objects is reached.
• Parallel reads — listed keys are read through futures::stream::buffer_unordered(concurrency), so up to concurrency GetObject calls are in flight at once.
• Streaming decode — json_lines / raw_text open each body as an AsyncBufRead and decode line-by-line; only json_array buffers a full object before parsing.

Throughput scales with concurrency and is ultimately bounded by S3 / network bandwidth — benchmark with your own object sizes and parallelism.

Dataset discovery

The source supports live introspection via Source::discover(): one ListObjectsV2 delimiter (/) listing under the configured prefix (bucket root when unset) enumerates the "directories" directly below it, returning one dataset descriptor per common prefix with:

• name — the full key prefix (e.g. raw/orders/), kind: prefix
• config_patch — { "prefix": "raw/orders/" }, ready to deep-merge over the connection config as a matrix row

When the listing returns no common prefixes but does return objects directly under the prefix (a "leaf directory"), each object becomes a descriptor instead (kind: object, config_patch: { "prefix": "<full key>" }), capped at the single listing page of 1000. schema and estimated_rows are never set — either would require reading or paging the whole listing. Discovery issues exactly one listing call and never recurses.

Lineage dataset URI

s3://<bucket> or s3://<bucket>/<prefix> — e.g. s3://my-bucket/data/2026/.

Feature flags

Enable the connector itself in the CLI/umbrella via the source-s3 feature.

Troubleshooting / FAQ

See also

• Connector reference · Compression cookbook
• Related crates: faucet-sink-s3 · faucet-source-gcs · faucet-source-parquet

Sharded execution (cluster Mode B)

Under faucet serve --cluster, a top-level shard: { count: N } block splits this source across cluster workers automatically — no connector config needed. Each worker reads the objects whose key hashes to its shard index (stable FNV-1a modulo count), so the partition is disjoint and complete: every object is read by exactly one worker, and the partition stays stable as new objects appear. Outside the cluster coordinator a run reads every object, unchanged.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.