Amazon Kinesis for Data Engineers

Data Streams shards, producers and consumers; Firehose delivery with dynamic partitioning and Parquet conversion; monitoring, scaling and comparisons.

1. The Kinesis Family

2. Kinesis Data Streams

import json, boto3

kinesis = boto3.client("kinesis")

def send(events: list[dict]) -> None:
    records = [{"Data": json.dumps(e).encode(), "PartitionKey": e["account_id"]} for e in events]
    for i in range(0, len(records), 500):  # PutRecords takes up to 500 records
        batch = records[i:i + 500]
        resp = kinesis.put_records(StreamName="txn-events", Records=batch)
        if resp["FailedRecordCount"]:  # partial failures do not raise
            retry = [r for r, res in zip(batch, resp["Records"]) if "ErrorCode" in res]
            kinesis.put_records(StreamName="txn-events", Records=retry)  # retry with backoff in real code

Starting positions: TRIM_HORIZON (oldest available), LATEST, AT_TIMESTAMP, or AT_SEQUENCE_NUMBER. Replay by starting a consumer from an earlier position.

3. Amazon Data Firehose

Firehose buffers incoming records (by size or time) and writes batches to the destination, with optional Lambda transformation, format conversion to Parquet/ORC (using a Glue table schema), and dynamic partitioning based on record content:

{
  "BucketARN": "arn:aws:s3:::acme-lake-prod",
  "RoleARN": "arn:aws:iam::111122223333:role/firehose-to-s3",
  "Prefix": "raw/txn/channel=!{partitionKeyFromQuery:channel}/dt=!{timestamp:yyyy-MM-dd}/",
  "ErrorOutputPrefix": "raw/txn_errors/!{firehose:error-output-type}/dt=!{timestamp:yyyy-MM-dd}/",
  "BufferingHints": { "SizeInMBs": 128, "IntervalInSeconds": 300 },
  "CompressionFormat": "UNCOMPRESSED",
  "DynamicPartitioningConfiguration": { "Enabled": true },
  "ProcessingConfiguration": {
    "Enabled": true,
    "Processors": [{
      "Type": "MetadataExtraction",
      "Parameters": [
        { "ParameterName": "MetadataExtractionQuery", "ParameterValue": "{channel: .channel}" },
        { "ParameterName": "JsonParsingEngine", "ParameterValue": "JQ-1.6" }
      ]
    }]
  },
  "DataFormatConversionConfiguration": {
    "Enabled": true,
    "InputFormatConfiguration": { "Deserializer": { "OpenXJsonSerDe": {} } },
    "OutputFormatConfiguration": { "Serializer": { "ParquetSerDe": { "Compression": "SNAPPY" } } },
    "SchemaConfiguration": { "DatabaseName": "raw", "TableName": "txn", "RoleARN": "arn:aws:iam::111122223333:role/firehose-to-s3" }
  }
}

• Larger buffers mean fewer, bigger files (good for queries) but more delay; 128 MB / 300 s is a common balance.
• With Parquet conversion, compression is handled by the serializer, so set S3 compression to UNCOMPRESSED.
• Firehose can write directly into Apache Iceberg tables, including routing to different tables and applying updates and deletes.

4. Monitoring & Scaling

5. Common Interview Questions