AWS S3 Connector

The S3 connector enables streaming data between Surgewave and Amazon S3 or S3-compatible storage services like MinIO and LocalStack.

Overview

• Source: Poll S3 buckets for new objects and stream their contents to Surgewave topics
• Sink: Write Surgewave records to S3 as objects with configurable partitioning

Use Cases:

• Data lake ingestion and archival
• Log aggregation and analysis
• Backup and disaster recovery
• ETL pipelines with cloud storage

Quick Start

S3 Source Connector

Read objects from S3 and produce to Surgewave:

{
  "name": "s3-source",
  "config": {
    "connector.class": "S3SourceConnector",
    "s3.bucket.name": "my-data-bucket",
    "s3.prefix": "incoming/",
    "s3.region": "us-east-1",
    "s3.access.key": "AKIAIOSFODNN7EXAMPLE",
    "s3.secret.key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "topic": "s3-events",
    "format": "jsonlines"
  }
}

S3 Sink Connector

Write Surgewave records to S3:

{
  "name": "s3-sink",
  "config": {
    "connector.class": "S3SinkConnector",
    "s3.bucket.name": "my-archive-bucket",
    "s3.prefix": "events/",
    "s3.region": "us-east-1",
    "s3.access.key": "AKIAIOSFODNN7EXAMPLE",
    "s3.secret.key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "topics": "events,logs",
    "format": "jsonlines",
    "flush.size": "1000",
    "rotate.interval.ms": "3600000"
  }
}

Configuration Reference

Connection Settings

Option	Type	Default	Description
s3.bucket.name	string	Required	S3 bucket name
s3.region	string	us-east-1	AWS region
s3.endpoint	string	-	Custom endpoint for S3-compatible services
s3.access.key	password	-	AWS access key ID
s3.secret.key	password	-	AWS secret access key
s3.prefix	string	-	Object key prefix filter

Source Settings

Option	Type	Default	Description
topic	string	Required	Destination Surgewave topic
format	string	json	Object format: json, jsonlines, csv, raw
poll.interval.ms	long	10000	Polling interval in milliseconds
mode	string	list	Mode: list (poll) or event (S3 events)
delete.after.read	bool	false	Delete objects after processing

Sink Settings

Option	Type	Default	Description
topics	string	Required	Source Surgewave topics (comma-separated)
format	string	json	Output format: json, jsonlines, parquet, avro
partitioner	string	default	Partitioning: default, time, field, custom
flush.size	int	1000	Records per file before flush
rotate.interval.ms	long	3600000	Max time before rotating file (1 hour)
timezone	string	UTC	Timezone for time-based partitioning

Authentication

AWS Credentials

The connector supports multiple authentication methods:

1. Explicit Credentials

{
  "s3.access.key": "AKIAIOSFODNN7EXAMPLE",
  "s3.secret.key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
}

2. Default Credential Chain

If credentials are not specified, the connector uses the AWS SDK default credential chain:

• Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
• Shared credentials file (~/.aws/credentials)
• IAM role for EC2/ECS/Lambda

3. IAM Role (Recommended for AWS)

{
  "s3.bucket.name": "my-bucket",
  "s3.region": "us-east-1"
}

S3-Compatible Services

MinIO

{
  "s3.bucket.name": "my-bucket",
  "s3.endpoint": "http://localhost:9000",
  "s3.access.key": "minioadmin",
  "s3.secret.key": "minioadmin"
}

LocalStack

{
  "s3.bucket.name": "my-bucket",
  "s3.endpoint": "http://localhost:4566",
  "s3.access.key": "test",
  "s3.secret.key": "test"
}

Format Support

Source Formats

Format	Description
json	JSON array or single object per file
jsonlines	Newline-delimited JSON (one record per line)
csv	CSV with header row (converted to JSON)
raw	Raw bytes as single record

Sink Formats

Format	Extension	Content-Type
json	.json	application/json
jsonlines	.jsonl	application/x-ndjson
parquet	.parquet	application/octet-stream
avro	.avro	application/avro

Partitioning Strategies

Default Partitioner

Objects organized by topic and partition:

s3://bucket/prefix/topic-name/0/1704067200000.json

Time Partitioner

Objects organized by timestamp:

s3://bucket/prefix/topic-name/2024/01/15/10/1704067200000.json

Field Partitioner

Objects organized by a field value from the record.

Examples

Data Lake Ingestion

Archive all events to S3 with hourly rotation:

{
  "name": "events-to-s3",
  "config": {
    "connector.class": "S3SinkConnector",
    "s3.bucket.name": "company-data-lake",
    "s3.prefix": "raw/events/",
    "s3.region": "us-west-2",
    "topics": "user-events,system-events",
    "format": "jsonlines",
    "partitioner": "time",
    "flush.size": "10000",
    "rotate.interval.ms": "3600000"
  }
}

Log Processing Pipeline

Read log files from S3 and process:

{
  "name": "logs-from-s3",
  "config": {
    "connector.class": "S3SourceConnector",
    "s3.bucket.name": "application-logs",
    "s3.prefix": "logs/prod/",
    "s3.region": "us-east-1",
    "topic": "log-events",
    "format": "jsonlines",
    "poll.interval.ms": "60000",
    "delete.after.read": "true"
  }
}

Development with MinIO

Local development setup:

# Start MinIO
docker run -p 9000:9000 -p 9001:9001 \
  -e "MINIO_ROOT_USER=minioadmin" \
  -e "MINIO_ROOT_PASSWORD=minioadmin" \
  minio/minio server /data --console-address ":9001"

# Create bucket
mc alias set local http://localhost:9000 minioadmin minioadmin
mc mb local/test-bucket

{
  "name": "minio-sink",
  "config": {
    "connector.class": "S3SinkConnector",
    "s3.bucket.name": "test-bucket",
    "s3.endpoint": "http://localhost:9000",
    "s3.access.key": "minioadmin",
    "s3.secret.key": "minioadmin",
    "topics": "test-topic",
    "format": "json"
  }
}

Troubleshooting

Common Issues

Access Denied

• Verify IAM permissions include s3:GetObject, s3:PutObject, s3:ListBucket
• Check bucket policy allows access from your IP/VPC
• Ensure credentials are correct and not expired

Slow Performance

• Increase flush.size for sink connectors
• Use jsonlines format instead of json for better streaming
• Consider regional bucket placement

Missing Objects

• Check s3.prefix filter matches your objects
• Verify poll.interval.ms is appropriate for your use case
• Enable delete.after.read only after confirming processing works

Monitoring

Check connector status:

surgewave connect status s3-source

View task details:

surgewave connect tasks list s3-source

See Also

• Azure Blob Storage Connector
• Google Cloud Storage Connector
• Custom Connectors