Day 3 - Kinesis Data Firehose: Delivery Streams and Loading

📌 Key Takeaways

• Firehose (now named Amazon Data Firehose) is a fully managed, near-real-time loading pipe that lands a stream in a destination with no shards and no consumer code.
• The buffer is flushed when either the size or the time threshold is reached first. A larger size means bigger files and better efficiency; a shorter interval means lower latency and smaller files.
• Before loading, you can insert a Lambda transformation and a Parquet/ORC format conversion (which references a schema in the Glue catalog).
• Destinations are built in: S3, Redshift (COPY via S3), OpenSearch, Splunk, HTTP endpoints, and so on.
• Firehose itself offers no retention or replay. Records that fail transformation or delivery are routed to an S3 error location, so always monitor that path.

KDS vs Firehose: The Most Important Distinction

The Kinesis Data Streams we saw yesterday is powerful but labor-intensive. You calculate the shard count, write consumer code (KCL/Lambda), manage checkpoints, and even implement yourself how the data should be laid out in S3. Often what you really want is "just load my streaming data into S3 or Redshift for me." Amazon Kinesis Data Firehose (now named Amazon Data Firehose) answers that need.

Firehose in one line is a "fully managed delivery pipe." When producers push data in, Firehose buffers it, optionally transforms, compresses, or converts its format, and loads it into the destination automatically. You do not worry about shards, consumer code, or scaling.

On DEA-C01, the difference between the two is the basis for judging scenario questions. The table below summarizes it.

💡 Related theory: To memorize it in one line: "If you need complex real-time processing, multiple consumers, or replay, use KDS; if you only need to load data into a destination, use Firehose." An even stronger combination is to chain the two. With KDS as the source and Firehose reading that stream to load into S3, one side can run real-time analytics with Lambda or KCL while the other side has Firehose automatically archive the raw data in the data lake. When an exam question asks for both "real-time analytics + raw data retention," this combination is often the answer.

Buffering: The Heart of Firehose

Firehose does not send records one by one. It collects them in a buffer and delivers them to the destination all at once. There are two buffer flush conditions, and delivery happens when whichever is reached first is met.

- Buffer size:     e.g. 5 MB    (1-128 MB for S3)
- Buffer interval: e.g. 300 sec (0-900 sec for S3, default 300 sec)

A larger buffer size produces bigger files (better downstream query efficiency) and better cost efficiency but adds latency, while a shorter buffer interval delivers faster but produces many small files. Tuning this trade-off is the core of Firehose design.

import boto3
firehose = boto3.client("firehose")
 
firehose.create_delivery_stream(
    DeliveryStreamName="events-to-s3",
    S3DestinationConfiguration={
        "RoleARN": "arn:aws:iam::123456789012:role/firehose-role",
        "BucketARN": "arn:aws:s3:::my-data-lake",
        "Prefix": "events/year=!{timestamp:yyyy}/month=!{timestamp:MM}/",
        "BufferingHints": {"SizeInMBs": 5, "IntervalInSeconds": 300},
        "CompressionFormat": "GZIP"
    }
)

The "small files problem" is a chronic disease of data lakes. Query engines such as Athena and Spark have per-file overhead, so a huge number of small files is much slower and more expensive than a few large ones. If you make the Firehose buffer size large enough (for example 128 MB), data lands as a small number of large files and downstream query performance improves. When an exam scenario says "Athena queries are slow + Firehose loading," think of increasing the buffer size.

Transformation: On-the-Fly Processing with Lambda

Before loading, Firehose can transform records with Lambda. Lambda receives the incoming records, cleans, filters, or reformats them, and returns them, and the result is what gets loaded into the destination. It is used to normalize logs to standard JSON, mask sensitive fields, or filter out bad records.

import base64, json
 
def handler(event, context):
    output = []
    for record in event["records"]:
        payload = json.loads(base64.b64decode(record["data"]))
        payload["ingested_at"] = context.aws_request_id   # example of processing
        transformed = json.dumps(payload) + "\n"           # add newline -> load line by line
        output.append({
            "recordId": record["recordId"],
            "result": "Ok",                  # Ok / Dropped / ProcessingFailed
            "data": base64.b64encode(transformed.encode()).decode()
        })
    return {"records": output}

