How to Fix the Kafka Small File Problem in Amazon S3 & GCS

1. Introduction

When building real-time streaming architectures with Apache Kafka, streaming directly into cloud storage (like Amazon S3 or Google Cloud Storage) seems straightforward. However, high-throughput topics written with low batch timeouts quickly create millions of micro-files—often under 1 MB each.

This "Small File Problem" leads to two massive issues:

1. Exploding Cloud Costs: AWS charges per 1,000 PUT requests ($0.005 per 1,000 requests adds up fast when writing 100 files/second).

2. Abysmal Query Speeds: Query engines like AWS Athena, Trino, and Snowflake must open, read, and close thousands of separate file footers, slowing down analytical queries from seconds to minutes.

In this guide, we’ll explore how to configure Kafka Connect (S3 Sink Connector) and PySpark Structured Streaming to batch, compress, and optimize real-time data before it hits long-term storage.

2. Solution Strategy 1: Tuning Kafka Connect S3 Sink

If you are using Kafka Connect, the key is balancing Latency vs. File Size. You should target file sizes between 128 MB and 512 MB per object.

Modify your connect-s3-sink.json configuration with these three parameters:

{

"name": "s3-sink-telemetry-optimized",

"config": {

"connector.class": "io.confluent.connect.s3.S3SinkConnector",

"tasks.max": "4",

"topics": "telemetry-events",

"s3.bucket.name": "datatipss-analytics-lake",

"s3.part.size": "67108864",

"flush.size": "100000",

"rotate.interval.ms": "300000",

"storage.class": "io.confluent.connect.s3.storage.S3Storage",

"format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",

"parquet.compression": "SNAPPY",

"schema.compatibility": "NONE"

}

}

Why these settings matter:

• s3.part.size = 67108864 (64 MB): Sets the minimum upload buffer part size for multi-part uploads.

• flush.size = 100000: Forces a file write only after accumulating 100,000 records.

• rotate.interval.ms = 300000 (5 minutes): Guarantees files are committed at least every 5 minutes so low-volume topics don't leave data stuck in memory forever.

3. Solution Strategy 2: Dynamic Repartitioning in PySpark

If you consume Kafka with PySpark Structured Streaming, simply saving the stream using .writeStream without repartitioning will create one file per Kafka partition for every trigger interval.

To avoid thousands of micro-partitions, use .repartition() or configure Trigger Intervals combined with Snappy-compressed Parquet:

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, from_json

from pyspark.sql.types import StructType, StringType, TimestampType

spark = SparkSession.builder \

.appName("KafkaToS3BatchOptimizer") \

.config("spark.sql.parquet.compression.codec", "snappy") \

.getOrCreate()

# 1. Read from Kafka

kafka_stream = spark.readStream \

.format("kafka") \

.option("kafka.bootstrap.servers", "localhost:9092") \

.option("subscribe", "user-clicks") \

.option("startingOffsets", "latest") \

.load()

# 2. Process Data Payload

schema = StructType().add("user_id", StringType()).add("action", StringType())

processed_df = kafka_stream \

.selectExpr("CAST(value AS STRING) as json_payload") \

.select(from_json(col("json_payload"), schema).alias("data")) \

.select("data.*")

# 3. Write with Trigger Processing Time to enforce batching

query = processed_df.repartition(2) \

.writeStream \

.format("parquet") \

.option("path", "s3a://datatipss-data-lake/user-clicks/") \

.option("checkpointLocation", "s3a://datatipss-data-lake/checkpoints/user-clicks/") \

.trigger(processingTime="5 minutes") \

.start()

query.awaitTermination()

4. Production Best Practices: Compaction Jobs

Even with batching, spike traffic can occasionally produce sub-optimal file sizes. Implement an automated Compaction Job (using Delta Lake OPTIMIZE or PySpark) that runs overnight to rewrite micro-files into uniform 512 MB Parquet files.