File & Object Storage Integration #

In modern data pipeline architecture, data flowing through Apache Kafka often has two main destinations: consumed directly for real-time needs, and stored into long-term storage for big data analytics or archiving. Cloud Object Storage like Amazon S3, Google Cloud Storage (GCS), or Azure Blob Storage is the most popular storage medium because of its very cheap costs, unlimited capacity, and very high durability. However, moving data from Kafka to object storage isn’t just writing plain text messages into a folder. We must think about how data is organized so it’s efficient when read by query engines (like AWS Athena or Snowflake), how to compress data, and how to overcome the small file problem that can damage storage metadata performance. This article will comprehensively dissect file and object storage integration strategies.

Main Uses of Cloud Object Storage (S3 / GCS) #

Before diving into technical aspects, let’s discuss why we need to connect Kafka with Cloud Object Storage using Kafka Connect:

1. Data Lake and Lakehouse Foundation #

Object storage acts as a central repository (landing zone) where all raw data from various Kafka topics is collected before further processing. This data can later be read in parallel by query engines like Amazon Athena, Presto, Trino, Databricks, Apache Spark, or Snowflake for business intelligence (BI) and machine learning (ML) needs.

2. Long-Term Data Archiving (Cold Storage Log Archiving) #

Kafka is designed to store stream data quickly and dynamically, not to store historical data for years (because SSD disk storage costs on Kafka brokers are very expensive). By streaming data to S3 or GCS, we can implement very cheap cold tier data storage policies and keep data for regulatory compliance (compliance audit trails) for years.

3. Disaster Recovery #

Storing copies of all Kafka events to external object storage provides an additional security layer. If our entire Kafka cluster experiences an unrecoverable total failure, we can still reconstruct historical data states by replaying data from S3.

The Small File Problem and Its Critical Impact #

One of the biggest challenges when writing data from streaming systems (like Kafka) to file-based storage systems (like S3) is the Small File Problem.

What Is the Small File Problem? #

If we write data to S3 every time a new message enters Kafka, or use too-short time intervals (for example every 5 seconds), we produce millions of very small files (for example ranging from 2 KB to 50 KB).

This problem has very bad operational impacts:

  • HTTP Request Cost Overhead: Cloud providers like AWS charge based on the number of HTTP PUT (when writing) and GET (when reading) requests. Writing millions of small files dramatically multiplies our monthly bill.
  • Destroyed Query Performance: When query engines like AWS Athena or Spark read data for analysis, the engine must open, read, and close hundreds of thousands of small files. Time wasted on HTTP network negotiation and file metadata processing is far larger than actual data reading time. Queries that should finish in 3 seconds can bloat to 30 minutes.
  • Metadata Engine Load: Distributed file systems (like HDFS) store file metadata in NameNode memory. Too many small files make the NameNode run out of memory (OutOfMemory).
flowchart TD
    subgraph BadApproach["Bad Scenario: Small Files (High Latency & High Cost)"]
        direction TB
        KafkaMentah1[("Kafka Topics")] -->|Every 5 seconds / 10 Records| S3Sink1["S3 Sink Task"]
        S3Sink1 -->|Write HTTP PUT| S3Bucket1[/"AWS S3 Bucket"\]
        S3Bucket1 --> F1["file1.json (5 KB)"]
        S3Bucket1 --> F2["file2.json (8 KB)"]
        S3Bucket1 --> F3["file3.json (4 KB)"]
        F1 & F2 & F3 -->|Slow Query| Athena1["AWS Athena Query Engine"]
    end
    
    subgraph GoodApproach["Good Scenario: Controlled Rotation (Low Cost & High Performance)"]
        direction TB
        KafkaMentah2[("Kafka Topics")] -->|Buffer: 100,000 Records / 15 Minutes| S3Sink2["S3 Sink Task"]
        S3Sink2 -->|Write HTTP PUT at Once| S3Bucket2[/"AWS S3 Bucket"\]
        S3Bucket2 --> LargeFile["consolidated_file.parquet (128 MB)"]
        LargeFile -->|Fast Query| Athena2["AWS Athena Query Engine"]
    end

Solution: Controlling File Rotation Parameters #

