Exam Room · Solutions Architect

Lab: Land a Stream in S3 as Parquet

· 0 min read

Cloud Architecture · part of The Exam Room

This is a lab in the Solutions Architect Associate hands-on track. The reference posts argue the decisions; this one stands a pipeline up and makes it behave. The full lab is in lab-saa-04-stream-to-parquet.zip.

If this is your first lab, do the one-time, once-per-account setup: run the zip’s preflight.sh to confirm your account is ready, then deploy the lab reaper, a standing backstop that auto-deletes any lab you forget to tear down after 24 hours. This one runs a two-shard Kinesis data stream, which bills per shard-hour for as long as it exists, so tear down when you are finished.

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. Roughly 300 bytes each, so around 60 KB a second, a little over 200 MB an hour. 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.

Athena will read that JSON as it stands. It will also read every byte of every column to answer a question about one of them. Parquet stores values by column and carries min and max statistics per page, so a query for one column loads one column and skips pages whose range cannot match. 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.

Firehose flushes when either the buffer size or the buffer interval is reached, whichever comes first, and with record format conversion enabled SizeInMBs cannot be set below 64. 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. The interval therefore sets the object count, and the object count is what Athena’s query planner has to list and open.

Five stages from producer to Athena, with one error exit and two buffer settings compared A left-to-right pipeline of five boxes: a producer writing JSON records, a Kinesis data stream with two shards, Firehose buffering and converting, an S3 prefix of readings slash dt equals date, and Athena reading projected partitions. One arrow leaves the Firehose box downwards to a sixth box labelled errors slash format-conversion-failed, for records the Glue schema does not fit. A note under the Athena box says a date outside the projected range returns zero rows rather than an error. Below, two lanes compare buffer intervals over one hour. The upper lane, buffer interval 900 seconds, is drawn as eight separate blocks and labelled eight objects an hour. The lower lane, buffer interval 60 seconds, is drawn as one dense striped strip and labelled 120 objects an hour. Both lanes carry the same note: buffer size never fires, because 64 MiB is the floor. Producer PutRecords, JSON ~200 a second Kinesis stream 2 shards 1 MiB/s each Firehose buffer, then JSON to Parquet S3 readings/ dt=2027-04-19/ Athena projected partitions errors/format-conversion-failed/ records the Glue schema does not fit a date outside the projected range returns zero rows, not an error One hour in the bucket Buffer interval 900s size never fires: 64 MiB floor 8 objects Buffer interval 60s size never fires: 64 MiB floor 120 objects objects an hour = 3600 / interval, times the shard count

What you’re given

CloudFormation builds the Kinesis data stream in provisioned mode with two shards, an S3 bucket, a Glue database and table, an Athena workgroup with a results location, and the IAM roles. Two shards is more capacity than 200 records a second needs, since one shard takes 1 MiB per second or 1,000 records per second of writes. It is there because Firehose applies its buffering hints per shard, which is visible in the object count and in nothing else.

scripts/produce.py batches with PutRecords, 500 records to a call, using the sensor id as the partition key so a sensor’s readings stay ordered within a shard. Each record is one JSON document on one line.

The Glue table carries the schema Firehose reads and the projection rules Athena reads:

CREATE EXTERNAL TABLE readings (
  reading_id  string,
  sensor_id   string,
  reading_at  timestamp,
  celsius     double,
  battery_pct int )
PARTITIONED BY (dt string)
STORED AS PARQUET
LOCATION 's3://<bucket>/readings/'
TBLPROPERTIES (
  'projection.enabled'            = 'true',
  'projection.dt.type'            = 'date',
  'projection.dt.format'          = 'yyyy-MM-dd',
  'projection.dt.range'           = '2027-04-01,NOW',
  'projection.dt.interval'        = '1',
  'projection.dt.interval.unit'   = 'DAYS',
  'storage.location.template'     = 's3://<bucket>/readings/dt=${dt}/' )

dt appears in PARTITIONED BY and not in the column list. A partition key whose name matches a table column is an error, because the partition value lives in the path rather than in the file. With projection.enabled set, Athena computes the partition locations from these rules in memory instead of calling GetPartitions, so no crawler runs and no MSCK REPAIR TABLE is needed.

The Firehose stream in template.yaml reads the Kinesis stream and writes to the bucket. It ships with format conversion switched off, no prefix, and default buffering. That is the gap.

Your task

Three edits, all inside ExtendedS3DestinationConfiguration.

One: the conversion.

DataFormatConversionConfiguration:
  Enabled: true
  InputFormatConfiguration:
    Deserializer:
      OpenXJsonSerDe: {}
  OutputFormatConfiguration:
    Serializer:
      ParquetSerDe:
        Compression: SNAPPY
  SchemaConfiguration:
    DatabaseName: !Ref GlueDatabase
    TableName: !Ref GlueTable
    RoleARN: !GetAtt FirehoseRole.Arn
CompressionFormat: UNCOMPRESSED

The OpenX JSON SerDe is the one to reach for unless the input carries timestamps it cannot parse; it handles ISO-8601 with up to nine fractional digits, and epoch seconds or milliseconds. CompressionFormat has to be UNCOMPRESSED, which is also its default, because the serialiser is what compresses here. Parquet’s default is Snappy, in a framing format Hadoop and Athena both read. VersionId is left out, so Firehose uses LATEST and picks up column changes to the Glue table on its own.

Two: the buffering.

BufferingHints:
  SizeInMBs: 128
  IntervalInSeconds: 900

