Complete Guide to AWS Lambda in Production

Q: How do I test Lambda functions locally?

Use SAM CLI (sam local invoke, sam local start-api) to test functions locally. Use LocalStack for AWS service emulation. Write unit tests for handler logic and integration tests for AWS service interactions.

Introduction

AWS Lambda is the most widely used serverless platform. Getting a Lambda function to work is easy. Running it reliably in production is hard. Cold starts, connection pooling, deployment strategies, observability, and security all require careful attention. This guide walks through everything you need to run Lambda functions in production with confidence.

Cold Start Optimization

Cold starts are the biggest performance challenge in Lambda. When a function has not been invoked recently, AWS provisions a new execution environment, loads the runtime, and initializes your code. This adds 1-10 seconds of latency.

Measuring Cold Starts

import json
import time
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Track initialization time
init_start = time.time()

# Heavy initialization outside handler
import boto3
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("users")

init_duration = time.time() - init_start
logger.info(json.dumps({"event": "cold_start_init", "duration_ms": round(init_duration * 1000)}))

def lambda_handler(event, context):
    start = time.time()
    
    # Your function logic
    user_id = event["pathParameters"]["userId"]
    response = table.get_item(Key={"id": user_id})
    
    duration = time.time() - start
    logger.info(json.dumps({
        "event": "invocation_complete",
        "duration_ms": round(duration * 1000),
        "request_id": context.aws_request_id
    }))
    
    return {
        "statusCode": 200,
        "body": json.dumps(response.get("Item", {}), default=str)
    }

Reducing Package Size

Large deployment packages slow down cold starts. Minimize dependencies.

# Bad: importing entire library
import numpy  # 55MB package

# Good: import only what you need
from numpy import array, mean  # Still loads full library

# Best: use lightweight alternatives
# Instead of numpy for simple math:
def calculate_average(numbers):
    return sum(numbers) / len(numbers)

Lambda Layers

Layers let you share dependencies across functions without bundling them in every deployment package. This reduces package size and deployment time.

# Create a layer with shared dependencies
mkdir -p python/lib/python3.11/site-packages
pip install requests -t python/lib/python3.11/site-packages/
zip -r requests-layer.zip python/

# Publish the layer
aws lambda publish-layer-version \
  --layer-name requests-layer \
  --zip-file fileb://requests-layer.zip \
  --compatible-runtimes python3.11

# Use the layer in your function
# The layer ARN is configured in the function's layer list
import requests  # Available from the layer

def lambda_handler(event, context):
    response = requests.get("https://api.example.com/data")
    return {"statusCode": 200, "body": response.text}

Provisioned Concurrency

For latency-sensitive functions, provisioned concurrency keeps instances warm.

# Enable provisioned concurrency
aws lambda put-provisioned-concurrency-config \
  --function-name my-api \
  --qualifier live \
  --provisioned-concurrent-executions 20

# Application-level: track if this is a warm or cold invocation
import os

def lambda_handler(event, context):
    # Check if this is a cold start
    is_cold = os.environ.get("AWS_LAMBDA_INITIALIZATION_TYPE") == "provisioned-concurrency"
    
    if is_cold:
        logger.info("Running on provisioned concurrency (no cold start)")
    
    # Process request
    return handle_request(event)

Runtime Selection

Runtime	Cold Start (avg)	Best For
Go	~100ms	High performance, low cold start
Node.js	~200ms	Web APIs, fast startup
Python	~300ms	Data processing, ML inference
.NET	~500ms	Enterprise apps
Java	~1000ms	Heavy frameworks (Spring)

Choose Go or Node.js for latency-sensitive endpoints. Use Python for data processing where cold starts are less critical.

Deployment Patterns

Blue/Green Deployment

Deploy a new version alongside the old one. Shift traffic gradually.

# Publish new version
aws lambda publish-version --function-name my-api

# Update alias to point to new version (10% canary)
aws lambda update-alias \
  --function-name my-api \
  --name live \
  --function-version 2 \
  --routing-config '{"AdditionalVersionWeights": {"1": 0.9}}'

# Full rollout: shift 100% to new version
aws lambda update-alias \
  --function-name my-api \
  --name live \
  --function-version 2 \
  --routing-config '{}'

Traffic Shifting with CodeDeploy

# appspec.yml for CodeDeploy deployment
version: 0.0
Resources:
  - myLambdaFunction:
      Type: AWS::Lambda::Function
      Properties:
        Name: my-api
        Alias: live
        CurrentVersion: 1
        TargetVersion: 2
        DeploymentPreference:
          Type: Canary10Percent5Minutes