To avoid the small file problem, we must force the Kafka Connect Sink to buffer data in worker memory first and consolidate it into one large file (ideal size ranging from 64 MB to 256 MB) before sending to S3.

We can control file rotation using the following three parameters in the Sink Connector configuration:

  1. flush.size: The minimum number of data rows (records) that must be collected in memory before writing to S3. (Example: set to 100000 records).
  2. rotate.interval.ms: The maximum time duration (in milliseconds) before a file rotates based on the arrival time of the first data in that batch. (Example: 1200000 ms or 20 minutes).
  3. rotate.schedule.interval.ms: File rotation based on wall-clock time on the Connect worker machine. Very useful if we want files to rotate exactly at minute 00 every hour (for example at 13:00, 14:00) so partition folders stay clean.

The file rotation process triggers whichever parameter is reached first (whichever comes first).

Error Recovery Mechanisms When Writing to S3 (Fault Tolerance on Uploads) #

Writing large binary files to cloud object storage brings its own network reliability challenges. If the HTTP connection drops while a 100 MB file is 90% uploaded, re-uploading the file from scratch is a huge resource waste.

1. Leveraging S3 Multipart Uploads #

To guarantee upload resilience, the S3 Sink Connector automatically splits large files into small parts (usually a minimum of 5 MB) and uploads them in parallel using the S3 Multipart Upload feature.

  • The s3.part.size parameter defines the minimum size per upload segment (default: 26214400 bytes or 25 MB). If our file reaches 100 MB, Connect uploads it in 4 parallel parts.
  • If one segment fails to upload from a transient network timeout, the Connect worker only retries uploading that problematic segment, not the entire file.

2. Handling S3 Storage Leaks (Aborted Multipart Uploads) #

If a Connect task permanently crashes mid multi-segment upload, the partially uploaded file parts remain stored in S3 dangling. S3 keeps charging storage costs for these orphaned parts even though the file was never fully compiled.

  • Best Solution: We must configure S3 Lifecycle Rules on our S3 bucket to automatically delete incomplete multipart uploads after 7 days:
    {
  "Rules": [
    {
      "ID": "Delete Pending Multipart Uploads",
      "Status": "Enabled",
      "Filter": {},
      "AbortIncompleteMultipartUpload": {
        "DaysAfterInitiation": 7
      }
    }
  ]
}

Dynamic Time-Based Partitioning Strategies #

So query engines can scan data quickly without reading the entire S3 bucket contents (which could be petabyte-sized), we must divide data files into orderly partition folder structures. The best strategy is using Time-Based Partitioning.

The ideal folder structure compatible with Hive partition schemas is:

s3://my-data-lake-bucket/topics/orders/year=2026/month=06/day=08/hour=20/orders_offset_10200.parquet

Why Is EventTime Far Better than WallClockTime? #

When distributing data into time partition folders, we must choose the time extraction method used:

  • WallClockTime: Uses the local time when the Connect worker processes that data.
    • Problem: If consumer processing delays (consumer lag) happen for 5 hours, transaction data occurring at 13:00 is written to the 18:00 partition folder. This scrambles historical data analysis.
  • EventTime: Uses the original timestamp of when the event actually happened in the real world (usually extracted from a data payload column like transaction_timestamp or Kafka record metadata).
    • Advantage: Guarantees data always lands in the correct time partition folder according to the original event time, making historical replay queries easier.

Time Partition Configuration Parameters #

Here are the parameters we must configure to implement EventTime-based partitioning:

# Using the built-in Confluent TimeBasedPartitioner
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner

# Directory writing format properties in S3
path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH

# Time range duration per partition folder (1 hour = 3600000 ms)
partition.duration.ms=3600000

# Using the EventTime extractor based on a field inside the JSON/Avro payload
timestamp.extractor=RecordField
timestamp.field=created_at

# Timezone used for partition folder standardization
timezone=UTC
locale=id-ID

Advanced Partitioning Strategies #

Besides time-based partitioning, modern data lake architectures often require partitioning data by other business logic criteria to support multi-tenancy data isolation or sharper query performance.

1. Field Partitioner #

