Most Serverless Framework projects start as an API. Then someone asks where the clickstream goes, or the IoT telemetry, or the audit events, and the team discovers that the patterns they know — API Gateway in front of Lambda in front of DynamoDB — do not fit high-volume append-only data at all. This tutorial builds the other half: a streaming ingestion pipeline that lands events in an Apache Iceberg table on S3 and lets analysts query them with Athena, defined entirely in one serverless.yml.
The shape we are building:
producers -> Kinesis Data Stream -> Lambda (enrich/validate) -> Firehose -> Iceberg table on S3 -> Athena
\-> DLQ (S3 / SQS)
Stack: Serverless Framework V4, Node.js 22 on arm64, AWS SDK v3, Kinesis Data Streams in on-demand mode, Amazon Data Firehose, AWS Glue Data Catalog, and Athena.
When to use this instead of the API-plus-DynamoDB reflex
Reach for a streaming pipeline when all of these are true, and not otherwise:
- Events are appended, never updated in place by the producer.
- Volume is high enough that per-item writes are a cost or throughput problem (roughly: sustained thousands of events per second, or millions per day).
- The consumers are analytical — dashboards, ML features, ad-hoc SQL — rather than a user waiting on a response.
- You need replay. Kinesis retains records for 24 hours by default and up to 365 days, so a consumer bug is recoverable instead of a data-loss incident.
If you only need ordered fan-out to a handful of consumers, EventBridge or SQS is simpler and cheaper. If you need sub-millisecond point lookups by key, that is still DynamoDB. A pipeline is worth its operational weight only when the query pattern is "scan a lot of history and aggregate".
1. The stream and the ingestion function
service: events-pipeline
frameworkVersion: '4'
provider:
name: aws
runtime: nodejs22.x
architecture: arm64
memorySize: 512
region: eu-west-1
environment:
DELIVERY_STREAM: !Ref DeliveryStream
STAGE: ${sls:stage}
iam:
role:
statements:
- Effect: Allow
Action:
- firehose:PutRecordBatch
Resource: !GetAtt DeliveryStream.Arn
build:
esbuild:
bundle: true
minify: true
target: node22
exclude: ['@aws-sdk/*']
functions:
ingest:
handler: src/ingest.handler
timeout: 60
events:
- stream:
type: kinesis
arn: !GetAtt EventStream.Arn
batchSize: 500
maximumBatchingWindow: 10
startingPosition: LATEST
parallelizationFactor: 4
maximumRetryAttempts: 3
bisectBatchOnFunctionError: true
functionResponseType: ReportBatchItemFailures
destinations:
onFailure:
arn: !GetAtt IngestDlq.Arn
type: sqs
resources:
Resources:
EventStream:
Type: AWS::Kinesis::Stream
Properties:
Name: ${self:service}-${sls:stage}-events
StreamModeDetails:
StreamMode: ON_DEMAND
RetentionPeriodHours: 168
IngestDlq:
Type: AWS::SQS::Queue
Properties:
QueueName: ${self:service}-${sls:stage}-ingest-dlq
MessageRetentionPeriod: 1209600
The event-source-mapping options carry most of the reliability of this design:
ON_DEMANDstream mode removes shard-count capacity planning. Start here; move to provisioned shards only when you have a steady, predictable rate and the on-demand premium shows up in the bill.batchSize: 500withmaximumBatchingWindow: 10trades up to ten seconds of latency for far fewer invocations. On an analytics pipeline nobody notices ten seconds; your invocation count drops by orders of magnitude.parallelizationFactor: 4runs four concurrent batches per shard while preserving order within a partition key — the main lever for iterator age when a single shard gets hot.functionResponseType: ReportBatchItemFailureswithbisectBatchOnFunctionErroris the difference between one poison record and a stalled shard. Without it, a single record that always throws blocks everything behind it until it expires.
2. Partial batch failures, done properly
src/ingest.ts:
import type { KinesisStreamHandler, KinesisStreamRecord } from 'aws-lambda';
import { FirehoseClient, PutRecordBatchCommand } from '@aws-sdk/client-firehose';
const firehose = new FirehoseClient({});
const DeliveryStreamName = process.env.DELIVERY_STREAM!;
type Enriched = { event_id: string; event_type: string; tenant_id: string; occurred_at: string; payload: unknown };
const decode = (r: KinesisStreamRecord): Enriched => {
const raw = JSON.parse(Buffer.from(r.kinesis.data, 'base64').toString('utf8'));
if (!raw.event_id || !raw.event_type || !raw.tenant_id) throw new Error('missing required field');
return {
event_id: String(raw.event_id),
event_type: String(raw.event_type),
tenant_id: String(raw.tenant_id),
occurred_at: new Date(raw.occurred_at ?? r.kinesis.approximateArrivalTimestamp * 1000).toISOString(),
payload: raw.payload ?? {},
};
};
export const handler: KinesisStreamHandler = async (event) => {
const batchItemFailures: { itemIdentifier: string }[] = [];
const records: { Data: Buffer }[] = [];
const seqOf: string[] = [];
for (const r of event.Records) {
try {
// Newline-delimited JSON: Firehose treats each line as one record downstream.
records.push({ Data: Buffer.from(JSON.stringify(decode(r)) + '\n') });
seqOf.push(r.kinesis.sequenceNumber);
} catch (err) {
console.warn(JSON.stringify({ event: 'decode_failed', seq: r.kinesis.sequenceNumber, err: String(err) }));
// Malformed input is not retryable: drop it, do not report it as a failure.
}
}
for (let i = 0; i < records.length; i += 500) {
const slice = records.slice(i, i + 500);
const res = await firehose.send(new PutRecordBatchCommand({ DeliveryStreamName, Records: slice }));
if (res.FailedPutCount) {
res.RequestResponses?.forEach((rr, idx) => {
if (rr.ErrorCode) batchItemFailures.push({ itemIdentifier: seqOf[i + idx] });
});
}
}
// Kinesis retries from the earliest reported sequence number onward.
batchItemFailures.sort((a, b) => a.itemIdentifier.localeCompare(b.itemIdentifier));
return { batchItemFailures: batchItemFailures.slice(0, 1) };
};
Two subtleties worth internalising:
Distinguish retryable from non-retryable. A record that fails JSON.parse will fail again on every retry. Log it, optionally copy it to a quarantine bucket, and move on. A Firehose throttle will succeed later — that one belongs in batchItemFailures.
Kinesis partial-batch semantics are checkpoint-based, not per-item. Unlike SQS, reporting a sequence number rewinds the shard to that point and redelivers everything after it. That is why we return only the lowest failing sequence number, and why every downstream write must be idempotent on event_id.
3. Firehose into an Iceberg table
Firehose can write directly into Apache Iceberg tables registered in the Glue Data Catalog. Compared with the older "raw JSON to S3, then a Glue crawler, then hope" pattern, you get schema in the catalog from day one, ACID semantics, and row-level updates and deletes — which matters the first time legal asks you to erase one user's history from three years of Parquet.
resources:
Resources:
LakeBucket:
Type: AWS::S3::Bucket
Properties:
BucketName: ${self:service}-${sls:stage}-lake-${aws:accountId}
BucketEncryption:
ServerSideEncryptionConfiguration:
- ServerSideEncryptionByDefault: { SSEAlgorithm: AES256 }
PublicAccessBlockConfiguration:
BlockPublicAcls: true
BlockPublicPolicy: true
IgnorePublicAcls: true
RestrictPublicBuckets: true
LakeDatabase:
Type: AWS::Glue::Database
Properties:
CatalogId: !Ref AWS::AccountId
DatabaseInput:
Name: ${self:service}_${sls:stage}
DeliveryStream:
Type: AWS::KinesisFirehose::DeliveryStream
Properties:
DeliveryStreamName: ${self:service}-${sls:stage}-delivery
DeliveryStreamType: DirectPut
IcebergDestinationConfiguration:
RoleARN: !GetAtt FirehoseRole.Arn
CatalogConfiguration:
CatalogArn: !Sub 'arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog'
DestinationTableConfigurationList:
- DestinationDatabaseName: ${self:service}_${sls:stage}
DestinationTableName: events
UniqueKeys: ['event_id']
BufferingHints:
IntervalInSeconds: 300
SizeInMBs: 128
S3Configuration:
BucketARN: !GetAtt LakeBucket.Arn
RoleARN: !GetAtt FirehoseRole.Arn
ErrorOutputPrefix: errors/!{firehose:error-output-type}/
RetryOptions:
DurationInSeconds: 300
CloudWatchLoggingOptions:
Enabled: true
LogGroupName: /aws/kinesisfirehose/${self:service}-${sls:stage}
LogStreamName: iceberg
BufferingHints is the knob that decides whether your table is pleasant or miserable to query. Small buffers mean fresh data and thousands of tiny Parquet files; large buffers mean fewer, bigger files and a five-minute lag. For analytics, 300 seconds and 128 MB is a sane default. Whatever you choose, schedule Iceberg table maintenance — OPTIMIZE ... REWRITE DATA USING BIN_PACK and VACUUM on a daily EventBridge-scheduled Lambda — or query times will degrade steadily as the small-file count grows.
The Iceberg table itself is created once with Athena DDL rather than CloudFormation; run it from a deployment step or a one-off:
CREATE TABLE events (
event_id string,
event_type string,
tenant_id string,
occurred_at timestamp,
payload string
)
PARTITIONED BY (day(occurred_at), event_type)
LOCATION 's3://events-pipeline-prod-lake-123456789012/events/'
TBLPROPERTIES ('table_type' = 'ICEBERG', 'format' = 'parquet');
Partition on the column you filter by most — almost always time — plus at most one low-cardinality dimension. Partitioning by tenant_id with ten thousand tenants produces ten thousand directories and a planning phase slower than the scan itself.
4. The Firehose IAM role
Firehose assumes a role you provide. Scope it to this bucket, this database, this table:
FirehoseRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal: { Service: firehose.amazonaws.com }
Action: sts:AssumeRole
Condition:
StringEquals: { 'sts:ExternalId': !Ref AWS::AccountId }
Policies:
- PolicyName: firehose-iceberg
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action: ['s3:AbortMultipartUpload', 's3:GetBucketLocation', 's3:GetObject', 's3:ListBucket', 's3:ListBucketMultipartUploads', 's3:PutObject', 's3:DeleteObject']
Resource:
- !GetAtt LakeBucket.Arn
- !Sub '${LakeBucket.Arn}/*'
- Effect: Allow
Action: ['glue:GetDatabase', 'glue:GetTable', 'glue:GetTableVersion', 'glue:GetTableVersions', 'glue:UpdateTable']
Resource:
- !Sub 'arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog'
- !Sub 'arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${self:service}_${sls:stage}'
- !Sub 'arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${self:service}_${sls:stage}/*'
- Effect: Allow
Action: ['logs:PutLogEvents']
Resource: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${self:service}-${sls:stage}:*'
The sts:ExternalId condition on the trust policy blocks the confused-deputy case where another account's Firehose could be pointed at your role.
5. Querying, and keeping the bill honest
SELECT event_type, count(*) AS events, count(DISTINCT tenant_id) AS tenants
FROM events_pipeline_prod.events
WHERE occurred_at >= current_timestamp - interval '7' day
GROUP BY event_type
ORDER BY events DESC;
Athena charges per terabyte scanned, so the pipeline's design decisions show up directly in query cost:
- Always filter on the partition column. A query without a time predicate scans the whole table. Set an Athena workgroup
BytesScannedCutoffPerQueryto make runaway queries fail fast rather than bill. - Select columns, not
*. Parquet is columnar;SELECT *defeats the entire point. - Keep the raw
payloadas a JSON string and extract withjson_extract_scalarfor exploration. Promote fields to real columns once an access pattern proves itself — Iceberg supports schema evolution, so adding a column later does not mean rewriting history. - Lifecycle the S3 error prefix. Firehose's
errors/output grows quietly; expire it after 30 days.
6. What to alarm on
A pipeline fails silently more often than an API does. Four alarms cover most of it:
| Metric | Why |
|---|---|
GetRecords.IteratorAgeMilliseconds (Kinesis) | The single best indicator that consumption has fallen behind production. Alarm above a few minutes. |
WriteProvisionedThroughputExceeded | Producers are being throttled; records are being lost at the source. |
DeliveryToIceberg.Success (Firehose) | Below 1 means records are landing in the S3 error prefix instead of the table. |
SQS ApproximateNumberOfMessagesVisible on the DLQ | Any non-zero value is a batch that exhausted its retries. |
Add a daily freshness check — a scheduled Lambda running SELECT max(occurred_at) FROM events and alarming if the answer is older than an hour. It catches the failure mode where every component reports healthy and no data has arrived since Tuesday.
7. Cost intuition
For a pipeline handling roughly 100 million events a day at 1 KB each (about 100 GB/day):
- Kinesis on-demand is priced per GB ingested plus a per-hour stream charge; at this volume it is typically the largest line item, and the point at which you should price out provisioned shards.
- Lambda is nearly free here because batching means about 200,000 invocations a day at ~50 ms each, not 100 million.
- Firehose is per GB ingested, with a modest surcharge for Iceberg delivery.
- S3 is cheap for Parquet (expect 5-10x compression) — but small-file overhead and old Iceberg snapshots are not, which is why
VACUUMbelongs on a schedule. - Athena is entirely a function of how well you partitioned.
The ratio to remember: batching at the Lambda layer and buffering at the Firehose layer are what keep the per-request costs from dominating. A naive "one Lambda invocation and one S3 PUT per event" version of this pipeline costs roughly two orders of magnitude more for the same data.
Migration path from an existing API
If you already have events flowing through an API Gateway endpoint, you do not need a rewrite. Point the existing handler at PutRecords on the Kinesis stream (batched, with the tenant or entity ID as the partition key), and run the two write paths side by side until Athena counts match your current store. Then retire the old path. We cover the same dual-write, verify, cut-over sequence in our Migration to Serverless Infrastructure work.
Summary
- Use a streaming pipeline for high-volume append-only analytical data, not as a default.
- Batch hard at the event source mapping; it is where the cost lives.
ReportBatchItemFailuresplusbisectBatchOnFunctionErrorplus an SQS on-failure destination is the minimum viable reliability configuration for a Kinesis consumer.- Land in Iceberg rather than raw JSON: catalogued schema, schema evolution, and row-level deletes.
- Schedule
OPTIMIZEandVACUUMfrom day one, and alarm on iterator age and data freshness.
Building or untangling a serverless data pipeline? Our Serverless Integration Services and Serverless Architecture Design teams do this work on AWS every week — get in touch.