Using the SQS Ingest Lambda for Delta Lake

The sqs-ingest Lambda function is a powerful AWS SQS-based solution for ingesting JSON data directly into Delta Lake tables. Modeled after kafka-delta-ingest, this function enables reliable, scalable append-only writes to Delta Lake from various data sources.

How the SQS Ingest Solution Works

1. JSON Data Production: Applications and services send JSON-formatted messages to an SQS queue
2. Queue Processing: The sqs-ingest Lambda function polls the queue for new messages
3. Record Conversion: Each JSON message is translated into a row in the Delta Lake table
4. Concurrent Write Safety: Optional DynamoDB locking ensures safe multiple-writer scenarios

Key Capabilities

• Direct Delta Lake Writes: Converts JSON messages to Delta table rows in a single step
• Single Table Focus: Designed to work with one Delta table per Lambda instance
• Flexible Message Handling: Supports both direct SQS messages and SNS-wrapped messages
• Buffering Control: Configurable message batching for performance optimization

When to Use SQS Ingest

SQS Ingest is ideal for:

• JSON-based data pipelines: APIs, webhooks, and event-driven architectures
• Small to medium write volume: Thousands to hundreds of thousands of records per hour
• Append-only workloads: Time-series data, event logs, transaction records
• Multiple producer scenarios: Microservices architectures with diverse data sources

Benefits of SQS Ingest

• Direct Integration: No intermediate Parquet files required
• JSON Native: Works directly with JSON data formats
• Reliable Delivery: Built-in SQS message persistence and retry logic
• Scalable Processing: Lambda auto-scaling handles variable message volumes
• Cost-Effective: Pay-per-use model with no idle resources

Implementation Guide

Step 1: Create SQS Queue

aws sqs create-queue \
  --queue-name my-data-ingestion-queue.fifo

Step 2: Deploy sqs-ingest Lambda

Using the provided Terraform configuration:

cd oxbow/lambdas/sqs-ingest
cargo lambda build --release --output-format zip
aws lambda create-function \
  --function-name sqs-ingest 
  --handler not.required.for.rust 
  --runtime provided.al2 
  --code S3Bucket=my-bucket,S3Key=lambda.zip
  --environment Variables='{"DELTA_TABLE_URI":"s3://my-bucket/delta-table/"}'

Step 3: Configure Lambda Trigger

resource "aws_lambda_event_source_mapping" "sqs_trigger" {
  event_source_arn = aws_sqs_queue.data_queue.arn
  function_name    = aws_lambda_function.sqs_ingest.arn
  batch_size       = 10  # Default SQS batch size
  enabled          = true
}

Step 4: Application Integration (Python Example)

import json
import boto3

sqs = boto3.client('sqs')


def send_event(data):
    """Send JSON data to SQS for Delta table ingestion"""
    response = sqs.send_message(
        QueueUrl='https://sqs.us-west-2.amazonaws.com/123456789012/my-data-ingestion-queue.fifo',
        MessageBody=json.dumps(data),
        MessageGroupId='event-group',
        MessageDeduplicationId=str(uuid.uuid4())
    )
    return response


# Example event data
event = {
    "timestamp": "2024-01-01T12:00:00Z",
    "user_id": "user123",
    "event_type": "purchase",
    "amount": 99.99,
    "items": [
        {"sku": "ABC123", "quantity": 2},
        {"sku": "DEF456", "quantity": 1}
    ]
}

send_event(event)

Configuration Options

Environment Variables

Variable	Default	Description
RUST_LOG	error	Log level control (trace, debug, info, warn, error)
DELTA_TABLE_URI	Required	S3 URI of the Delta table to append to
BUFFER_MORE_QUEUE_URL	None	SQS queue URL for additional message buffering
BUFFER_MORE_MESSAGES	0	Number of additional messages to consume per invocation
AWS_S3_LOCKING_PROVIDER	None	Set to dynamodb for concurrent write safety
DYNAMO_LOCK_TABLE_NAME	None	DynamoDB table for locking (required for concurrent writes)
UNWRAP_SNS_ENVELOPE	None	Enable for SNS-wrapped messages

Example Environment Configuration