If we want to separate data files in S3 by specific columns (for example country_code or department), we can use the FieldPartitioner.

  • Use Case: Makes local queries easier to only sweep specific region data (e.g., /country=ID/ or /country=SG/).
  • Configuration Properties:
    partitioner.class=io.confluent.connect.storage.partitioner.FieldPartitioner
partition.field.name=country_code

2. Hybrid Partitioner (Custom Combination) #

Often we want to combine business and time separation simultaneously, for example separating folders by tenant ID, then inside it divided by year, month, and day folders: s3://data-lake/raw-zone/tenant_id=company_a/year=2026/month=06/day=08/file.parquet

To achieve this, we can chain Custom Partitioning configuration by combining directory writing properties:

partitioner.class=io.confluent.connect.storage.partitioner.FieldPartitioner
partition.field.name=tenant_id
# Enabling time-based sub-partitioning under the field folder
path.format=tenant_id=${tenant_id}/year=YYYY/month=MM/day=dd

Data Serialization Formats and Compression #

The physical file storage format in S3 greatly affects downstream analysis query performance.

1. Avoid Using Raw JSON #

JSON is a plain text format that wastes a lot of storage space and is slow for query engines to process because every line must be parsed individually.

2. Use Columnar Formats (Parquet or ORC) #

Apache Parquet is a columnar storage format highly optimized for big data analytics.

  • High Compression: Parquet stores data per column contiguously, so compression algorithms work very efficiently (because data in one column has the same type). File sizes can shrink up to 80% compared to JSON.
  • Column Projection (Pruning): If our query only analyzes total sales (SELECT SUM(total_price) FROM orders), engines like Athena only download the total_price column from S3 disk and completely ignore other columns (like customer_address or notes). This saves up to 95% network I/O and drastically cuts Athena costs.

3. Enable Additional Compression #

Always enable file-level compression. For the Parquet format, the built-in Snappy compression is the best industry standard choice because it offers an excellent balance between high compression ratios and very fast CPU decompression speeds.

Comprehensive AWS S3 Sink Connector Configuration Example #

Below is a complete JSON configuration file example for deploying a safe Amazon S3 Sink Connector, using Parquet format with Snappy compression, EventTime-based partitioning, and optimal file rotation to avoid the small file problem:

{
  "name": "s3-parquet-sales-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "3",
    "topics": "mysql-db-orders",
    "s3.region": "ap-southeast-1",
    "s3.bucket.name": "badricreativetech-data-lake",
    "topics.dir": "raw-zone",
    
    "//": "--- Data Format and Compression Configuration ---",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",
    
    "//": "--- Small File Problem Prevention via Rotation ---",
    "//": "Write to S3 only after 100,000 records are collected in memory",
    "flush.size": "100000",
    "//": "Or rotate files periodically every 20 minutes (1200000 ms)",
    "rotate.interval.ms": "1200000",
    
    "//": "--- EventTime-Based Time Partition Configuration ---",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "partition.duration.ms": "3600000",
    "timezone": "Asia/Jakarta",
    "locale": "id-ID",
    
    "//": "Extracting partition time from the created_at column in the record payload",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "created_at",
    
    "//": "--- Avro Converter Integrated with Schema Registry ---",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "schema.compatibility": "NONE",
    
    "//": "--- S3 Security and Access ---",
    "//": "Using the Connect worker instance IAM Role (security recommendation)",
    "s3.credentials.provider.class": "com.amazonaws.auth.DefaultAWSCredentialsProviderChain"
  }
}

Summary #

  • Cloud Storage Ingestion — Cloud Object Storage (like S3 or GCS) is the ideal landing zone for long-term archiving and the foundation for building Data Lake / Lakehouse ecosystems.
  • Small File Problem — Repeatedly writing small files burdens read performance and multiplies HTTP request costs. Use the flush.size and rotate.interval.ms parameters to consolidate files.
  • Time-Based Partitioning — Time-based folder structures limit the amount of data scanned when queries run (partition pruning), saving query costs in AWS Athena.
  • EventTime vs WallClockTime — Always use EventTime based on the original transaction column to ensure data enters accurate time partition folders even with processing delays.
  • Parquet format — The Parquet columnar binary format with Snappy compression is highly recommended because it saves disk space and significantly speeds up analytical queries.

← Previous: Database Integration & CDC Next: External System Integration →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact