S3 Sink Node

The s3Sink node in ZephFlow enables you to write processed data to Amazon S3 (or S3-compatible storage) with built-in batching, date-partitioned keys, and multiple encoding formats.

Overview

The S3 sink node uploads pipeline output to an S3 bucket. It uses built-in batching that accumulates records and uploads them in date-partitioned files. You can also connect to S3-compatible endpoints such as MinIO by providing a custom endpoint override.

Key Features

S3 Integration: Write data directly to Amazon S3 or S3-compatible storage (e.g., MinIO)
Built-in Batching: Automatically accumulates records and flushes at a fixed batch size or time interval
Date-Partitioned Keys: Automatically organizes files by year/month/day partitions
Multiple Encoding Formats: Support for JSON, CSV, and Parquet output formats
Retry with Exponential Backoff: Automatic retries on upload failures with doubling delay
Custom Endpoint Support: Connect to any S3-compatible endpoint via endpoint override

Parameters

The most commonly set fields of the s3sink node config:

Field	Type	Description
`regionStr`	`String`	AWS region (e.g., `"us-east-1"`)
`bucketName`	`String`	S3 bucket name
`keyName`	`String`	Key prefix / folder path for uploaded objects
`encodingType`	`String`	Format for serializing output data
`credentialId`	`String`	ID of an access key / secret key credential in `jobContext.otherProperties` (omit for default AWS credentials)
`s3EndpointOverride`	`String`	Custom S3-compatible endpoint URL (omit for standard AWS S3)

See Config Object for all fields.

info

More details about encoding type support can be found here.

Config Object

The full config object for the S3 sink node (S3SinkDto.Config):

Field	Type	Required	Default	Description
`regionStr`	`String`	Yes	—	AWS region (e.g., `"us-east-1"`)
`bucketName`	`String`	Yes	—	S3 bucket name
`keyName`	`String`	Yes	—	Key prefix / folder path for uploaded objects
`encodingType`	`String`	Yes	—	Format for serializing output data. Supported values: `CSV`, `JSON_OBJECT`, `JSON_ARRAY`, `JSON_OBJECT_LINE`, `PARQUET`. See encoding constraints below
`credentialId`	`String`	No	`null`	Credential ID for authentication
`s3EndpointOverride`	`String`	No	`null`	Custom S3-compatible endpoint URL (e.g., MinIO)
`batching`	`boolean`	No	`true`	Enable batching mode
`batchSize`	`int`	No	`10000`	Number of records accumulated before flushing
`flushIntervalMillis`	`long`	No	`10000`	Maximum time in milliseconds between flushes
`avroSchema`	`Map<String, Object>`	No	`null`	Avro schema definition for the record structure
`compressionType`	`String`	No	`null`	Compression applied to each written object. Only `GZIP` is supported; it appends `.gz` to the file extension. Not allowed with `PARQUET`, which is compressed internally

Built-in Batching Behavior

The S3 sink operates in batching mode by default with the following internal defaults:

Setting	Default	Description
Batch size	10,000 records	Number of records accumulated before flushing
Flush interval	10,000ms (10s)	Maximum time between flushes regardless of batch size

These defaults can be overridden via the batchSize and flushIntervalMillis config fields.

Encoding Types

The following encoding types are supported:

Encoding Type	Description
`CSV`	Comma-separated values
`JSON_OBJECT`	Single JSON object. Not compatible with batching mode
`JSON_ARRAY`	Array of JSON objects
`JSON_OBJECT_LINE`	One JSON object per line (JSONL)
`PARQUET`	Apache Parquet format. Requires batching enabled and `avroSchema` provided

S3 Key Pattern

Files are date-partitioned using the current date:

{folder}/year=2024/month=01/day=15/{uuid}.{ext}

For example, with folder output/records and JSON_OBJECT_LINE encoding:

output/records/year=2024/month=01/day=15/a3f8b2c1-7d4e-4a9b-b5f6-1c2d3e4f5a6b.jsonl

Best Practices

Encoding Selection

JSON_OBJECT_LINE: Best general-purpose choice — one JSON object per line, widely supported by downstream tools
JSON_ARRAY: Use when downstream consumers expect a JSON array
CSV: Use for tabular data or when integrating with tools that prefer CSV

Security and Credentials

When no credential is provided, the S3 sink uses the default AWS credential chain (environment variables, IAM roles, etc.)
For explicit credentials, add a credential to jobContext.otherProperties where username is the access key ID and password is the secret access key, and reference it with credentialId
Prefer IAM roles over explicit credentials in production environments

S3-Compatible Endpoints

