AWS Lambda Deep Dive — Functions, Triggers, Cold Starts, and Layers

# lambda_s3_processor.py
# AWS Lambda function triggered by S3 uploads
import json
import boto3
import csv
import urllib.parse
from datetime import datetime

s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')

def lambda_handler(event, context):
    """
    Process CSV file uploaded to S3 and store in DynamoDB.
    Triggered by S3:PutObject event.
    """
    print(f"Event received: {json.dumps(event, default=str)[:200]}...")

    # Get bucket and key from the S3 event
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'])

    print(f"Processing s3://{bucket}/{key}")

    # Download and process the CSV
    response = s3.get_object(Bucket=bucket, Key=key)
    lines = response['Body'].read().decode('utf-8').splitlines()
    reader = csv.DictReader(lines)

    table = dynamodb.Table('ProcessedFiles')
    row_count = 0

    for row in reader:
        item = {
            'file_key': key,
            'row_id': str(row_count),
            'data': json.dumps(row),
            'processed_at': datetime.now().isoformat(),
        }
        table.put_item(Item=item)
        row_count += 1

    print(f"Processed {row_count} rows from {key}")

    return {
        'statusCode': 200,
        'body': json.dumps({
            'processed_rows': row_count,
            'source_file': key,
        })
    }

# Test locally
def test_locally():
    """Simulate Lambda execution locally."""
    test_event = {
        "Records": [{
            "s3": {
                "bucket": {"name": "dodatech-uploads"},
                "object": {"key": "sales/2026/06/15/sales_data.csv"}
            }
        }]
    }

    result = lambda_handler(test_event, None)
    print(f"\nResult: {json.dumps(result, indent=2)}")

if __name__ == "__main__":
    # Simulate without actual AWS calls
    print("=== AWS Lambda: S3 Processor ===")
    print("\nFunction configuration:")
    print(f"  Runtime: Python 3.12")
    print(f"  Memory: 512 MB")
    print(f"  Timeout: 30 seconds")
    print(f"  Trigger: S3 (s3:ObjectCreated:*)")
    print(f"  Destination: DynamoDB (ProcessedFiles)")
    print("\nNote: This example requires AWS credentials to run fully.")
    print("See the mock simulation below for local testing.\n")
    test_locally()

# lambda_mock_simulator.py
# Simulate Lambda triggers and cold starts for learning
import time
import random

class LambdaSimulator:
    def __init__(self, runtime="python3.12", memory_mb=512):
        self.runtime = runtime
        self.memory = memory_mb
        self.warm = False

    def cold_start(self):
        penalty = random.uniform(0.2, 0.5)
        print(f"  [COLD START] Initializing runtime... ({penalty:.2f}s)")
        time.sleep(penalty)
        self.warm = True

    def invoke(self, event_name):
        is_cold = not self.warm
        if is_cold:
            self.cold_start()

        latency = random.uniform(0.05, 0.15)
        start = time.time()
        time.sleep(latency)

        print(f"  {'[WARM]' if not is_cold else '[COLD]'} {event_name}: "
              f"processed in {latency*1000:.0f}ms")
        return {"latency_ms": round(latency * 1000), "cold_start": is_cold}

sim = LambdaSimulator()

print("=== Lambda Invocation Simulation ===\n")
results = []
for event in ["S3:Upload", "API:GET /users", "API:POST /order", "SQS:Message", "S3:Upload"]:
    result = sim.invoke(event)
    results.append(result)

cold_count = sum(1 for r in results if r["cold_start"])
print(f"\nSummary: {len(results)} invocations, {cold_count} cold starts")

# layer-structure.yaml
# Lambda Layer directory structure
layer_name: pandas-layer
compatible_runtimes:
  - python3.11
  - python3.12
directory_structure:
  python/
    lib/
      python3.12/
        site-packages/
          pandas/
          numpy/
  # Zip and publish:
  # aws lambda publish-layer-version \
  #   --layer-name pandas-layer \
  #   --zip-file fileb://pandas-layer.zip \
  #   --compatible-runtimes python3.12

# lambda_benchmark.py
# Benchmark simulated Lambda cold vs warm starts
import time
import random

def benchmark_lambda(runtime_name, cold_start_range, handler_time_range, iterations=10):
    cold_times = []
    warm_times = []

    for i in range(iterations):
        is_cold = random.random() < 0.3  # 30% cold start probability

        if is_cold:
            start = time.time()
            cs_delay = random.uniform(*cold_start_range)
            time.sleep(cs_delay)
            handler_delay = random.uniform(*handler_time_range)
            time.sleep(handler_delay)
            cold_times.append((time.time() - start) * 1000)
        else:
            start = time.time()
            handler_delay = random.uniform(*handler_time_range)
            time.sleep(handler_delay)
            warm_times.append((time.time() - start) * 1000)

    return cold_times, warm_times

print("=== Lambda Runtime Cold Start Benchmark (simulated) ===")
runtimes = [
    ("Python 3.12", (0.2, 0.5), (0.02, 0.08)),
    ("Node.js 20", (0.1, 0.3), (0.01, 0.05)),
    ("Java 21", (1.0, 3.0), (0.03, 0.10)),
    ("Go 1.x", (0.1, 0.3), (0.005, 0.02)),
]

for name, cs_range, handler_range in runtimes:
    cold, warm = benchmark_lambda(name, cs_range, handler_range, 20)
    avg_cold = sum(cold) / len(cold) if cold else 0
    avg_warm = sum(warm) / len(warm) if warm else 0
    print(f"\n  {name:<15} Cold: {avg_cold:6.0f}ms avg   Warm: {avg_warm:6.0f}ms avg")