Real-Time Data Pipeline: Kinesis Streams + Firehose + Lambda + S3
Scenario
Your IoT devices emit sensor readings at 50,000 events/minute. You need a pipeline that captures these events in real time, enriches each record with a timestamp partition key, compresses the data, and stores it in S3 under a date-partitioned prefix (year/month/day/hour/) for downstream analytics.
Build the full pipeline: Kinesis Data Stream → Lambda transformation → Kinesis Firehose → compressed Parquet in S3 with dynamic partitioning.
Your Objectives
- Create a Kinesis Data Stream with 4 shards and 24-hour retention.
- Create a Lambda function that reads from the stream, enriches each record, and returns it in the Firehose transformation response format (Base64-encoded, with result = Ok).
- Create a Firehose delivery stream with Kinesis Data Stream as source, Lambda as processor, GZIP compression, and S3 destination with dynamic partitioning using !{timestamp:yyyy/MM/dd/HH}/.
- Configure S3 buffering: 5 MB or 60 seconds, whichever comes first.
- Create an IAM role for Firehose with permissions to read from the Kinesis stream, invoke Lambda, and write to S3.
- Output the stream ARN, Firehose ARN, and the S3 bucket name.
Additional Context
Kinesis Data Stream vs Firehose: Kinesis Data Streams is a low-latency, high-throughput stream — consumers read records within milliseconds and must manage their own checkpointing. Firehose is a managed delivery service — it buffers records and delivers to S3, Redshift, OpenSearch, or HTTP endpoints, but has a minimum 60-second latency. For real-time processing use Kinesis + Lambda; for batch delivery to S3 use Firehose.
The Lambda transformation function for Firehose receives records in batches and must return each record with a result field: Ok (processed), Dropped (filtered out), or ProcessingFailed (retry). The data must be Base64-encoded in the response. A newline delimiter is required between JSON records for S3 analytics tools.
variables.tf
variable "aws_region" { type = string; default = "us-east-1" }
variable "stream_name" { type = string; default = "iot-sensor-stream" }
variable "environment" { type = string; default = "prod" }
variable "shard_count" { type = number; default = 4 }
variable "retention_hours" { type = number; default = 24 }main.tf
provider "aws" { region = var.aws_region }
locals { tags = { Environment = var.environment, ManagedBy = "terraform" } }
data "aws_caller_identity" "current" {}
# S3 bucket for the processed sensor data.
resource "aws_s3_bucket" "data_lake" {
bucket = "${var.stream_name}-data-lake-${var.environment}-${data.aws_caller_identity.current.account_id}"
tags = local.tags
}
resource "aws_s3_bucket_lifecycle_configuration" "data_lake" {
bucket = aws_s3_bucket.data_lake.id
rule {
id = "transition-to-ia"
status = "Enabled"
transition { days = 30; storage_class = "STANDARD_IA" }
filter { prefix = "" }
}
}
# Kinesis Data Stream — producers write to this stream.
resource "aws_kinesis_stream" "iot" {
name = "${var.stream_name}-${var.environment}"
shard_count = var.shard_count
retention_period = var.retention_hours
encryption_type = "KMS"
kms_key_id = "alias/aws/kinesis"
tags = local.tags
}
# Lambda transform — enriches records before Firehose delivers to S3.
data "aws_iam_policy_document" "lambda_assume" {
statement { effect = "Allow"; actions = ["sts:AssumeRole"]
principals { type = "Service"; identifiers = ["lambda.amazonaws.com"] } }
}
resource "aws_iam_role" "lambda" {
name = "${var.stream_name}-transform-role"
assume_role_policy = data.aws_iam_policy_document.lambda_assume.json
}
resource "aws_iam_role_policy_attachment" "lambda_logs" {
role = aws_iam_role.lambda.name
policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}
resource "aws_cloudwatch_log_group" "transform" {
name = "/aws/lambda/${var.stream_name}-transform-${var.environment}"
retention_in_days = 14
}
resource "aws_lambda_function" "transform" {
function_name = "${var.stream_name}-transform-${var.environment}"
runtime = "python3.12"
handler = "transform.lambda_handler"
role = aws_iam_role.lambda.arn
filename = "${path.module}/transform.zip"
timeout = 60 # Firehose waits up to 5 minutes; keep Lambda timeout short
description = "Enriches Kinesis records before S3 delivery"
environment { variables = { ENVIRONMENT = var.environment } }
depends_on = [aws_cloudwatch_log_group.transform]
tags = local.tags
}
# Firehose IAM role.
resource "aws_iam_role" "firehose" {
name = "${var.stream_name}-firehose-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{ Effect = "Allow"; Principal = { Service = "firehose.amazonaws.com" }; Action = "sts:AssumeRole" }]
})
}
resource "aws_iam_role_policy" "firehose" {
name = "firehose-permissions"
role = aws_iam_role.firehose.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{ Effect = "Allow"; Action = ["kinesis:GetRecords","kinesis:GetShardIterator","kinesis:DescribeStream","kinesis:ListShards"]
Resource = aws_kinesis_stream.iot.arn },
{ Effect = "Allow"; Action = ["s3:PutObject","s3:GetObject","s3:ListBucket"]; Resource = ["${aws_s3_bucket.data_lake.arn}","${aws_s3_bucket.data_lake.arn}/*"] },
{ Effect = "Allow"; Action = ["lambda:InvokeFunction"]; Resource = aws_lambda_function.transform.arn },
{ Effect = "Allow"; Action = ["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"]; Resource = "*" }
]
})
}
# Kinesis Firehose delivery stream.
resource "aws_kinesis_firehose_delivery_stream" "iot" {
name = "${var.stream_name}-firehose-${var.environment}"
destination = "extended_s3"
tags = local.tags
# Source: read from the Kinesis Data Stream.
kinesis_source_configuration {
kinesis_stream_arn = aws_kinesis_stream.iot.arn
role_arn = aws_iam_role.firehose.arn
}
extended_s3_configuration {
role_arn = aws_iam_role.firehose.arn
bucket_arn = aws_s3_bucket.data_lake.arn
buffering_size = 5 # MB — deliver when buffer reaches 5 MB
buffering_interval = 60 # seconds — deliver after 60s even if buffer not full
compression_format = "GZIP"
# Dynamic partitioning — partition by timestamp embedded in the event.
prefix = "sensors/!{timestamp:yyyy}/!{timestamp:MM}/!{timestamp:dd}/!{timestamp:HH}/"
error_output_prefix = "errors/!{firehose:error-output-type}/!{timestamp:yyyy}/!{timestamp:MM}/!{timestamp:dd}/"
# Lambda data transformation — enrich records before S3 delivery.
processing_configuration {
enabled = true
processors {
type = "Lambda"
parameters {
parameter_name = "LambdaArn"
parameter_value = "${aws_lambda_function.transform.arn}:$LATEST"
}
parameters {
parameter_name = "BufferSizeInMBs"
parameter_value = "1"
}
parameters {
parameter_name = "BufferIntervalInSeconds"
parameter_value = "60"
}
}
}
cloudwatch_logging_options {
enabled = true
log_group_name = "/aws/kinesisfirehose/${var.stream_name}-${var.environment}"
log_stream_name = "S3Delivery"
}
}
}outputs.tf
output "kinesis_stream_arn" { value = aws_kinesis_stream.iot.arn }
output "firehose_arn" { value = aws_kinesis_firehose_delivery_stream.iot.arn }
output "data_lake_bucket" { value = aws_s3_bucket.data_lake.bucket }
output "transform_lambda" { value = aws_lambda_function.transform.arn }Workflow Commands
terraform init && terraform apply
# Produce test records to the Kinesis stream
aws kinesis put-records \
--stream-arn $(terraform output -raw kinesis_stream_arn) \
--records '[
{"Data":"eyJ0ZW1wIjo0Mi41LCJkZXZpY2VJZCI6ImRldi0wMDEifQ==","PartitionKey":"dev-001"},
{"Data":"eyJ0ZW1wIjo0MC4xLCJkZXZpY2VJZCI6ImRldi0wMDIifQ==","PartitionKey":"dev-002"}
]'
# Wait 60-120 seconds for Firehose buffering, then check S3
aws s3 ls s3://$(terraform output -raw data_lake_bucket)/sensors/ --recursive
# Check Firehose delivery metrics
aws cloudwatch get-metric-statistics \
--namespace AWS/Firehose \
--metric-name DeliveryToS3.Records \
--dimensions Name=DeliveryStreamName,Value=$(terraform output -raw firehose_arn | awk -F'/' '{print $NF}') \
--start-time $(date -u -v-10M +%Y-%m-%dT%H:%M:%SZ) \
--end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
--period 60 --statistics Sum
terraform destroy
✓ Expected Output After Apply
Apply complete! Resources: 10 added, 0 changed, 0 destroyed. Outputs: data_lake_bucket = "iot-sensor-stream-data-lake-prod-123456789012" firehose_arn = "arn:aws:firehose:us-east-1:123456789012:deliverystream/iot-sensor-stream-firehose-prod" kinesis_stream_arn = "arn:aws:kinesis:us-east-1:123456789012:stream/iot-sensor-stream-prod" transform_lambda = "arn:aws:lambda:us-east-1:123456789012:function:iot-sensor-stream-transform-prod"
