Event-Driven Workflows with EventBridge Pipes and Schema Registry

What you'll build

A serverless workflow where every new order written to a DynamoDB table flows through an EventBridge Pipe — filtered to INSERT events only, enriched by a Lambda function into a clean domain event — and lands in a consumer Lambda, while the EventBridge Schema Registry discovers the event's schema automatically. No stream-polling glue code anywhere.

flowchart LR
    DDB[DynamoDB table<br/>orders stream] --> P[Pipe: filter INSERT<br/>+ enrich via Lambda]
    P --> B[Default event bus<br/>schema discovery on]
    B --> R[Rule: OrderCreated] --> C[Consumer Lambda]

We route the pipe through an event bus rather than straight into the consumer because schema discovery only works on event buses — that one hop is what gets you auto-discovered, versioned schemas for free.

Prerequisites

• An AWS account with credentials configured and permissions for DynamoDB, Lambda, IAM, EventBridge, and the Schema Registry.
• AWS CLI v2 — commands verified against v2.34.
• Lambda runtime python3.14 (current as of July 2026); any machine with zip and a shell. Everything here fits in the free tier apart from pennies of pipe requests; schema discovery's first 5 million events per month are free.

Set two variables every command below uses:

export AWS_REGION=us-east-1
export ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)

1. Create the table with a stream

aws dynamodb create-table \
  --table-name orders \
  --attribute-definitions AttributeName=orderId,AttributeType=S \
  --key-schema AttributeName=orderId,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

STREAM_ARN=$(aws dynamodb describe-table --table-name orders \
  --query 'Table.LatestStreamArn' --output text)

NEW_AND_OLD_IMAGES puts the full item into each stream record, so the enrichment step has everything it needs.

2. Deploy the enrichment and consumer functions

The enricher receives a batch (a JSON array) of raw stream records and must return an array — each element becomes one event on the bus. It unmarshals DynamoDB's typed attribute format into a plain domain event:

cat > enrich.py <<'EOF'
def handler(records, context):
    events = []
    for r in records:
        img = r["dynamodb"]["NewImage"]
        events.append({
            "orderId": img["orderId"]["S"],
            "total": float(img["total"]["N"]),
            "currency": img.get("currency", {}).get("S", "USD"),
        })
    return events
EOF

cat > consumer.py <<'EOF'
import json

def handler(event, context):
    print("OrderCreated:", json.dumps(event["detail"]))
EOF

Create a shared execution role and both functions:

aws iam create-role --role-name pipes-demo-lambda-role \
  --assume-role-policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}'
aws iam attach-role-policy --role-name pipes-demo-lambda-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

sleep 10   # IAM is eventually consistent; give the role time to propagate

zip enrich.zip enrich.py && zip consumer.zip consumer.py

aws lambda create-function --function-name order-enricher \
  --runtime python3.14 --handler enrich.handler --zip-file fileb://enrich.zip \
  --role arn:aws:iam::$ACCOUNT_ID:role/pipes-demo-lambda-role

aws lambda create-function --function-name order-consumer \
  --runtime python3.14 --handler consumer.handler --zip-file fileb://consumer.zip \
  --role arn:aws:iam::$ACCOUNT_ID:role/pipes-demo-lambda-role

3. Create the pipe's execution role

The pipe needs to read the stream, invoke the enricher, and put events on the default bus:

aws iam create-role --role-name orders-pipe-role \
  --assume-role-policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"pipes.amazonaws.com"},"Action":"sts:AssumeRole"}]}'

cat > pipe-policy.json <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {"Effect": "Allow",
     "Action": ["dynamodb:DescribeStream", "dynamodb:GetRecords",
                "dynamodb:GetShardIterator", "dynamodb:ListStreams"],
     "Resource": "$STREAM_ARN"},
    {"Effect": "Allow", "Action": "lambda:InvokeFunction",
     "Resource": "arn:aws:lambda:$AWS_REGION:$ACCOUNT_ID:function:order-enricher"},
    {"Effect": "Allow", "Action": "events:PutEvents",
     "Resource": "arn:aws:events:$AWS_REGION:$ACCOUNT_ID:event-bus/default"}
  ]
}
EOF

aws iam put-role-policy --role-name orders-pipe-role \
  --policy-name orders-pipe-policy --policy-document file://pipe-policy.json

4. Create the pipe

Filtering, batching, enrichment, and target wiring — all declarative:

