SERVERLESS DATA ARCHITECTURE

Why Serverless for Data Engineering?

Traditional data infrastructure requires provisioning servers, managing clusters, and paying for idle capacity. Serverless architecture flips this model—you only pay for what you use, automatically scale to zero, and eliminate operational overhead.

At STARK Industries, we migrated several data pipelines to serverless architecture and achieved:

• 70% cost reduction compared to always-on EC2 instances
• Zero infrastructure management - no patching, scaling, or monitoring servers
• Automatic scaling from zero to thousands of concurrent executions
• Built-in fault tolerance with automatic retries and dead-letter queues

Serverless Data Architecture Patterns

Our serverless data platform follows an event-driven architecture where data flowing through S3 triggers Lambda functions for processing, with results stored in DynamoDB:

Pattern 1: S3 Event-Driven Processing

The most common serverless pattern: trigger Lambda functions automatically when files are uploaded to S3. Perfect for ETL pipelines, image processing, and data validation.

Lambda Function for CSV Processing

import json
import boto3
import csv
from io import StringIO
from datetime import datetime

s3_client = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('processed-events')

def lambda_handler(event, context):
    """
    Triggered when CSV file is uploaded to S3.
    Processes each row and stores in DynamoDB.
    """
    
    # Get bucket and key from S3 event
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    print(f"Processing file: s3://{bucket}/{key}")
    
    try:
        # Download file from S3
        response = s3_client.get_object(Bucket=bucket, Key=key)
        content = response['Body'].read().decode('utf-8')
        
        # Parse CSV
        csv_reader = csv.DictReader(StringIO(content))
        
        processed_count = 0
        errors = []
        
        # Process each row
        for row in csv_reader:
            try:
                # Data validation
                if not row.get('user_id') or not row.get('event_type'):
                    errors.append(f"Missing required fields in row: {row}")
                    continue
                
                # Transform data
                item = {
                    'event_id': f"{row['user_id']}#{datetime.utcnow().isoformat()}",
                    'user_id': row['user_id'],
                    'event_type': row['event_type'],
                    'timestamp': int(datetime.utcnow().timestamp()),
                    'value': float(row.get('value', 0)),
                    'source_file': key,
                    'processed_at': datetime.utcnow().isoformat()
                }
                
                # Write to DynamoDB
                table.put_item(Item=item)
                processed_count += 1
                
            except Exception as e:
                errors.append(f"Error processing row {row}: {str(e)}")
        
        # Store processed file metadata
        metadata = {
            'file_name': key,
            'bucket': bucket,
            'processed_at': datetime.utcnow().isoformat(),
            'rows_processed': processed_count,
            'errors_count': len(errors)
        }
        
        # Write to processed files tracking table
        metadata_table = dynamodb.Table('file-metadata')
        metadata_table.put_item(Item=metadata)
        
        # Archive original file
        archive_key = f"archive/{key}"
        s3_client.copy_object(
            Bucket=bucket,
            CopySource={'Bucket': bucket, 'Key': key},
            Key=archive_key
        )
        s3_client.delete_object(Bucket=bucket, Key=key)
        
        print(f"Successfully processed {processed_count} rows")
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'processed': processed_count,
                'errors': len(errors)
            })
        }
        
    except Exception as e:
        print(f"Error processing file: {str(e)}")
        
        # Send to dead-letter queue for manual review
        sns = boto3.client('sns')
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789:processing-errors',
            Subject=f'Lambda processing error: {key}',
            Message=str(e)
        )
        
        raise

S3 Event Configuration

# serverless.yml - Serverless Framework configuration
service: data-processor

provider:
  name: aws
  runtime: python3.9
  region: us-east-1
  environment:
    TABLE_NAME: processed-events
  iamRoleStatements:
    - Effect: Allow
      Action:
        - s3:GetObject
        - s3:PutObject
        - s3:DeleteObject
      Resource: "arn:aws:s3:::data-bucket/*"
    - Effect: Allow
      Action:
        - dynamodb:PutItem
        - dynamodb:GetItem
      Resource: "arn:aws:dynamodb:us-east-1:*:table/processed-events"

functions:
  processCSV:
    handler: processor.lambda_handler
    timeout: 300
    memorySize: 1024
    events:
      - s3:
          bucket: data-bucket
          event: s3:ObjectCreated:*
          rules:
            - prefix: uploads/
            - suffix: .csv

