Ten Million Objects a Day: Paying The Small-Files Tax on a Kafka-to-S3 Pipeline

We had a pipeline that copied Kafka topics to S3 so analysts could query them with Athena. It worked, and the data showed up within about fifteen minutes. It also wrote on the order of ten million objects a day, most of them only a few kilobytes.

S3 bills mostly by bytes, but it also bills per request. At a few terabytes a day you would expect storage to dominate. For us, a big share of the bill was PUT requests, and Athena queries were slow because they opened millions of tiny files.

This post covers what we changed. The short version: object count is a design input, not an accident. If the write path partitions by a high-cardinality key, the number of objects grows with that key, and nothing downstream can fix it cheaply.

Where the small files came from

The original sink partitioned objects by values inside the message. A typical record looked like this (anonymised):

{ "op": "i", "tenant": "tenant-a", "entity": "users", "doc_id": "607f..." }

and was written under a path like s3://bucket/topic/tenant=tenant-a/entity=users/.

Path partitioning is a sensible idea for readers. For a streaming writer it has a cost, because every Kafka partition holds records for every tenant. On each flush, a writer that consumes one Kafka partition opens one file per distinct path value it has seen since the last flush.

objects per flush  ≈  kafka_partitions × distinct(tenant, entity) per partition
objects per day    ≈  that × flushes per day

Some illustrative numbers:

That is for one topic with one entity type. Add a few entity types and a few topics and you get the ten million a day we were paying for.

The objects were also tiny. With a few terabytes split across millions of objects, the average size is well under a megabyte. That hurts in three ways:

• Request cost. At the S3 Standard list price of about $0.005 per 1,000 PUTs, ten million PUTs is roughly $50 a day before you store a single byte. GETs from Athena add their own per-request cost.
• Query latency. Athena (Trino under the hood) plans splits per file. Millions of small files mean millions of splits, S3 LISTs and GETs, and most of the time goes on file overhead instead of scanning.
• Compression. Small JSON files compress badly, and there was no columnar format at all, so every query scanned whole records.

Goals

1. Fewer, larger objects. Ideally hundreds of megabytes each, and a count that no longer grows with the number of tenants.
2. Parquet with a schema, so Athena can prune columns and compress well.
3. Freshness stays at about fifteen minutes for the raw copy.
4. No new always-on cluster if we could avoid one.

The design: don't partition while writing, partition while compacting

The key decision was to split the pipeline in two.

                 write path (every 15 min)                          compaction (hourly)
Kafka ──► Kafka Connect S3 sink ──► s3://raw/topic/ingest_hour=H/... ──► Athena INSERT INTO ──► s3://lake/table/tenant=.../entity=.../dt=.../
           no tenant in the path        Parquet, one file per partition     ≤ 100 partitions per query
           SMT adds a schema            per flush

1. Write without data partitioning

The rule for the raw path: nothing derived from the payload. The only thing allowed in the prefix is the ingest hour, so the compaction job can select one hour cheaply. Time is low-cardinality: every record a writer flushes falls into one or two hours, so it adds no objects. The Kafka partition and start offset are already in each filename. Each Kafka partition gets one file per flush, whatever tenants it contains. Object count now depends only on the partition count and the flush schedule:

objects per day ≈ kafka_partitions × flushes per day  (+ extra files when size-based rotation kicks in)

In the 48 × 500 example above, that goes from millions to a few thousand. That is three orders of magnitude, from moving one decision out of the write path.

A sketch of the connector config:

connector.class=io.confluent.connect.s3.S3SinkConnector
topics=events.raw
format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
# Time-based on ingest (wall clock), hourly. No payload fields in the path.
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner
timestamp.extractor=Wallclock
path.format='ingest_hour'=YYYY-MM-dd-HH
partition.duration.ms=3600000
locale=en-US
parquet.codec=snappy

# Rotate on the wall clock so data lands within ~15 minutes, and on size so a busy
# partition still produces bounded files. rotate.schedule.interval.ms needs a timezone.
rotate.schedule.interval.ms=900000
timezone=UTC
flush.size=2000000

transforms=wrap
transforms.wrap.type=com.example.connect.WrapJsonWithSchema

2. Get Parquet out of schemaless JSON

Parquet needs a schema, and the topics carried plain JSON with no registry. The S3 sink's ParquetFormat will refuse schemaless records, and we didn't want to model every tenant's fields.

The fix was a small single message transform (SMT) that wraps each record in a fixed, tiny schema:

before:  {"op":"i","tenant":"tenant-a","entity":"users","doc_id":"607f..."}   (schemaless)

after:   struct {
           tenant: string     ← extracted, becomes the lake partition key
           entity: string     ← extracted, becomes the lake partition key
           msg:    string     ← the original JSON, untouched
         }

This gives Parquet a stable schema. The fields we partition on later are real columns, and the payload is preserved byte for byte, so downstream parsing stays the reader's choice. Athena can still reach into msg with json_extract_scalar when it needs to, and the tenant and entity columns make the compaction query cheap.

