Develop Code for Lambda | 🏗️ Build A Real-Time Data Processing Pipeline

Exam Guide: Developer - Associate
🏗️ Domain 1: Development with AWS Services
📘 Task 2: Develop Code for Lambda

Lambda is the most heavily tested service on the DVA-C02. It is also perhaps, one of the most used and talked about services in general too, well after EC2 and S3. So, you need to know how to configure it, write code for it, handle errors, tune performance, and integrate it with practically every other AWS service.

📘Concepts

Lambda Execution Model

When you invoke a Lambda function, AWS:
1. Finds or creates an execution environment (container)
2. Loads your code and initializes it (cold start)
3. Runs your handler function
4. Keeps the environment warm for reuse (subsequent invocations skip step 2)

Code outside the handler runs once during cold start and is reused. This is why you initialize SDK clients and database connections outside the handler.

Key Configuration Parameters

Invocation Types and Error Handling

Lambda Destinations vs DLQ

Lambda Destinations are the modern approach and preferred over DLQs. DLQs are still valid for SQS event source mappings.

VPC Access

By default, Lambda runs in an AWS-managed VPC and can access the internet and public AWS services. When you attach Lambda to your own VPC:

Without VPC:  Lambda → Internet → AWS Services ✅  |  Lambda → Private RDS ❌
With VPC:     Lambda → Private RDS ✅  |  Lambda → Internet ❌ (needs NAT Gateway)
              Lambda → VPC Endpoint → AWS Services ✅

Key Points:

1. VPC Lambda needs private subnets with a NAT Gateway for internet access
2. Use VPC endpoints (PrivateLink) to access DynamoDB, S3, SQS without NAT
3. Use RDS Proxy for connection pooling because Lambda can overwhelm databases with concurrent connections

Cold Starts

A cold start happens when Lambda creates a new execution environment. Strategies to reduce them:

1. Provisioned Concurrency: pre-warms environments (costs money when idle)
2. Keep deployment packages small: smaller packages initialize faster
3. Initialize outside the handler: SDK clients, DB connections, config
4. Use ARM (Graviton2): often faster and 20% cheaper

Kinesis And Lambda Key Settings

🏗️ Build A Real Time Data Processing Pipeline

Now let's put these concepts into practice by building a Real-Time Data Processing Pipeline using Lambda:

• A Lambda function with VPC access connecting to an RDS database
• Lambda layers for shared dependencies
• A DLQ (Dead Letter Queue) and Lambda Destinations for error handling
• A Kinesis stream processor that transforms data in near real time
• Performance tuning with memory configuration

This covers all the skills for this task: Lambda configuration, VPC access, error handling, event lifecycle, integrations, and performance tuning.

Prerequisites

• An AWS account (free tier covers most of this)

Part I

Create and Configure a Lambda Function

Create the Function

Step 01: Open the Lambda

Step 02: **Click **Create function

Step 03: Create function
Choose Author from scratch

• Function name: DataProcessor
• Runtime: Python 3.12

Click Create function

✅Green banner: Successfully created the function "DataProcessor".

Before writing code, let's walk through the configuration tabs.

Step 04: Click the Configuration tab.
General configuration
Click Edit:

• Memory: 256 MB (default is 128 MB)
• Ephemeral storage: 512 MB (default)
• Timeout: 0 min 30 sec (default is 3 seconds)

Click Save.

Lambda allocates CPU proportionally to memory. At 1,769 MB you get one full vCPU. Increasing memory from 128 MB to 256 MB doubles your CPU therefore your function might run in half the time, costing the same or less.

Step 05: Environment variables
Click Environment variables → Edit → Add environment variable:

• Key: LOG_LEVEL, Value: INFO
• Key: STAGE, Value: dev

Click Save.

✅Green banner: Successfully updated the function "DataProcessor".

import os

# Access environment variables in your code
log_level = os.environ.get('LOG_LEVEL', 'INFO')
stage = os.environ.get('STAGE', 'dev')

Lambda encrypts all environment variables at rest by default using an AWS managed KMS key. For extra security, you can use a customer managed KMS key and decrypt in your code.

Part II

Create a Lambda Layer

Layers let you share code and libraries across multiple functions. Let's create one with a utility module.

Build and Upload a Layer

Layers must be uploaded as a zip file.
We'll use AWS CloudShell: a browser-based terminal built into the AWS Console so that you don't need anything installed locally.

Step 01: Open CloudShell by clicking the terminal icon (>_) in the top navigation bar of the AWS Console next to the search bar

Step 02: Wait for it to initialize as it might take a few seconds the first time

Step 03:
Run these commands:

