SAA Lab 04 - Land a stream in S3 as Parquet

Scaffold: 4/5. The two-shard Kinesis data stream, the bucket, the Glue database and table with its projection rules, the Athena workgroup, the IAM roles and the four scripts are all built and wired. What is missing is the Firehose destination: the conversion, the buffering and the prefix.

The scenario

A producer writes about 200 JSON readings a second into a Kinesis data stream: a reading id, a sensor id, a timestamp, a temperature, a battery percentage. The readings have to land in S3 and be queryable by day in Athena, and almost every query names two or three of the five columns.

Firehose converts JSON to Parquet or ORC in flight, and it needs three things to do it: a deserialiser to read the incoming JSON, a schema, and a serialiser to write the columnar file. The schema comes from an AWS Glue Data Catalog table and from nowhere else, and the same table is what Athena reads.

The requirement

Three edits, all inside ExtendedS3DestinationConfiguration in src/template.yaml. The TODO block in that resource spells each one out.

  1. The conversion. DataFormatConversionConfiguration with an OpenXJsonSerDe deserialiser, a ParquetSerDe serialiser compressed with Snappy, and a SchemaConfiguration naming the Glue database and the readings table. CompressionFormat stays UNCOMPRESSED, because the serialiser is what compresses here.
  2. The buffering. SizeInMBs: 128 and IntervalInSeconds: 900. Turning conversion on moves the default size to 128 MiB and makes 64 the lowest value the API accepts, so a stream that converts cannot be asked for small objects. At this volume one shard’s buffer holds roughly 27 MB after fifteen minutes, so the size condition never fires and the interval decides every flush.
  3. The prefix. readings/dt=!{timestamp:yyyy-MM-dd}/, an ErrorOutputPrefix carrying !{firehose:error-output-type}, and CustomTimeZone: Australia/Perth. The default prefix is yyyy/MM/dd/HH, which has no key=value in any segment, so Athena sees no partitions in it.

Run it

./scripts/deploy.sh                 # stream, bucket, Glue table, Firehose; ~4 minutes
./scripts/produce.py --minutes 35   # ~200 records a second
./scripts/measure.py                # object count, bytes scanned, engine time
./scripts/set-buffer.py 60          # UpdateDestination, interval only
./scripts/produce.py --minutes 35
./scripts/measure.py
./scripts/teardown.sh

Give the first producer run longer than one buffer interval, or the bucket is still empty when you go looking. At 900 seconds the first objects appear about fifteen minutes in, two of them, one per shard.

measure.py records each run in .measurements.json beside the lab and prints every run it has recorded, so the second invocation shows both configurations on the same screen. --forget clears them.

Run it three times before you believe any of it. Engine time moves around with what else the account is doing, and the difference you are after is a factor, not a millisecond count.

Then break it on purpose

Take CustomTimeZone out and redeploy in the Perth morning. Everything still works, the objects are still Parquet, and the first eight hours of the day land in yesterday’s partition, where a query for today will not find them.

Then send one record whose battery_pct is a quoted string. It does not fail the stream; it lands under errors/format-conversion-failed/ and is missing from the table, which is quieter and worse.

Reveal the solution

SRC=solution ./scripts/deploy.sh && ./scripts/produce.py --minutes 35 && ./scripts/measure.py

Cost

Two Kinesis shard-hours per hour the stack exists, Firehose ingestion on a couple of gigabytes, a few Athena queries against a 10 MB per-query minimum, and a little S3. Small, and the shards bill whether anything is writing to them or not. Tear down when you are finished; the lab reaper is a backstop, not a plan.

Untested

Every script here is syntax-checked and both templates parse, but this bundle has not been deployed against a real AWS account end to end. Treat the first run as the shakedown.