S3 sink (Parquet / JSON Lines)

The S3 sink writes CDC events as Parquet (default) or JSON Lines files to S3-compatible object storage - AWS S3, MinIO, GCS via S3-interop, Azure Blob, R2, Wasabi, or local filesystem. Files are partitioned by table and UTC date in Hive style, with atomic multipart commits.

When to use S3

The S3 sink is the lakehouse-first path out of DeltaForge. CDC events land directly in queryable Parquet without a Kafka detour.

Real-world applications

Pros and cons

Configuration

sinks:
  - type: s3
    config:
      id: orders-lake
      bucket: my-data-lake
      prefix: cdc/orders
      region: us-east-1
      access_key_id: ${S3_ACCESS_KEY}
      secret_access_key: ${S3_SECRET_KEY}
      format: parquet
      compression: snappy
      file_roll:
        max_bytes: 268435456
        max_events: 1000000
        max_age_secs: 300
        idle_age_secs: 600
      send_timeout_secs: 60
      required: true

File rolling defaults

A writer rolls when any threshold fires.

A background task sweeps every ~5 seconds, so max_age_secs and idle_age_secs fire even when the source is fully idle — you don’t need new events arriving to roll an aged or idle writer. This bounds how long a partition’s tail data stays buffered and uncommitted after a burst.

Compression options

File layout

{bucket}/{prefix}/table={table}/year={Y}/month={MM}/day={DD}/{ulid}.{ext}

• ULID is monotonic-by-time and sortable; generated at writer creation
• Extension is parquet, jsonl, jsonl.gz, or jsonl.zst
• Partition path is Hive-style for compatibility with every analytics engine

Example:

my-data-lake/cdc/orders/table=orders/year=2026/month=05/day=19/01H8M3...ABC.parquet

Schema and column mapping

Parquet files have a flat envelope schema:

User-data column types are derived from source DDL via the same TypeConversionOpts used by Avro. Key choices:

The flat layout (before_id, after_id) trades nesting for usability — every analytics engine handles flat columns out of the box, and predicate pushdown stays efficient.

Delivery semantics

Quick summary (full details in Guarantees & Correctness):

Exactly-once at the event level requires Iceberg (atomic snapshot commits) — Phase 2.

Production prerequisite: bucket lifecycle policy

Every production S3 sink deployment must configure this lifecycle rule:

{
  "Rules": [
    {
      "ID": "abort-abandoned-multiparts",
      "Status": "Enabled",
      "Filter": {},
      "AbortIncompleteMultipartUpload": { "DaysAfterInitiation": 1 }
    }
  ]
}

DeltaForge does not track multipart upload IDs externally — if the process crashes mid-upload, the partial parts linger on S3 until lifecycle policy expires them. Without this rule, you accumulate S3 storage cost for every aborted batch.

All major S3-compatible providers (AWS, MinIO, Wasabi, Backblaze B2, GCS S3-interop, R2) support AbortIncompleteMultipartUpload.

Examples

MinIO for development

sinks:
  - type: s3
    config:
      id: dev-minio
      bucket: deltaforge-dev
      prefix: cdc
      endpoint: "http://localhost:9000"
      region: us-east-1
      access_key_id: minioadmin
      secret_access_key: minioadmin
      virtual_hosted_style: false
      format: parquet
      compression: snappy

AWS S3 with IAM instance role

sinks:
  - type: s3
    config:
      id: prod-lake
      bucket: my-prod-lake
      prefix: cdc/orders
      region: us-east-1
      # Omit access keys — falls through to the default credential chain
      # (IAM instance role, IRSA on EKS, env vars, ~/.aws/credentials, ...)
      virtual_hosted_style: true
      format: parquet
      compression: snappy
      file_roll:
        max_bytes: 268435456    # 256 MiB
        max_events: 2000000
        max_age_secs: 600

JSON Lines for audit logs

sinks:
  - type: s3
    config:
      id: audit-bucket
      bucket: cdc-audit
      prefix: 2026
      region: us-west-2
      format: jsonl
      compression: gzip
      file_roll:
        max_bytes: 16777216     # 16 MiB — keep files small for `aws s3 cp | jq`
        max_age_secs: 60         # 1 min — low latency to readable

Local filesystem for testing

sinks:
  - type: s3
    config:
      id: local-sink
      bucket: /var/lib/deltaforge/lake
      local: true
      format: parquet