Without conversion the default size is 5 MiB. Turning conversion on moves the default to 128 and makes 64 the lowest value the API accepts, so a stream that converts cannot be asked for small objects. The interval range is 0 to 900 seconds with a default of 300; below 60 seconds Firehose switches to multipart upload, and the S3 PUT charges go up with it.

Three: the prefix.

Prefix: 'readings/dt=!{timestamp:yyyy-MM-dd}/'
ErrorOutputPrefix: 'errors/!{firehose:error-output-type}/dt=!{timestamp:yyyy-MM-dd}/'
CustomTimeZone: Australia/Perth

The default prefix is yyyy/MM/dd/HH, which is four path segments with no key=value in any of them, so Athena sees no partitions in it. Writing dt= into the prefix is what makes the path Hive-style and lines it up with storage.location.template. Two rules bite here. ErrorOutputPrefix cannot be null once Prefix contains an expression, and it has to include at least one !{firehose:error-output-type}, which in turn is not allowed in Prefix.

Timestamps in a prefix are evaluated in UTC unless you say otherwise, so a reading taken at 07:00 in Perth lands in yesterday’s partition, and a Perth analyst filtering dt = '2027-04-19' misses the first eight hours of their own day. Firehose supports Australia/Perth as a prefix time zone; set it and the partition boundary falls where the reader expects 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 lists the day’s prefix, then runs the same query under both configurations and reads DataScannedInBytes and EngineExecutionTimeInMillis back from GetQueryExecution. A run looks like this:

s3://lab-saa-04-.../readings/dt=2027-04-19/
  interval 900s   8 objects    4.1 MB total   avg 526 KB
  interval  60s 120 objects    4.4 MB total   avg  37 KB

SELECT sensor_id, max(celsius) FROM readings
  WHERE dt = '2027-04-19' GROUP BY sensor_id

  interval 900s   scanned 1.9 MB   engine 1,140 ms
  interval  60s   scanned 2.2 MB   engine 3,380 ms

Two columns out of five, so the scan is a fraction of what the same query costs against the JSON. The scan barely moves between the two configurations; the engine time does, because Athena lists the partition location and then opens, reads a footer from, and schedules every object it finds. Athena bills bytes scanned rounded up to the nearest megabyte with a 10 MB minimum per query, so at this size both runs bill the same minimum and only the clock separates 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.

If it fails

  • The objects are JSON, not Parquet. The stream deployed before the conversion block was added, or Enabled is false. aws firehose describe-delivery-stream prints the live configuration; conversion changes apply to data written after the update, so earlier objects stay as they were.
  • InvalidArgumentException on deploy, mentioning SizeInMBs. The template still carries a size under 64. Conversion and small objects are mutually exclusive by API.
  • Everything lands under errors/format-conversion-failed/. The Glue schema and the records disagree. Conversion failures and Lambda processing failures are the only two things Firehose sends to the error prefix; every other delivery problem it retries, and with a Kinesis source it retries for as long as the stream’s retention allows. So an object under that prefix is a schema problem, not a transient one. Nested JSON needs a STRUCT column rather than a string, and a field typed int in the catalogue but quoted in the JSON is an error under the Hive JSON SerDe. Fields absent from the schema are dropped rather than rejected, which is quieter and worth checking for.
  • Athena returns zero rows. Projection returns nothing rather than erroring when a date is outside projection.dt.range, or when storage.location.template does not match the real path. Compare the template against the output of aws s3 ls --recursive character by character; a missing trailing slash is enough.
  • HIVE_PARTITION_SCHEMA_MISMATCH. The Glue table changed after objects were written with the older column list. Drop and recreate the table, or query only the partitions written since.
  • The day’s partition is short by eight hours. CustomTimeZone is missing, so the prefix is being evaluated in UTC.
  • A record appears under the wrong day near midnight. Expected. Firehose evaluates the timestamp namespace against the arrival time of the oldest record in the object, not the event time in the record, so a buffer that spans midnight files all of its records under one date.

Reveal the solution

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

What you just learned

  • Firehose converts JSON to Parquet or ORC and nothing else. CSV or delimited text has to pass through a Lambda transform that emits JSON first, and with conversion enabled S3 is the only destination the stream can have.
  • The schema is a Glue Data Catalog table, and the same table is what Athena reads. One definition serves the writer and the reader, and leaving VersionId at LATEST means a column added in Glue reaches the files without touching the stream.
  • Conversion sets a floor of 64 MiB on the buffer size and moves the default to 128 MiB. Under that floor the interval is the only live control, and object count is 3600 divided by the interval, multiplied by the shard count.
  • The buffer interval sets two things that pull against each other: how soon a record can be queried, and how fast the query runs once it can be. Fifteen minutes of delay gave eight objects an hour; one minute gave 120 of them and a query that took roughly three times as long over the same data.
  • A prefix is what makes a partition. The default yyyy/MM/dd/HH is not Hive-style, and partition projection then needs a storage.location.template that matches it exactly. Writing dt= into the prefix instead is the shorter path, and a custom time zone is what keeps the day boundary where the people reading the data live.
  • Records that do not fit the schema are not lost. They land under ErrorOutputPrefix with !{firehose:error-output-type} in the path, which is the first place to look when a table is emptier than the stream.

Next

The rest of the Solutions Architect Associate lab track is in the README.

These posts are LLM-aided. Backbone, original writing, and structure by Craig. Research and editing by Craig + LLM. Proof-reading by Craig.