Streaming Playback Events: Backlog, Memory Failures, and Small Files

Do the arithmetic before touching configuration. 6,000 records/s x 60 s = 360,000 records per microbatch; at 2 KB that is roughly 720 MB of raw payload held per batch, and sessionization state plus dimension lookups typically multiply heap use several times over. Moving to 600 s multiplies batch size by ten, so it converts a struggling job into one that can never finish a batch. The container-killed message points at executor heap, not at a Kinesis limit.

Ordered plan:

1. Shorten the interval (10-20 s) so each batch fits comfortably in memory, and enable Glue auto-scaling with dynamic allocation.
2. Switch the reader to enhanced fan-out. The default 5 GetRecords calls/s/shard shared limit is a hard ceiling at 48 shards while a backlog exists, and any additional consumer competes for it.
3. Ensure parallelism of at least 48 concurrent shard consumers; 10 G.1X workers cannot hold that with a 60 s batch.
4. Before draining, raise stream retention above 24 h. A 5.5 h backlog is safe today but a larger spike would lose records rather than delay them, which violates the no-data-loss requirement.
5. Remove per-batch work: sample the debug write to about 1% (do not delete the prefix until you confirm nothing downstream reads it) and broadcast the dimension tables instead of joining per record.
6. Only re-run once IteratorAge falls monotonically for 30+ minutes; if it does not, stop and re-check parallelism before scaling workers further.

Track IteratorAgeMilliseconds and per-batch heap, not CPU, as the success signals.

The file count is structural: each microbatch writes one file per Spark task per partition, so 48+ tasks x 60 batches/hour produces tens of thousands of objects. At 28,000/hour you reach roughly 670,000 objects per day, and Athena query planning over that many files costs more time than the scan itself.

Preferred design: land into an Iceberg table partitioned by days(event_date), hour(event_time) and commit each microbatch with a MERGE on event_id. Deduplication then belongs to the table, so file layout is free to change. Run rewrite_data_files every hour over recent partitions with a target file size of 256 MB and a minimum input-files threshold, a weekly job for older partitions, and snapshot expiration so metadata does not grow without bound.

If plain Parquet must be retained: write each hour to a staging prefix, then perform an INSERT OVERWRITE of that hour partition from a de-duplicated read. That is atomic and idempotent, but it requires the hour to be closed, so enforce a watermark before the overwrite runs.

Either way, keep the raw immutable source (or Iceberg snapshots) so a bad compaction can be re-derived rather than repaired by hand, and alert on both file count per partition and average file size.

Extending allowed lateness to 48 hours is the wrong first move: it holds state for the whole window, and it is still wrong for the tail beyond 48 hours. The durable fix is to stop treating the streaming aggregate as authoritative.

1. Keep the short watermark for low-latency operational metrics and label those outputs provisional.
2. Make the lake table the source of truth. Late events still land with their event_time; an hourly reconciliation job recomputes the affected hour from the table and overwrites or MERGEs the corrected aggregate.
3. Partition on ingestion time (from Kinesis arrival, which is monotonic) rather than client event time, so a 48-hour-skewed event cannot write into a partition that was already finalized.
4. Publish a 'reconciled as of' column so consumers can tell a provisional value from a closed one.

Tell dashboard owners: financial metrics read the reconciled table at a two-hour lag; operational dashboards may read provisional numbers but must display the freshness timestamp. Document the 0.7% late tail so nobody reopens this as a bug.