aws pipes create-pipe \
  --name orders-pipe \
  --role-arn arn:aws:iam::$ACCOUNT_ID:role/orders-pipe-role \
  --source "$STREAM_ARN" \
  --source-parameters '{
    "DynamoDBStreamParameters": {"StartingPosition": "TRIM_HORIZON",
      "BatchSize": 10, "MaximumBatchingWindowInSeconds": 5},
    "FilterCriteria": {"Filters": [{"Pattern": "{\"eventName\": [\"INSERT\"]}"}]}
  }' \
  --enrichment arn:aws:lambda:$AWS_REGION:$ACCOUNT_ID:function:order-enricher \
  --target arn:aws:events:$AWS_REGION:$ACCOUNT_ID:event-bus/default \
  --target-parameters '{"EventBridgeEventBusParameters": {
    "Source": "myapp.orders", "DetailType": "OrderCreated"}}'

Use TRIM_HORIZON, not LATEST — a pipe can take several minutes to start polling, and LATEST silently drops anything written in that window. The filter means updates and deletes never invoke the enricher, and you're only billed for events that pass it.

5. Turn on schema discovery and route to the consumer

aws schemas create-discoverer \
  --source-arn arn:aws:events:$AWS_REGION:$ACCOUNT_ID:event-bus/default

aws events put-rule --name order-created \
  --event-pattern '{"source": ["myapp.orders"], "detail-type": ["OrderCreated"]}'

aws lambda add-permission --function-name order-consumer \
  --statement-id order-created-rule --action lambda:InvokeFunction \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:$AWS_REGION:$ACCOUNT_ID:rule/order-created

aws events put-targets --rule order-created \
  --targets "Id"="consumer","Arn"="arn:aws:lambda:$AWS_REGION:$ACCOUNT_ID:function:order-consumer"

Verify it works

Wait for the pipe to reach RUNNING, then write an order:

aws pipes describe-pipe --name orders-pipe --query 'CurrentState' --output text
# RUNNING

aws dynamodb put-item --table-name orders \
  --item '{"orderId": {"S": "ord-1001"}, "total": {"N": "42.5"}, "currency": {"S": "USD"}}'

aws logs tail /aws/lambda/order-consumer --since 10m

Within a minute or two you should see the enriched, unmarshalled event:

2026-07-20T18:04:11 ... OrderCreated: {"orderId": "ord-1001", "total": 42.5, "currency": "USD"}

Delete the item (aws dynamodb delete-item --table-name orders --key '{"orderId": {"S": "ord-1001"}}') and confirm nothing new is logged — the REMOVE event was filtered out. Then check the registry (discovery can lag a few minutes behind the first event):

aws schemas list-schemas --registry-name discovered-schemas

{"Schemas": [{"SchemaName": "myapp.orders@OrderCreated", "VersionCount": 1, ...}]}

Run aws schemas describe-schema --registry-name discovered-schemas --schema-name 'myapp.orders@OrderCreated' to see the inferred OpenAPI document. Change the event's shape later and the registry versions it automatically.

Troubleshooting

InvalidParameterValueException ... The role defined for the function cannot be assumed by Lambda. — You created the function too soon after the role. IAM propagation takes a few seconds; rerun create-function after ~10 seconds.

Pipe is RUNNING but the enricher never fires. Filter patterns match the raw stream record, so item attributes use DynamoDB's typed JSON. A pattern like {"dynamodb": {"NewImage": {"status": ["shipped"]}}} never matches — it must be {"dynamodb": {"NewImage": {"status": {"S": ["shipped"]}}}}. Check aws pipes describe-pipe --name orders-pipe --query 'StateReason' for IAM failures too.

The specified log group does not exist. from aws logs tail. The consumer was never invoked. Usual cause: put-targets without lambda add-permission, which fails silently (it shows up only as FailedInvocations in the rule's CloudWatch metrics). Rerun step 5's add-permission command exactly.

Schema missing from discovered-schemas. The discoverer only sees events sent after it was created — send a fresh put-item and wait a few minutes. Also note events larger than 1000 KiB are skipped with no error.

Next steps

Download typed code bindings (Java, Python, TypeScript, Go) for your discovered schema from the EventBridge console or the AWS Toolkit in your IDE, so consumers get compile-time checks against OrderCreated. Add a DeadLetterConfig and MaximumRetryAttempts to the pipe's source parameters so poison records don't block the shard. And when the enrichment is a simple reshape like this one, try replacing the Lambda entirely with the pipe's input transformer — one less function to own. Clean up with aws pipes delete-pipe --name orders-pipe, then delete the table, functions, rule, discoverer, and roles.