The transform itself is a few dozen lines: parse JSON, read two fields, build a Struct against a constant Schema, and return a new record. Because the schema never changes, there is nothing to evolve and nothing to register.

3. Compact hourly into a partitioned lake

Every hour a job runs Athena SQL that reads the previous hour's raw files and writes them into the lake table, now partitioned by tenant, entity and date. This is where the partitioning we gave up at write time comes back, and it happens once an hour on an hour's worth of data. That means each output partition gets one reasonably sized file instead of four tiny ones per flush per Kafka partition.

There is a limit to work around: an Athena CREATE TABLE AS SELECT or INSERT INTO can write to at most 100 partitions per query. One hour of our busiest topic touched more than 500 (tenant, entity) combinations, and a single statement fails with HIVE_TOO_MANY_OPEN_PARTITIONS.

The documented workaround is to split the work into batches of at most 100 partitions:

-- 1. Discover what this hour touched.
SELECT DISTINCT tenant, entity
FROM raw.events
WHERE ingest_hour = '2021-06-15-10';

-- 2. For each batch of ≤ 100 (tenant, entity) pairs, write that slice.
INSERT INTO lake.events
SELECT msg, tenant, entity, '2021-06-15' AS dt
FROM raw.events
WHERE ingest_hour = '2021-06-15-10'
  AND (tenant, entity) IN ( ('tenant-a','users'), ('tenant-a','events'), ... );

The orchestration is a short script: run the discovery query, chunk the result, submit INSERT INTO statements with bounded concurrency, and poll GetQueryExecution until each one finishes. Things that matter in practice:

• Idempotency. If a batch fails halfway, rerunning it must not duplicate rows. Either delete that hour's slice before rewriting it, or write the hour to a staging prefix and swap it in once all batches succeed.
• Concurrency limits. Athena caps concurrent DML queries per account. Fire 6 batches at once, not 60.
• Lateness. Kafka Connect writes on its own schedule, so the job for hour H runs a few minutes after H+1 starts, giving the last flush time to land.

What we tried first, and why each one lost

We tried more designs than the one we shipped. The failures were as useful as the result.

1. Keep path partitioning with a disk-buffering writer. A writer that stages per-path files on local disk before uploading handles high cardinality without blowing up memory. But it still writes one object per path per flush, so PUT count and small-file count stay the same, and it needed a fleet of memory- and disk-heavy nodes. It worked, cost more and didn't fix the problem.

2. Path partitioning in Kafka Connect. Same object count. On top of that, the S3 sink keeps an in-memory buffer for every open output partition until it rotates. With hundreds of tenants per Kafka partition and several partitions per task, the heap estimate for our largest topic reached hundreds of gigabytes. This is a well-known property of the connector, discussed at length in its issue tracker, not a tuning problem.

3. No partitioning at all. This was great for S3: about a 90% drop in PUTs and better compression. It was terrible for Athena, because every query scans every file for the time range. We had moved the cost from storage to queries.

4. No partitioning, then sort in Athena. Sorting by tenant would let Parquet min/max statistics skip row groups. But a global ORDER BY funnels everything through a final stage, and Athena failed with Query exhausted resources at this scale factor on inputs much smaller than one hour of our data.

5. No partitioning, then sort with Glue or a dedicated Spark cluster. This is technically the cleanest option. But a managed Spark job sized for this volume cost about as much as the savings, and running our own Spark cluster on Spot meant another pipeline to operate. It was rejected on total cost, not on correctness.

6. No partitioning, then partition (not sort) in Athena. This is the design we shipped. INSERT INTO with a partitioned target doesn't need a global sort, only a shuffle by partition key, which Athena handles comfortably. The 100-partition limit is annoying, but batching it is a few lines of code.

Results

We compared one day of the old layout against the new layout for the three largest streams:

The busiest stream improved least, which is what the formula predicts. It has the most Kafka partitions and the highest volume, so size-based rotation produces extra files per partition per flush. The other two are small enough that an hour of data fits in a handful of files per lake partition.

Other effects:

• PUT costs fell in proportion to object count.
• Storage fell too, because Parquet with Snappy compresses much better than small JSON objects.
• Athena queries got faster and cheaper together: fewer splits, column pruning on Parquet, and partition pruning on tenant, entity and dt.
• The writer got lighter. Kafka Connect no longer holds a buffer per tenant, so tasks need modest heaps and restarts are uneventful.

Takeaways

• Count objects, not just bytes. For streaming writes to object storage, partitions × path cardinality × flushes is the number to watch, and it compounds without warning.
• Partition where the data is at rest, not where it is in motion. A streaming writer sees every key in every partition. A batch job sees an hour at once and can lay out files properly.
• A tiny fixed schema is still a schema. Wrapping JSON in {tenant, entity, msg} was enough to unlock Parquet without modelling anyone's payload.
• Read the limits table before designing. Athena's 100-partition cap shaped the compaction job more than any performance concern.
• Write the rejected options down. Half the value of this project was the comparison table. It stopped the "why not just use Spark?" conversation before it started.