mkdir -p python
cat > python/utils.py << 'EOF'
"""
Shared utilities for Lambda functions.
This module is packaged as a Lambda Layer so multiple functions can use it.
"""
import json
import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=0.5):
    """Retry with exponential backoff and jitter."""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt)
            jitter = random.uniform(0, delay)
            time.sleep(delay + jitter)

def format_response(status_code, body):
    """Standard API Gateway response format."""
    return {
        'statusCode': status_code,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps(body, default=str)
    }
EOF

zip -r utils-layer.zip python/

Step 04: Download the zip
Click Actions ▼ (top right of CloudShell) → Download file → type utils-layer.zip → click Download

Why the python/ folder? Lambda layers must follow a specific directory structure. For Python, your code must be inside a python/ folder in the zip. Lambda adds this path to sys.path automatically so your functions can import utils directly.

Step 05: Upload the Layer via Console
In the Lambda console, click Layers in the left sidebar

Step 06: Click Create layer

• Name: shared-utils
• Upload: Choose the utils-layer.zip file
• Compatible runtimes - optional: Python 3.12 Click Create

✅Green banner: Successfully created layer shared-utils version 1.

Step 07: Attach the Layer to Your Function
Go back to the DataProcessor function
Scroll down to the Layers section
Click Edit

Step 08: Edit layers
Click Add a layer
Choose Custom layers
Select shared-utils
Version: 1
Click Add

Click Save

✅Green banner: Successfully updated the function "DataProcessor".

Step 09: Now you can use it in your function code:

from utils import format_response, retry_with_backoff

def lambda_handler(event, context):
    return format_response(200, {'message': 'Hello from DataProcessor'})

A function can have up to 5 layers. The total unzipped size of the function + all layers can't exceed 250 MB. Layers are versioned and immutable. Each upload creates a new version.

Part III

Error Handling: DLQ and Lambda Destinations

Create a Dead Letter Queue

Step 01: Open the SQS console

Step 02: Click Create queue

• Type: Standard
• Name: data-processor-dlq

Click Create queue

✅Green banner: Queue data-processor-dlq created successfully

Step 03: Copy the Queue ARN

Step 04: Create a Success Queue (for Destinations)
Create another queue:

• Name: data-processor-success

Copy the Queue ARN

Step 05: Lambda → DataProcessor → Configuration → Permissions → Click the Role name
Add permissions → Attach policies
Search for AmazonSQSFullAccess and attach it

Step 06: Configure the DLQ on the Lambda Function
Go back to the DataProcessor function
Configuration tab → Asynchronous invocation → Edit

• Maximum age of event: 1 h 0 min 0 sec
• Retry attempts: 2
• Dead-letter queue service ▼: Select Amazon SQS
• Queue ▼: Select data-processor-dlq

Click Save

✅Green banner: Successfully updated the function "DataProcessor".

Destinations are the modern approach. They capture both success AND failure.

Step 07: Configure Lambda Destinations
Still in Asynchronous invocation, find the Destinations section
Click Add destination

Step 08: Add destination

• Source: Asynchronous invocation
• Condition: On success
• Destination type: SQS queue
• Destination: data-processor-success

Click Save

✅Green banner: Your changes have been saved.

Step 09: Click Add destination again

• Source: Asynchronous invocation
• Condition: On failure
• Destination type: SQS queue
• Destination: data-processor-dlq

Click Save

✅Green banner: Your changes have been saved.

Lambda Destinations are preferred over DLQs because:

• Destinations capture both success and failure (DLQ only captures failure)
• Destinations include more context (request payload, response, error details)
• Destinations work with async invocations only (same as DLQ)
• DLQs are still valid for SQS event source mappings

Test Error Handling

Step 10: Let's create a function that sometimes fails to see the error handling in action:

Deploy this code, then test it

import json
import random

def lambda_handler(event, context):
    """
    This function randomly fails to demonstrate error handling.
    When invoked asynchronously:
    1. First attempt fails → Lambda retries
    2. Second attempt fails → Lambda retries again
    3. Third attempt fails → Event goes to DLQ / failure destination
    """
    action = event.get('action', 'process')

    if action == 'fail':
        # Always fail — will end up in DLQ after 3 attempts
        raise Exception("Simulated failure for testing DLQ")

    if action == 'random':
        # 50% chance of failure
        if random.random() < 0.5:
            raise Exception("Random failure occurred")

    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Processed successfully', 'action': action})
    }

Step 11: Click Test tab

Step 12: Test event

• Test event action: Create new event
• Invocation type: Synchronous
• Event name: TestFailure
• Even JSON:

{
  "action": "fail"
}

Click Test

❌ Executing function: failed