Infrastructure as Code with SAM

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./src
      Handler: app.handler
      Runtime: python3.11
      MemorySize: 512
      Timeout: 30
      Environment:
        Variables:
          TABLE_NAME: !Ref UsersTable
      Policies:
        - DynamoDBReadPolicy:
            TableName: !Ref UsersTable
      Events:
        GetUser:
          Type: Api
          Properties:
            Path: /users/{userId}
            Method: GET
      AutoPublishAlias: live
      DeploymentPreference:
        Type: Canary10Percent5Minutes
        Alarms:
          - !Ref ApiErrorAlarm

  UsersTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH

  ApiErrorAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      MetricName: Errors
      Namespace: AWS/Lambda
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 1
      Threshold: 5
      ComparisonOperator: GreaterThanThreshold

Connection Pooling

Each Lambda execution environment opens its own database connection. With high concurrency, this exhausts the connection pool.

The Problem

100 concurrent Lambda invocations
→ 100 separate execution environments
→ 100 database connections
→ Connection pool exhausted (max: 50)
→ Connection refused errors

Solution: RDS Proxy

RDS Proxy manages a connection pool shared across Lambda execution environments.

import os
import boto3
import json

# Connect through RDS Proxy (endpoint is different from direct RDS)
def get_db_connection():
    proxy_endpoint = os.environ["DB_PROXY_ENDPOINT"]
    # RDS Proxy handles pooling automatically
    return psycopg2.connect(
        host=proxy_endpoint,
        dbname=os.environ["DB_NAME"],
        user=os.environ["DB_USER"],
        password=get_secret()["password"]
    )

def lambda_handler(event, context):
    conn = get_db_connection()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (event["user_id"],))
        user = cursor.fetchone()
        return {"statusCode": 200, "body": json.dumps(user)}
    finally:
        cursor.close()
        conn.close()

Alternative: DynamoDB

DynamoDB does not use connection pools. Each Lambda invocation makes an independent HTTP API call. This makes it ideal for serverless workloads.

import boto3
import json

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("users")

def lambda_handler(event, context):
    response = table.get_item(Key={"id": event["user_id"]})
    return {
        "statusCode": 200,
        "body": json.dumps(response.get("Item", {}), default=str)
    }

Observability

Structured Logging with CloudWatch

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    request_id = context.aws_request_id
    function_name = context.function_name
    
    # Log structured event
    logger.info(json.dumps({
        "level": "INFO",
        "request_id": request_id,
        "function": function_name,
        "event_type": "request_start",
        "user_id": event.get("user_id"),
        "http_method": event.get("httpMethod"),
        "path": event.get("path")
    }))
    
    try:
        result = process_request(event)
        logger.info(json.dumps({
            "level": "INFO",
            "request_id": request_id,
            "event_type": "request_success",
            "duration_ms": 150
        }))
        return result
    except Exception as e:
        logger.error(json.dumps({
            "level": "ERROR",
            "request_id": request_id,
            "event_type": "request_error",
            "error": str(e),
            "error_type": type(e).__name__
        }))
        return {"statusCode": 500, "body": json.dumps({"error": "Internal error"})}

AWS X-Ray Tracing

from aws_xray_sdk.core import patch_all
import boto3
import json

# Patch all AWS SDK calls for tracing
patch_all()

def lambda_handler(event, context):
    # X-Ray traces this function and all AWS SDK calls
    user_id = event["user_id"]
    
    # Subsegment for custom logic
    from aws_xray_sdk.core import xray_recorder
    with xray_recorder.in_subsegment("fetch_user_data"):
        dynamodb = boto3.resource("dynamodb")
        table = dynamodb.Table("users")
        response = table.get_item(Key={"id": user_id})
    
    with xray_recorder.in_subsegment("transform_data"):
        user = response.get("Item", {})
        user["full_name"] = f"{user.get('first_name', '')} {user.get('last_name', '')}"
    
    return {"statusCode": 200, "body": json.dumps(user, default=str)}

Custom Metrics with CloudWatch

import boto3
import time

cloudwatch = boto3.client("cloudwatch")

def emit_metric(metric_name, value, unit="Count"):
    cloudwatch.put_metric_data(
        Namespace="MyApp/Lambda",
        MetricData=[{
            "MetricName": metric_name,
            "Value": value,
            "Unit": unit,
            "Dimensions": [
                {"Name": "Function", "Value": "user-api"}
            ]
        }]
    )

def lambda_handler(event, context):
    start = time.time()
    
    try:
        result = process_request(event)
        duration = (time.time() - start) * 1000
        emit_metric("RequestCount", 1)
        emit_metric("RequestDuration", duration, "Milliseconds")
        return result
    except Exception as e:
        emit_metric("ErrorCount", 1)
        raise

Security Hardening

Least Privilege IAM Roles

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:Query"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123:table/users"
    },
    {
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": "arn:aws:secretsmanager:us-east-1:123:secret:db-credentials-*"
    }
  ]
}

Secrets Management

import os
import json
import boto3
from functools import lru_cache

@lru_cache(maxsize=1)
def get_db_credentials():
    """Load credentials once per execution environment."""
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(
        SecretId=os.environ["DB_SECRET_ARN"]
    )
    return json.loads(response["SecretString"])

