Kinesis Firehose Data Transformation with Lambda

1. The Role of Lambda in Firehose Data Transformation

In the previous lesson, we established that Firehose is a pipe for moving streaming data from a source to a destination. But what if the data isn't in the right format? What if you need to add information or filter out irrelevant records before they are stored?

This is where AWS Lambda comes in. Firehose can be configured to automatically invoke a Lambda function to process each batch of incoming data before it's delivered to the destination. This enables lightweight, real-time ETL without needing a separate fleet of servers or a complex processing framework.

Common use cases for this pattern include:

• Data Normalization: Converting data from different producers into a standard format (e.g., converting all timestamps to UTC).
• Data Enrichment: Adding new fields to records, such as looking up customer details based on a user ID.
• Data Filtering: Dropping records that are incomplete, irrelevant, or don't meet business rules.
• Format Conversion: Changing the data format, for example, from JSON to CSV or XML to JSON.

The overall architecture looks like this:

2. The Data Transformation Flow and Lambda Contract

To correctly implement a transformation, it's crucial to understand how Firehose and Lambda communicate. The process happens in batches:

1. Firehose buffers incoming records up to a configured size (e.g., 1 MB) or for a certain time (e.g., 60 seconds).
2. Firehose then synchronously invokes your Lambda function, passing the entire batch of records in the event payload.
3. Your Lambda function processes each record in the batch and prepares a transformed batch.
4. The Lambda function returns the transformed batch to Firehose.
5. Firehose proceeds to deliver the transformed data to the configured destination.

The exact structure of the data passed between Firehose and Lambda is a strict "contract" that you must follow.

The Lambda Event (Input)

Your Lambda function receives a JSON object that contains a list of records. Each record object has the following structure:

{
  "recordId": "49635...",
  "approximateArrivalTimestamp": 1678890156023,
  "data": "eyJU SUNLRVJfU1lNQk9MIjoiUVhaIiwiU0VDVE9SIjoiSEVBTFRIQ0FSR..."
}

• recordId: A unique identifier for the record within this specific invocation. You must return this ID in your response.
• data: The actual record data, which is Base64-encoded. Your function must first decode this data to work with it.

The Lambda Response (Output)

Your function must return a JSON object that also contains a list of records. Each record in your response must have this structure:

{
  "recordId": "49635...",
  "result": "Ok",
  "data": "MTY3ODg5MDE1NjAyMyxRWlosSEVBTFRIQ0FSRSwtMC4wNSw4NC41MQo="
}

• recordId: The same recordId from the corresponding input record. This is how Firehose matches the transformed record to the original.
• result: A status string indicating the outcome of the transformation for that record. It must be one of the following:
  • Ok: The record was transformed successfully. Firehose will deliver the new data.
  • Dropped: You have intentionally decided to filter out this record. Firehose will not deliver it to the destination.
  • ProcessingFailed: Your function failed to process this record. Firehose will retry invoking the Lambda function with this record. If it continues to fail, the record will be sent to your configured error-handling S3 bucket (which we will cover in the next lesson).
• data: The transformed record data, which you must Base64-encode before returning it.

{
  "recordId": "rec-001",
  "result": "Ok",
  "data": "eyJuYW1lIjoiYWJjIiwic3RhdHVzIjoicHJvY2Vzc2VkIn0="
}

3. Implementing a Transformation: A Practical Example

Now, let's walk through how to build and configure this in the AWS console. We'll use a practical example of processing stock ticker data. The goal is to:

1. Parse incoming JSON records.
2. Add a timestamp to each record.
3. Filter out any records that are not in the 'RETAIL' sector.
4. Convert the final records to a CSV format.

The following AWS blog post provides an excellent step-by-step guide with code.

Dissecting the Lambda Function Code

Let's break down the key logic from the Node.js code provided in the article.

// The handler function is the entry point
exports.handler = (event, context, callback) => {
    
    // The output is an array of transformed records
    const output = event.records.map((record) => {
        // 1. Decode the data from Base64 to a UTF8 string
        const entry = (new Buffer(record.data, 'base64')).toString('utf8');
        
        // ... (Code to parse the JSON string 'entry') ...
        let parsed_match = JSON.parse(match);

        // 2. Apply transformation logic
        if (parsed_match.SECTOR != 'RETAIL') {
            // This record is not wanted, so we drop it
            return {
                recordId: record.recordId,
                result: 'Dropped',
                data: record.data, // Return original data for dropped records
            };
        } else {
            // This record is wanted, so we transform it
            const result = `${new Date().getTime()},${parsed_match.TICKER_SYMBOL},...`;
            
            // 3. Encode the transformed data back to Base64
            const payload = (new Buffer(result, 'utf8')).toString('base64');
            
            // 4. Return the successful transformation
            return {
                recordId: record.recordId,
                result: 'Ok',
                data: payload,
            };
        }
    });

    // 5. Return the full batch of processed records to Firehose
    callback(null, { records: output });
};

This code perfectly illustrates the contract we discussed. It iterates through the records, decodes the data, applies business logic to decide whether to transform and keep (Ok) or filter out (Dropped), encodes the result, and wraps it all in the required response structure.

Conclusion

You've now learned how to embed custom processing logic directly into your Kinesis Firehose delivery stream using AWS Lambda. This is a fundamental pattern for building efficient, serverless data-processing pipelines on AWS.

Key Takeaways:

• Synchronous Invocation: Firehose invokes Lambda synchronously with a batch of records.
• The Lambda Contract is Key: You must adhere to the specific input and output structure, including the recordId, result status (Ok, Dropped, ProcessingFailed), and Base64 encoding/decoding of the data payload.
• Operational Limits: Be mindful of the 6 MB payload size limit for both the request and response, and the 5-minute maximum execution time for the Lambda function.
• Use Cases: This pattern is ideal for lightweight transformations like filtering, enrichment, and format conversion directly within the stream.

In this lesson, we saw how to return a ProcessingFailed status. But what does Firehose do with those failed records? In our next lesson, we will answer that question by exploring how to configure error handling for Firehose delivery streams using an S3 backup location.