The function will fail. Since this is a synchronous test invocation, you'll see the error immediately. To test the DLQ flow, you need an asynchronous invocation (from S3, SNS, or EventBridge).

Part IV

Lambda with VPC Access

When Lambda needs to access private resources like RDS or ElastiCache, you attach it to a VPC. Let's walk through the configuration.

Understanding VPC Access

Without VPC:
  Lambda → Internet → AWS Services (DynamoDB, S3, SQS)  ✅
  Lambda → Private RDS                                    ❌

With VPC:
  Lambda → VPC → Private RDS                              ✅
  Lambda → VPC → Internet                                 ❌ (no NAT)
  Lambda → VPC → NAT Gateway → Internet                   ✅
  Lambda → VPC → VPC Endpoint → AWS Services              ✅

Configure VPC Access (Console Walkthrough)

Step 01: Open the DataProcessor function
Configuration tab → VPC → Edit

Step 02: Edit VPC
Select the default VPC
Select at least 2 private subnets (for high availability)
Select or create a security group that allows outbound traffic

Click Save

⚠️Important: The Lambda execution role needs the AWSLambdaVPCAccessExecutionRole managed policy.

Step 03: Connection Reuse Pattern

When connecting to databases from Lambda, initialize the connection outside the handler:

import os
import json
import boto3

# This runs ONCE during cold start, then reuses across warm invocations
# This is critical for database connections — you don't want to open
# a new connection on every single invocation
db_host = os.environ.get('DB_HOST')
db_name = os.environ.get('DB_NAME')

# In a real app, you'd initialize your database connection here
# connection = pymysql.connect(host=db_host, ...)

def lambda_handler(event, context):
    """
    The handler runs on every invocation.
    The connection above is reused across warm invocations.
    """
    # Use the connection...
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Connected to database'})
    }

Always initialize SDK clients and database connections outside the handler. This code runs once during the cold start and is reused for subsequent warm invocations.
For RDS specifically, use RDS Proxy to manage connection pooling. Lambda can open hundreds of connections simultaneously, which can overwhelm a database.

Part V

Process Streaming Data with Kinesis

Let's build a real-time clickstream processor.

Create a Kinesis Data Stream

Step 01: Open the Kinesis console
Click Create data stream

• Data stream name: clickstream
• Capacity mode: On-demand

Click Create data stream

✅Green banner: Data stream clickstream successfully created

Step 02: Create the Stream Processor Function
Go to Lambda → Create function
Configure:

• Function name: ClickstreamProcessor
• Runtime: Python 3.12
• Memory: 512 MB
• Timeout: 60 seconds

Step 03: Paste this code

import json
import base64
from datetime import datetime

def lambda_handler(event, context):
    """
    Processes clickstream records from Kinesis in near real time.

    Key concepts for the exam:
    - Kinesis data is base64 encoded
    - Records come in batches
    - The partition key determines which shard receives the record
    - Use a high-cardinality partition key (like userId) for even distribution
    """
    processed = 0
    failed = 0

    print(f"Received {len(event['Records'])} records from Kinesis")

    for record in event['Records']:
        try:
            # Kinesis data is base64 encoded — decode it
            raw_data = base64.b64decode(record['kinesis']['data']).decode('utf-8')
            data = json.loads(raw_data)

            # Extract clickstream fields
            user_id = data.get('userId', 'anonymous')
            page = data.get('page', 'unknown')
            action = data.get('action', 'unknown')
            timestamp = data.get('timestamp', datetime.utcnow().isoformat())

            # Process the click event
            print(f"Click: user={user_id}, page={page}, action={action}, time={timestamp}")

            # In a real app, you'd:
            # - Aggregate metrics
            # - Write to DynamoDB or S3
            # - Trigger alerts for specific patterns

            processed += 1

        except Exception as e:
            print(f"Failed to process record: {str(e)}")
            failed += 1

    print(f"Batch complete: {processed} processed, {failed} failed")

    return {
        'statusCode': 200,
        'body': json.dumps({
            'processed': processed,
            'failed': failed
        })
    }

Click Deploy

✅Green banner: Successfully updated the function "ClickstreamProcessor".

Step 04: Add Permissions
The Lambda execution role needs permission to read from Kinesis:
Go to Configuration → Permissions → click the role name
Add permissions → Attach policies
Search for and attach AmazonKinesisReadOnlyAccess

Step 05: Connect Kinesis to Lambda
In the ClickstreamProcessor function, click Add trigger

Step 06: Add trigger
Select a source ▼: Kinesis

• Kinesis stream: clickstream
• Batch size: 100
• Starting position: Latest
• Batch window: 5
• On-failure destination: ``
• Retry attempts: 3
• Split batch on error: ✔ enabled
• Concurrent batches per shard: 10