Firehose can also convert JSON into the Parquet/ORC columnar format as a built-in feature (Data Format Conversion). It references the schema in the Glue Data Catalog. Loading in a columnar format makes Athena queries much faster and cheaper.

Loading by Destination

Firehose's strength is that its destinations are built in.

• S3: The most common destination. Loads into the raw zone of a data lake as GZIP/Parquet.
• Redshift: Internally writes to S3 first and then loads into Redshift with the COPY command. It is not a direct INSERT.
• OpenSearch: Log and search indexing.
• Splunk / third-party HTTP endpoints.

-- The form Firehose runs internally when loading into Redshift
COPY events FROM 's3://my-data-lake/redshift-staging/manifest'
IAM_ROLE 'arn:aws:iam::123456789012:role/redshift-copy'
FORMAT AS JSON 'auto';

The fact that Firehose's Redshift loading is "via S3, then COPY" is an exam point. Redshift is an MPP (massively parallel processing) data warehouse where single-row INSERTs are extremely inefficient, so bulk loading is always done by loading in parallel with COPY from S3. Firehose implements this best practice automatically under the hood. Even if COPY fails, the data remains in the intermediate S3 bucket it wrote first, so it can be loaded again.

Error Handling

If the transformation Lambda fails or loading to the destination fails, Firehose does not discard the data; it sends it to the designated S3 backup/error prefix. This lets you investigate and reprocess failed records later. Configuring this backup bucket is the key to operational stability. The exception is when the destination is S3 and the S3 delivery itself fails, because then there is nowhere to send the data (see "Delivery Guarantees and Failure Handling by Destination" below).

Three Sources, Two Names

There are three ways to put data into a Firehose stream. Which one you choose changes the limits and the behavior on failure.

• PutRecordBatch carries up to 500 records or 4 MiB per call, whichever is smaller. The limit for a single record is 1,000 KiB before base64 encoding.
• In 2024 the service name changed from Amazon Kinesis Data Firehose to Amazon Data Firehose. The documentation calls a "delivery stream" a "Firehose stream," but the API name (CreateDeliveryStream) is unchanged. Exam passages still use the old name.

Buffer Hints by Destination

The buffer range differs by destination. If you memorize it as a single "Firehose is 60 seconds," you will get it wrong.

• Setting the interval to 0 seconds delivers within a few seconds without buffering (zero buffering). In the past the minimum was 60 seconds, and the old claim that "Firehose has a minimum 60-second delay" comes from that.
• On an S3 destination, if you set the interval below 60 seconds, Firehose uses multipart upload to reduce latency, at the cost of higher S3 PUT request charges.
• Zero buffering cannot be used together with dynamic partitioning, and it does not apply to the buffer (60-900 seconds) on the S3 backup side either.
• The buffer hints are literally "hints." If delivery to the destination cannot keep up with ingestion, Firehose dynamically increases the buffer size to catch up.
• If you specify one of size and interval, you must specify the other as well.

⚠️ Pitfall: If you choose "reduce the buffer interval" as the remedy for "Athena queries are slow," it is the wrong direction. Reducing the interval makes files smaller and more numerous. The remedy for the small files problem is to increase the buffer size and also leave the interval generous. When ingestion volume is low, the time condition fills before the size condition is reached, so if the interval is short, increasing the size no matter how much does no good.

Limits and Return Contract of the Transformation Lambda

Firehose invokes Lambda synchronously with the batch of records collected in the buffer. So Lambda's synchronous invocation limits apply directly as constraints.

• For every record it receives, the response must return recordId, result, and data. result is one of three values.
  • Ok: Transformation succeeded. data goes to the destination.
  • Dropped: Intentionally discarded. This is not an error, and the record goes neither to the destination nor to the error folder.
  • ProcessingFailed: Transformation failed. The record goes to the error folder.
• If the transformation inflates records, the response can exceed 6 MB and the whole invocation can fail. If your function is an enrichment that adds many fields, reduce the transformation buffer.
• When writing to S3, Firehose concatenates records with no delimiter. To make JSON readable line by line, either append a newline in the transformation function (the example in the body above) or turn on the option to add a newline delimiter. If you miss this, Athena reads only the first record of the file.
• When transformation is enabled, you can keep a source record backup in a separate S3 path. It is the basis for going back to the originals if the transformation logic had a bug.