Use the s3EndpointOverride parameter to connect to MinIO, LocalStack, or other S3-compatible services
The region parameter is still required even when using a custom endpoint

Common Issues and Solutions

Credential or Permission Errors

Verify the IAM user or role has s3:PutObject permission on the target bucket and key prefix
When using explicit credentials, ensure the access key and secret key are correct
For S3-compatible endpoints, confirm the endpoint URL is reachable and the credentials match

Upload Failures

The S3 sink automatically retries failed uploads with exponential backoff (up to 3 retries with doubling delay). If uploads continue to fail, check:

Network connectivity to the S3 endpoint
Bucket existence and region configuration
Available disk space for temporary files (batching uses temp files before upload)

DAG Examples

Basic Usage

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "broker-address:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "transform"

  - id: "transform"
    commandName: "eval"
    config:
      expression: dict_merge($, dict(processed=true))
    outputs:
      - "sink"

  - id: "sink"
    commandName: "s3sink"
    config:
      regionStr: "us-east-1"            # AWS region
      bucketName: "my-data-bucket"      # S3 bucket name
      keyName: "output/records"         # Folder (key prefix)
      encodingType: "JSON_OBJECT_LINE"  # Encoding type
    outputs: []

With Explicit Credentials

jobContext:
  otherProperties:
    aws-cred:
      username: "AKIAIOSFODNN7EXAMPLE"                      # Access key ID
      password: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"  # Secret access key

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "s3sink"
    config:
      regionStr: "us-east-1"
      bucketName: "my-bucket"
      keyName: "data/output"
      encodingType: "JSON_OBJECT_LINE"
      credentialId: "aws-cred"
    outputs: []

Kafka-to-S3 Data Lake Pipeline

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "raw-events"
      groupId: "s3-archiver-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "transform"

  - id: "transform"
    commandName: "eval"
    config:
      expression: |
        dict_merge($, dict(
          event_date=epoch_to_ts_str($.timestamp, "yyyy-MM-dd"),
          processed=true
        ))
    outputs:
      - "sink"

  - id: "sink"
    commandName: "s3sink"
    config:
      regionStr: "us-east-1"
      bucketName: "data-lake-bucket"
      keyName: "events/raw"
      encodingType: "JSON_OBJECT_LINE"
    outputs: []

Log Archival with Filtering

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "application-logs"
      groupId: "log-archiver-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "filter"

  - id: "filter"
    commandName: "filter"
    config:
      expression: $.level == "ERROR" or $.level == "WARN"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "s3sink"
    config:
      regionStr: "us-west-2"
      bucketName: "log-archive-bucket"
      keyName: "logs/important"
      encodingType: "CSV"
    outputs: []

S3-Compatible Storage (MinIO)

jobContext:
  otherProperties:
    minio-cred:
      username: "minio-access-key"
      password: "minio-secret-key"

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "s3sink"
    config:
      regionStr: "us-east-1"
      bucketName: "my-minio-bucket"
      keyName: "output/data"
      encodingType: "JSON_OBJECT_LINE"
      credentialId: "minio-cred"
      s3EndpointOverride: "http://localhost:9000"  # MinIO endpoint
    outputs: []

Complete Pipeline

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "source"
    commandName: "kafkasource"
    config:
      broker: "kafka-broker:9092"
      topic: "input-topic"
      groupId: "processor-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "filter"

  - id: "filter"
    commandName: "filter"
    config:
      expression: $.status != null
    outputs:
      - "transform"

  - id: "transform"
    commandName: "eval"
    config:
      expression: dict_merge($, dict(processed_at=epoch_to_ts_str($.timestamp, "yyyy-MM-dd'T'HH:mm:ss")))
    outputs:
      - "validate"

  - id: "validate"
    commandName: "assertion"
    config:
      expression: $.processed_at != null
    outputs:
      - "sink"

  - id: "sink"
    commandName: "s3sink"
    config:
      regionStr: "us-east-1"
      bucketName: "output-bucket"
      keyName: "processed/records"
      encodingType: "JSON_OBJECT_LINE"
    outputs: []

kafkaSource: Read data from Kafka topics
kafkaSink: Write data to Kafka topics

Overview​

Key Features​

Parameters​

Config Object​

Built-in Batching Behavior​

Encoding Types​

S3 Key Pattern​

Best Practices​

Encoding Selection​

Security and Credentials​

S3-Compatible Endpoints​

Common Issues and Solutions​

Credential or Permission Errors​

Upload Failures​

DAG Examples​

Basic Usage​

With Explicit Credentials​

Kafka-to-S3 Data Lake Pipeline​

Log Archival with Filtering​

S3-Compatible Storage (MinIO)​

Complete Pipeline​

Related Nodes​