Click Add

✅Green banner: The trigger clickstream was successfully added to function ClickstreamProcessor.

Key Kinesis + Lambda settings:

• Batch size: how many records per invocation (max 10,000)
• Batch window: how long to wait to fill the batch (max 300s)
• Parallelization factor: (1–10) process multiple batches per shard concurrently
• Bisect batch on error: splits the batch in half on failure to isolate the bad record
• Starting position: LATEST (new records only) or TRIM_HORIZON (from the beginning)

Step 07: Test with Sample Data
Send test records to the stream:
Open the Kinesis console → click clickstream

Step 08: clickstream
Click Data viewer tab
Or use the AWS CLI to send test data:

`shell
aws kinesis put-record \
--stream-name clickstream \
--partition-key "USER-001" \
--data '{"userId":"USER-001","page":"/products/laptop","action":"view","timestamp":"2026-04-24T10:00:00Z"}'
`

Step 09: Check the ClickstreamProcessor CloudWatch logs to see the processed records

Part VI

Performance Tuning

Memory vs Duration Experiment

Let's see how memory affects performance:

Step 01: Open the ClickstreamProcessor function

Step 02: Configuration → General configuration → Edit
Set Memory to 128 MB → Save

Step 03: Run a test → note the Duration and Max Memory Used in the execution results

`json
{
"Records": [
{
"kinesis": {
"data": "eyJ1c2VySWQiOiJVU0VSLTAwMSIsInBhZ2UiOiIvcHJvZHVjdHMvbGFwdG9wIiwiYWN0aW9uIjoidmlldyIsInRpbWVzdGFtcCI6IjIwMjYtMDQtMjRUMTA6MDA6MDBaIn0="
},
"eventSource": "aws:kinesis",
"eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/clickstream"
},
{
"kinesis": {
"data": "eyJ1c2VySWQiOiJVU0VSLTAwMiIsInBhZ2UiOiIvY2hlY2tvdXQiLCJhY3Rpb24iOiJjbGljayIsInRpbWVzdGFtcCI6IjIwMjYtMDQtMjRUMTA6MDE6MDBaIn0="
},
"eventSource": "aws:kinesis",
"eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/clickstream"
}
]
}
`

Step 04: Change memory to 256 MB → test again

Step 05: Change to 512 MB → test again

Step 06: Change to 1024 MB → test again

You'll typically see:

The sweet spot is where increasing memory no longer significantly reduces duration.

Step 07: Concurrency Settings
Configuration → Concurrency and recursion detection
Concurrency→ Edit

Step 08: Edit concurrency

• USe unreserved account concurrency: Uses the shared regional pool (default 1000)
• Reserve concurrency: Guarantees capacity AND caps the function

Select Reserve concurrency: 50

Click Save

✅Green banner: Your changes have been saved.

This means:

• Your function is guaranteed 50 concurrent executions
• It can never exceed 50 (acts as a throttle)
• The remaining 950 are available for other functions

Reserved concurrency is free and serves two purposes: guaranteeing capacity AND protecting downstream services from being overwhelmed. Provisioned concurrency costs money even when idle but eliminates cold starts.

🏗️ What You Built | 📘Exam Concepts Recap

⚠️ Clean Up Protocol

1. Lambda → Delete DataProcessor, ClickstreamProcessor
2. Lambda Layers → Delete shared-utils
3. Kinesis → Delete clickstream stream
4. SQS → Delete data-processor-dlq, data-processor-success
5. IAM → Delete the Lambda execution roles
6. CloudWatch → Delete the log groups

Key Takeaways

1. Memory = CPU: more memory means more CPU. Find the sweet spot where cost and performance balance.
2. Initialize outside the handler: SDK clients, DB connections, config loading. Reused across warm invocations.
3. Lambda Destinations > DLQ: Destinations capture success AND failure with more context.
4. ReportBatchItemFailures for SQS, BisectBatchOnFunctionError for Kinesis to isolate bad records.
5. VPC Lambda loses internet: needs NAT Gateway or VPC endpoints for AWS services.
6. RDS Proxy for Lambda-to-RDS: manages connection pooling.
7. Layers: up to 5 per function, 250 MB total unzipped, versioned and immutable.
8. 15-minute timeout is the max: for longer tasks, use Step Functions or ECS.
9. Provisioned concurrency: eliminates cold starts but costs money when idle.
10. Kinesis parallelization factor" (1–10) lets you process multiple batches per shard.

Additional Resources

• What Is AWS Lambda
• Using Lambda to process records from Amazon Kinesis Data Streams
• Configuring AWS Lambda functions
• Best practices for working with AWS Lambda functions
• Configuring reserved concurrency for a function

🏗️