Pattern 2: DynamoDB Streams for Real-Time Aggregation

DynamoDB Streams + Lambda enables real-time aggregations and materialized views without complex streaming infrastructure.

import boto3
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')
aggregates_table = dynamodb.Table('user-aggregates')

def lambda_handler(event, context):
    """
    Triggered by DynamoDB stream.
    Maintains real-time aggregates of user events.
    """
    
    for record in event['Records']:
        if record['eventName'] in ['INSERT', 'MODIFY']:
            # Get new item
            new_item = record['dynamodb']['NewImage']
            
            user_id = new_item['user_id']['S']
            event_type = new_item['event_type']['S']
            value = Decimal(new_item['value']['N'])
            
            # Update aggregate
            try:
                response = aggregates_table.update_item(
                    Key={'user_id': user_id},
                    UpdateExpression="""
                        SET event_count = if_not_exists(event_count, :zero) + :inc,
                            total_value = if_not_exists(total_value, :zero) + :value,
                            last_event_type = :event_type,
                            last_updated = :timestamp
                    """,
                    ExpressionAttributeValues={
                        ':zero': 0,
                        ':inc': 1,
                        ':value': value,
                        ':event_type': event_type,
                        ':timestamp': int(datetime.utcnow().timestamp())
                    },
                    ReturnValues='ALL_NEW'
                )
                
                # Check for alerts
                new_values = response['Attributes']
                if new_values['event_count'] > 1000:
                    # Trigger alert for high-volume user
                    send_alert(user_id, new_values['event_count'])
                
            except Exception as e:
                print(f"Error updating aggregate for {user_id}: {str(e)}")
                raise
    
    return {'statusCode': 200}

def send_alert(user_id, count):
    """Send SNS notification for high-volume users"""
    sns = boto3.client('sns')
    sns.publish(
        TopicArn='arn:aws:sns:us-east-1:123456789:user-alerts',
        Subject=f'High volume detected for user {user_id}',
        Message=f'User {user_id} has generated {count} events'
    )

Pattern 3: Step Functions for Complex Workflows

AWS Step Functions orchestrate multi-step data workflows with built-in error handling, retries, and parallel processing.

Step Functions State Machine Definition

{
  "Comment": "Data processing workflow",
  "StartAt": "ValidateData",
  "States": {
    "ValidateData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:validate-data",
      "Next": "CheckValidation",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "HandleError"
      }]
    },
    "CheckValidation": {
      "Type": "Choice",
      "Choices": [{
        "Variable": "$.isValid",
        "BooleanEquals": true,
        "Next": "ProcessRecords"
      }],
      "Default": "SendErrorNotification"
    },
    "ProcessRecords": {
      "Type": "Map",
      "ItemsPath": "$.records",
      "MaxConcurrency": 10,
      "Iterator": {
        "StartAt": "ProcessSingleRecord",
        "States": {
          "ProcessSingleRecord": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:us-east-1:123456789:function:process-record",
            "End": true,
            "Retry": [{
              "ErrorEquals": ["States.TaskFailed"],
              "IntervalSeconds": 2,
              "MaxAttempts": 3,
              "BackoffRate": 2
            }]
          }
        }
      },
      "Next": "AggregateResults"
    },
    "AggregateResults": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:aggregate",
      "Next": "StoreResults"
    },
    "StoreResults": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:store-dynamodb",
      "End": true
    },
    "SendErrorNotification": {
      "Type": "Task",
      "Resource": "arn:aws:sns:us-east-1:123456789:errors",
      "End": true
    },
    "HandleError": {
      "Type": "Fail",
      "Cause": "Data processing failed"
    }
  }
}

DynamoDB Design for Serverless

DynamoDB is the perfect database for serverless - it scales automatically, has pay-per-request pricing, and provides single-digit millisecond latency.

Single-Table Design Pattern

# Example: Multi-entity table design
# PK (Partition Key) | SK (Sort Key)      | Data
# -------------------|--------------------|-----------------------
# USER#123           | PROFILE            | {name, email, ...}
# USER#123           | EVENT#2025-11-15   | {event_type, value}
# USER#123           | AGGREGATE#DAILY    | {count, total}
# FILE#abc.csv       | METADATA           | {size, rows, status}
# FILE#abc.csv       | ROW#0001           | {user_id, data}