export DELTA_TABLE_URI="s3://my-bucket/my-delta-table/"
export AWS_S3_LOCKING_PROVIDER="dynamodb"
export DYNAMO_LOCK_TABLE_NAME="delta_lock_table"
export RUST_LOG="info,sqs-ingest=debug"

Data Flow Diagram

graph LR
    A[API/Webhook] --> B[SQS Queue]
    B --> C[sqs-ingest Lambda]
    C --> D[Delta Lake Table]
    D --> E[Analytics Queries]

Message Processing Flow

sequenceDiagram
    participant App as Application
    participant SQS as SQS Queue
    participant Lambda as sqs-ingest Lambda
    participant Delta as Delta Table

    App->>SQS: Send JSON message
    Lambda->>SQS: Poll for messages (batch)
    SQS->>Lambda: Return JSON messages
    Lambda->>Lambda: Convert JSON to rows
    Lambda->>Delta: Append rows to table
    Lambda->>SQS: Delete processed messages

Performance Optimization

Buffering and Batching

• Batch Size: Adjust BUFFER_MORE_MESSAGES to increase messages per invocation
• Message Size: Keep individual messages under 256KB for optimal performance
• Concurrency: Set Lambda reserved concurrency based on expected throughput

Monitoring Setup

• CloudWatch Metrics: Track Lambda invocations, durations, and error rates
• SQS Metrics: Monitor queue depth and message age
• Delta Monitoring: Verify table updates and query performance

Error Handling

• Visibility Timeout: Set appropriately (30-300 seconds) for retry logic
• Dead Letter Queue: Configure for failed messages requiring manual intervention
• Logging Level: Enable debug logging for troubleshooting

Multi-Table Deployment

For multiple Delta tables, deploy separate Lambda instances:

# Terraform module for sales data
module "sales_ingest" {
  source = "./sqs-ingest-module"
  function_name = "sqs-ingest-sales"
  queue_name = "sales-events.fifo"
  delta_table_uri = "s3://data/sales/"
}

# Terraform module for user events
module "users_ingest" {
  source = "./sqs-ingest-module"
  function_name = "sqs-ingest-users"
  queue_name = "user-events.fifo"
  delta_table_uri = "s3://data/users/"
}

Migrating to SQS Ingest

From Custom Lambda Functions

1. Identify data sources: Inventory existing producers
2. Set up SQS queues: Create appropriately-sized queues
3. Deploy sqs-ingest: Configure with your Delta table URI
4. Dual-write phase: Send to both old system and new SQS queues
5. Cutover: Switch fully to sqs-ingest

From Other Queue Systems

1. Create SQS queues: Match existing queue topology
2. Modify producers: Update to send JSON to SQS
3. Deploy Lambda: Configure with proper environment variables
4. Test data flow: Verify end-to-end functionality

Advanced Usage

Schema Evolution

sqs-ingest handles schema changes automatically as your JSON data evolves:

# Initial message
{
  "user_id": "user123",
  "timestamp": "2024-01-01T12:00:00Z"
}

# Later message with new field
{
  "user_id": "user123",
  "timestamp": "2024-01-01T12:00:00Z",
  "session_id": "abc123"
}

JSON Complexity Handling

The function supports nested JSON structures:

{
  "event_id": "evt123",
  "user": {
    "id": "user123",
    "attributes": {
      "preferences": {
        "theme": "dark",
        "language": "en"
      }
    }
  },
  "items": [
    {"sku": "ABC123", "quantity": 2, "price": 19.99},
    {"sku": "DEF456", "quantity": 1, "price": 24.99}
  ]
}

Security Best Practices

• IAM Configuration: Apply least-privilege permissions for SQS, S3, and DynamoDB
• Encryption: Enable SSE-KMS for SQS queues and S3 buckets
• VPC Deployment: Deploy Lambda in VPC when accessing private resources
• Input Validation: Validate JSON messages before sending to queue
• Monitoring: Set up CloudWatch alarms for security events

Conclusion

The sqs-ingest Lambda provides a direct, efficient path from JSON data sources to Delta Lake tables. By eliminating intermediate storage steps and working directly with Delta, it simplifies your data architecture while maintaining reliability and performance. This solution is particularly well-suited for event-driven applications and microservices architectures that need to reliably capture business events in Delta Lake format.