Constraints When You Enable Format Conversion

• The reason the buffer lower bound is 64 MB is that a tiny columnar file is meaningless. Keep in mind that if you turn on format conversion for a low-volume stream, data does not arrive until the time condition is reached.

Dynamic Partitioning: Splitting Folders by Record Content

The default prefix is YYYY/MM/dd/HH (UTC), and this is the time Firehose received the record. To split by the time the event actually occurred or by a field such as customer_id, use dynamic partitioning.

Prefix:
  events/customer_id=!{partitionKeyFromQuery:customer_id}/dt=!{partitionKeyFromQuery:dt}/
ErrorOutputPrefix:
  errors/!{firehose:error-output-type}/

Partition key extraction (inline parsing, jq expressions)
  customer_id : .customer_id
  dt          : .event_timestamp | strftime("%Y-%m-%d")

• There are two ways to extract keys. For JSON, use inline parsing (jq expressions); if the data is compressed, encrypted, or not JSON, a Lambda computes and returns the keys.
• A separate buffer is created for each partition. The number of active partitions that can be open at the same time in one stream is 500 by default (an increase can be requested), and when it overflows, the excess records fall into the error prefix.
• So you must not use a field with many distinct values, such as user_id, as a partition key. It exceeds the active partition limit, and even if it does not, files get split finely per partition. Values with few distinct kinds, such as date, Region, or tenant, are the right choice.
• With dynamic partitioning, delivery latency can grow up to 1.5 times the configured buffer time, and zero buffering cannot be used. An additional charge based on the amount of data processed also applies.
• If you write the prefix in key=value/ form (Hive style), Athena and Glue recognize it as partitions.

Delivery Guarantees and Failure Handling by Destination

"Firehose never loses data" is only half true. Where the data ends up differs by the kind of failure.

• When S3 is the destination and "S3 delivery itself fails," there is nowhere to go. If you wrongly change the IAM role permissions or bucket policy and a day passes, Direct PUT data disappears. This is why you set alarms on delivery metrics such as DeliveryToS3.Success and on the data freshness metric.
• Delivery is at-least-once. If a delivery request times out and later succeeds, it can overlap with the retry and produce duplicates (S3 destinations are an exception). On the Redshift and OpenSearch side, design to absorb duplicates by key.
• With KDS as the source, even if a destination outage drags on, you can hold out by extending the KDS retention period. This is also why the "real-time analytics + raw data retention" combination is advantageous in terms of durability.

Cost Structure and Comparison with Alternatives

• The base price is proportional to the amount of data ingested. Direct PUT and KDS sources round record size up to 5 KB units (no rounding for MSK sources), so if you send 200-byte records one at a time, you are billed tens of times the actual size. Batching several items into one record at the producer is the pricing optimization.
• Format conversion, dynamic partitioning, and delivery inside a VPC are optional features with extra charges depending on throughput. VPC delivery also has an hourly per-AZ charge while the stream is active.
• Otherwise there is no charge for the time a stream is up. Unlike KDS provisioned mode (billed by shard hours), if there is no ingestion there is no ingestion charge.

Passage Clue to Correct Answer Lookup Table

Tomorrow we compare Amazon MSK, the managed offering of Apache Kafka and the last piece of streaming ingestion, with Kinesis.

📖 Glossary

• Firehose stream (delivery stream): The basic unit of Firehose. It receives data, buffers it, and hands it to the destination you set.
• Buffer hints: The size and time settings that decide how much to collect before sending. Delivery happens at whichever condition is filled first.
• Near real-time: A level where data arrives as late as the buffer allows. It means a delay of seconds to minutes.
• Zero buffering: A setting that sets the buffer interval to 0 seconds so data is delivered within a few seconds without being collected.
• Direct PUT: A method where the producer puts records directly using the Firehose API.
• Record format conversion: A feature that converts incoming JSON to Parquet or ORC before loading. It uses the schema of a Glue table.
• Dynamic partitioning: A feature that splits the S3 folders (prefixes) by values inside the record when loading.
• Small files problem: The phenomenon where too many finely split files slow down the query engine.
• COPY: A command that loads files in S3 into Redshift in parallel. Firehose uses it internally when loading into Redshift.
• Manifest file: A file listing the S3 objects to be loaded (or that were skipped). It is used to manually reload failed data.