Using Oxbow with AWS Kinesis Data Firehose

Amazon Kinesis Data Firehose is a fully managed service for delivering real-time streaming data to destinations such as Amazon S3, Redshift, Elasticsearch, and others. When used with Delta Lake, Kinesis Data Firehose provides a powerful solution for capturing high-volume streaming data and making it immediately queryable.

How the Integration Works

1. Data Firehose Configuration: Kinesis Data Firehose captures streaming data from various sources and delivers it to S3 in Apache Parquet format
2. S3 Event Triggering: When new Parquet files are written to S3, event notifications are sent to trigger processing
3. Oxbow Lambda Processing: The Oxbow Lambda function processes these Parquet files and converts them into Delta Lake table format
4. Delta Lake Table Updates: The processed data is appended to the Delta Lake table, making it immediately available for analytics

Benefits of This Architecture

• Real-time Data Availability: Streaming data becomes queryable in near real-time
• Automatic Schema Evolution: Oxbow handles schema changes automatically as your data evolves
• Scalability: Both Kinesis Data Firehose and Oxbow Lambda automatically scale to handle varying data volumes
• Cost Efficiency: Only pay for the data processed and storage used
• Data Reliability: Built-in retry mechanisms and data durability

Typical Use Cases

• Clickstream Analytics: Capture and analyze user behavior from web and mobile applications
• IoT Telemetry: Process and analyze sensor data from connected devices
• Log and Event Data: Centralize and analyze logs from distributed systems
• Real-time Monitoring: Build dashboards with up-to-the-minute business metrics
• Data Lake Ingestion: Continuously load data into your data lake for analytics

Implementation Guide

Step 1: Configure Kinesis Data Firehose

aws kinesisfirehose create-delivery-stream \
  --delivery-stream-name my-data-stream \
  --s3-destination-configuration '{
    "RoleARN": "arn:aws:iam::123456789012:role/firehose_delivery_role",
    "BucketARN": "arn:aws:s3:::my-firehose-bucket",
    "Prefix": "data/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/",
    "ErrorOutputPrefix": "errors/",
    "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 128},
    "CompressionFormat": "GZIP",
    "EncryptionConfiguration": {"NoEncryptionConfig": "NoEncryption"}
  }' \
  --extended-s3-destination-configuration '{
    "DataFormatConversionConfiguration": {
      "Enabled": true,
      "InputFormatConfiguration": {"Deserializer": {"OpenXJsonSerDe": {}}}
    },
    "ProcessingConfiguration": {"Enabled": true},
    "ParquetOutputFormat": true
  }'

Key configuration aspects:

• DataFormatConversionConfiguration.Enabled: Must be true for Parquet output
• ParquetOutputFormat: Specify Parquet as the output format
• Prefix: Organize data by date for efficient partitioning

Step 2: Deploy Oxbow Lambda

Using the Terraform configuration provided in the Oxbow repository:

cd oxbow
err=cargo lambda build --release --output-format zip --bin oxbow-lambda
err=terraform init
err=terraform apply

Step 3: Configure S3 Event Notifications

Set up S3 event notifications to trigger the Oxbow Lambda when new Parquet files are created:

resource "aws_s3_bucket_notification" "firehose_delta" {
  bucket = aws_s3_bucket.delta_tables.id

  lambda_function {
    lambda_function_arn = aws_lambda_function.oxbow.arn
    events              = ["s3:ObjectCreated:*"]
    filter_prefix       = "data/"
    filter_suffix       = ".parquet"
  }
}

Step 4: Monitor and Optimize

• Set up CloudWatch alarms for monitoring data flow
• Configure appropriate DynamoDB lock table capacity
• Monitor Lambda execution times and adjust memory allocations as needed

Architecture Diagram

graph LR
    A[Data Sources] --> B[Kinesis Data Firehose]
    B --> C[S3 Parquet Files]
    C -->|Event Notification| D[Oxbow Lambda]
    D --> E[Delta Lake Table]
    E --> F[Analytics & Reporting]

Performance Considerations

• Buffering: Configure appropriate buffer sizes (64MB-128MB) and intervals (1-5 minutes)
• Compression: Use GZIP or ZSTD compression to reduce storage costs
• Partitioning: Design S3 prefix structure to match your query patterns
• Concurrency: Monitor DynamoDB lock contention and adjust as needed
• Error Handling: Set up proper error routing and retry logic

By combining AWS Kinesis Data Firehose with Oxbow, you get a robust, scalable solution for streaming data ingestion into Delta Lake that's easy to manage and cost-effective to operate.