Multi-sink: Kafka + S3 fanout

sinks:
  # Hot path: real-time consumers
  - type: kafka
    config:
      id: realtime-kafka
      brokers: ${KAFKA_BROKERS}
      topic: cdc.orders
      required: true
  # Cold path: analytics on object storage
  - type: s3
    config:
      id: lake-s3
      bucket: my-lake
      prefix: cdc/orders
      region: us-east-1
      format: parquet
      compression: snappy
      required: false   # Don't gate Kafka's checkpoint on S3 outages

With commit_policy.mode: required, the Kafka sink gates checkpoints — if S3 fails, the Kafka sink’s checkpoint still advances and the source doesn’t replay for Kafka.

What required: false actually does (and doesn’t)

Short version:

• Within one batch, sinks run concurrently — a slow S3 does not slow the Kafka call for that batch.
• Across batches, the coordinator waits for all sink futures before pulling the next batch, so slow S3 still backpressures the source.
• A failed S3 batch is not retried in the same session. The S3 sink’s checkpoint stays at its prior position; events are re-delivered only on pipeline restart (the source’s replay-from-MIN mechanism picks them up).

required: false buys:

• Kafka’s checkpoint can advance even when S3 fails for this batch.
• Pipeline doesn’t fail-loop on a persistent S3 outage.

required: false doesn’t buy:

• Independent throughput between sinks — slow S3 still slows the source.
• In-flight retry for S3 — events between failure and restart are only delivered on restart.

Memory bound under slow S3: max_bytes × active_partitions (default 256 MiB × N). The internal BufWriter blocks on slow uploads, propagating backpressure; no unbounded growth.

For truly independent hot/cold paths, run two pipelines (same source DSN, different sinks) or use a Kappa-style architecture (source → Kafka pipeline → second DeltaForge pipeline → S3) so Kafka’s retention is the natural retry buffer.

See Guarantees & Correctness — Required vs. optional sinks for the full failure-mode table, retry-layer breakdown, and S3 sink atomicity properties.

Querying output

DuckDB (zero-setup)

INSTALL httpfs;
LOAD httpfs;
SET s3_region='us-east-1';
SET s3_access_key_id='...';
SET s3_secret_access_key='...';

SELECT op, after_id, after_amount, op_ts
FROM read_parquet('s3://my-lake/cdc/orders/table=orders/year=2026/month=05/**/*.parquet')
WHERE op = 'c' AND after_amount > 100.00
ORDER BY op_ts DESC LIMIT 100;

AWS Athena (Hive partition discovery)

CREATE EXTERNAL TABLE orders_cdc (
  op string,
  op_ts timestamp,
  source_table string,
  before_id bigint,
  after_id bigint,
  after_amount decimal(12, 2),
  after_paid boolean
)
PARTITIONED BY (year int, month int, day int)
STORED AS PARQUET
LOCATION 's3://my-lake/cdc/orders/table=orders/';

MSCK REPAIR TABLE orders_cdc;

Trino / Spark

Standard Parquet readers; the Hive partition discovery works without configuration.

Metrics

The sink emits Prometheus metrics under deltaforge_sink_s3_*:

Limitations and roadmap

Phase 1 (current):

• Parquet + JSON Lines formats
• S3, MinIO, GCS (via S3 interop), Azure, local FS
• DDL-derived schemas with native Decimal128
• File-level atomicity, multipart upload, ULID file names
• Hive partitioning by table + UTC date

Phase 2 (planned):

• Iceberg table format — exactly-once event-level via atomic snapshot commits, schema evolution, time travel
• Hour-granularity partitioning — opt-in partition_by: [day, hour]
• Delta Lake / Hudi — alternative table formats

Phase 2 shipped:

• ✅ Per-row DLQ — slow-path per-event retry isolates bad rows
• ✅ send_timeout_secs per-batch timeout (sink-internal)
• ✅ Streaming JSONL writer with gzip / zstd via async-compression — memory bounded by BufWriter chunk size (~8 MiB) instead of full file size
• ✅ List and Map<Utf8, Utf8> columns as native Arrow nested types (PG arrays and hstore)
• ✅ Coordinator-level sink_batch_deadline_secs — defense-in-depth outer timeout (see Guarantees)

Not on the roadmap:

• CSV / TSV output (low signal-to-noise vs JSONL)
• Avro Object Container Files (niche; Avro outside of Kafka is uncommon)