def lambda_handler(event, context):
    creds = get_db_credentials()
    # Credentials are cached across warm invocations
    conn = psycopg2.connect(
        host=creds["host"],
        dbname=creds["dbname"],
        user=creds["username"],
        password=creds["password"]
    )
    # ...

VPC Configuration

Functions that access private RDS instances must be in a VPC. This adds cold start latency.

# SAM template with VPC configuration
Resources:
  VpcFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./src
      Handler: app.handler
      Runtime: python3.11
      VpcConfig:
        SubnetIds:
          - subnet-abc123
          - subnet-def456
        SecurityGroupIds:
          - sg-abc123
      Policies:
        - VPCAccessPolicy: {}
        - DynamoDBReadPolicy:
            TableName: users

Input Validation

import json
import re

def validate_email(email):
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def lambda_handler(event, context):
    body = json.loads(event.get("body", "{}"))
    
    # Validate required fields
    required = ["email", "name"]
    for field in required:
        if field not in body:
            return {"statusCode": 400, "body": json.dumps({"error": f"Missing field: {field}"})}
    
    # Validate email format
    if not validate_email(body["email"]):
        return {"statusCode": 400, "body": json.dumps({"error": "Invalid email format"})}
    
    # Validate name length
    if len(body["name"]) > 100:
        return {"statusCode": 400, "body": json.dumps({"error": "Name too long"})}
    
    # Process valid request
    user = create_user(body)
    return {"statusCode": 201, "body": json.dumps(user)}

Error Handling and Retries

Idempotent Functions

Lambda retries failed async invocations. Make functions idempotent to handle duplicate processing.

import hashlib

def lambda_handler(event, context):
    # Generate idempotency key from event
    event_hash = hashlib.md5(json.dumps(event, sort_keys=True).encode()).hexdigest()
    idempotency_key = f"idempotency:{event_hash}"
    
    # Check if already processed
    if redis.exists(idempotency_key):
        return {"status": "already_processed", "result": redis.get(idempotency_key)}
    
    # Process the event
    result = process_event(event)
    
    # Store result with TTL
    redis.setex(idempotency_key, 3600, json.dumps(result))
    
    return {"status": "processed", "result": result}

Dead Letter Queues

# SAM template with DLQ
Resources:
  FunctionWithDLQ:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./src
      Handler: app.handler
      Runtime: python3.11
      EventInvokeConfig:
        MaximumEventAgeInSeconds: 3600
        MaximumRetryAttempts: 2
        DestinationConfig:
          OnFailure:
            Destination: !GetAtt FailureQueue.Arn
  
  FailureQueue:
    Type: AWS::SQS::Queue
    Properties:
      MessageRetentionPeriod: 1209600  # 14 days

Cost Optimization

Memory Tuning

Lambda charges by memory × time. More memory also means more CPU. Finding the right balance reduces cost.

# Power tuning: test different memory configurations
# Use aws-lambda-power-tuning tool to find optimal memory

# General guidelines:
# 128MB: Simple I/O operations
# 256MB: Light processing, API handlers
# 512MB: Moderate processing, database queries
# 1024MB: CPU-intensive tasks
# 3008MB: Maximum CPU for heavy computation

Avoiding Unnecessary Invocations

# Batch processing: process multiple items per invocation
def lambda_handler(event, context):
    # SQS batch: up to 10 messages per invocation
    batch_failures = []
    
    for record in event["Records"]:
        try:
            process_message(json.loads(record["body"]))
        except Exception as e:
            batch_failures.append({"itemIdentifier": record["messageId"]})
    
    if batch_failures:
        return {"batchItemFailures": batch_failures}
    
    return {"batchItemFailures": []}

Production Checklist

FAQ

How do I reduce Lambda cold starts?

Minimize package size, use Lambda layers for shared dependencies, initialize heavy resources outside the handler, choose fast-start runtimes (Go, Node.js), and use provisioned concurrency for latency-sensitive functions.

Should I use RDS Proxy with Lambda?

Yes, if you connect to RDS from Lambda. Without RDS Proxy, high concurrency exhausts database connection pools. RDS Proxy maintains a connection pool shared across Lambda execution environments.

How do I deploy Lambda without downtime?

Use alias routing with CodeDeploy. Publish a new version, shift 10% of traffic to it, monitor for errors, and complete the rollout. If errors exceed the threshold, CodeDeploy automatically rolls back.

What is the maximum concurrency for Lambda?

The default account limit is 1,000 concurrent executions per region. You can request a quota increase. Use reserved concurrency to limit specific functions and prevent them from consuming all account concurrency.

How do I handle long-running tasks?

Lambda has a 15-minute timeout. For longer tasks, use Step Functions to chain multiple Lambda functions, or use AWS Batch/Fargate for truly long-running workloads.

How do I test Lambda functions locally?

Use SAM CLI (sam local invoke, sam local start-api) to test functions locally. Use LocalStack for AWS service emulation. Write unit tests for handler logic and integration tests for AWS service interactions.