DynamoDB Table Configuration

import boto3

dynamodb = boto3.client('dynamodb')

# Create table with on-demand billing
table = dynamodb.create_table(
    TableName='serverless-data',
    KeySchema=[
        {'AttributeName': 'PK', 'KeyType': 'HASH'},
        {'AttributeName': 'SK', 'KeyType': 'RANGE'}
    ],
    AttributeDefinitions=[
        {'AttributeName': 'PK', 'AttributeType': 'S'},
        {'AttributeName': 'SK', 'AttributeType': 'S'},
        {'AttributeName': 'GSI1PK', 'AttributeType': 'S'},
        {'AttributeName': 'GSI1SK', 'AttributeType': 'S'}
    ],
    BillingMode='PAY_PER_REQUEST',  # No capacity planning needed
    StreamSpecification={
        'StreamEnabled': True,
        'StreamViewType': 'NEW_AND_OLD_IMAGES'
    },
    GlobalSecondaryIndexes=[{
        'IndexName': 'GSI1',
        'KeySchema': [
            {'AttributeName': 'GSI1PK', 'KeyType': 'HASH'},
            {'AttributeName': 'GSI1SK', 'KeyType': 'RANGE'}
        ],
        'Projection': {'ProjectionType': 'ALL'}
    }]
)

Cost Optimization Strategies

1. Right-Size Lambda Memory

# Test different memory configurations
# More memory = faster CPU = faster execution = potentially lower cost

# 128 MB: 5000ms execution = $0.0000208
# 512 MB: 1500ms execution = $0.0000125  ← Cheaper!
# 1024 MB: 1000ms execution = $0.0000167

2. Use S3 Lifecycle Policies

{
  "Rules": [{
    "Id": "ArchiveOldData",
    "Status": "Enabled",
    "Transitions": [
      {
        "Days": 30,
        "StorageClass": "INTELLIGENT_TIERING"
      },
      {
        "Days": 90,
        "StorageClass": "GLACIER"
      }
    ],
    "Expiration": {
      "Days": 365
    }
  }]
}

3. Enable DynamoDB Auto Scaling

For predictable workloads, provisioned capacity with auto-scaling can be 70% cheaper than on-demand pricing.

Monitoring and Debugging

Serverless applications require different monitoring approaches. Use these AWS services:

• CloudWatch Logs: Automatic logging for all Lambda invocations
• X-Ray: Distributed tracing across Lambda, DynamoDB, and S3
• CloudWatch Metrics: Invocations, duration, errors, throttles
• CloudWatch Alarms: Alert on error rates, duration, or cost thresholds

# Add structured logging to Lambda
import json
import logging

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

def lambda_handler(event, context):
    logger.info(json.dumps({
        'event': 'processing_started',
        'request_id': context.request_id,
        'function_name': context.function_name,
        'record_count': len(event['Records'])
    }))
    
    # ... processing logic ...
    
    logger.info(json.dumps({
        'event': 'processing_completed',
        'duration_ms': context.get_remaining_time_in_millis(),
        'status': 'success'
    }))

Real-World Results

Our serverless data platform at STARK Industries processes:

• 5 million events per day
• Monthly cost: $450 (vs. $1,500 for EC2-based solution)
• Zero downtime in 6 months of operation
• Automatic scaling to 10x load during peak hours
• p99 latency: 200ms for processing pipeline

When NOT to Use Serverless

Serverless isn't always the right choice. Avoid it for:

• Long-running processes: Lambda has 15-minute max execution time
• Consistent high-volume: Always-on servers may be cheaper
• Low-latency requirements: Cold starts can add 1-3 seconds
• Complex state management: Stateless functions complicate some workflows

Conclusion

Serverless architecture transforms how we build data systems—eliminating infrastructure management, reducing costs, and enabling teams to focus on business logic instead of operations. While it's not a silver bullet, serverless is ideal for event-driven data processing, periodic ETL jobs, and variable-load applications.

The patterns and examples in this article provide a foundation for building production-grade serverless data platforms. Start small, measure everything, and iterate based on real-world usage patterns.