Python Template: Incrementally Read S3 Objects from SQS Queue as Spark DataFrame | Hands on Labs

Modern data lakehouse architectures demand efficient, scalable, and cost-effective ways to process and manage large data volumes. This guide introduces a Python-based solution that enables incremental ingestion of S3 objects into your data lakehouse using Spark, with support for table formats like Hudi, Iceberg, and Delta Lake.

Overview

This solution offers an end-to-end pipeline to process new files in S3 incrementally. It leverages AWS S3 event notifications, SQS for buffering, and Spark for processing the data as DataFrames. The processed data can be written to various table formats, including Hudi, Iceberg, or Delta Lake.

Video Guide

Key Features

• Cost Efficiency: Reduces overhead by avoiding frequent S3 list() operations.
• Scalability: Handles large data volumes seamlessly using AWS services.
• Error Handling: Supports Dead Letter Queues (DLQs) for failed events.
• Flexibility: Compatible with AWS Lambda, Glue, EMR on EC2/EKS, or EMR Serverless.

Step-by-Step Implementation

Step 1: Setup S3 Events and SQS Queue

Begin by configuring S3 to send event notifications to an SQS queue upon new file uploads. Use the provided Python script to automate the setup:

Script

This script creates the necessary S3 event notifications and SQS queue. Once configured, any new files uploaded to the specified S3 bucket will generate messages in the queue.

Step 2: Configure the Consumer

The consumer polls the SQS queue for new messages, retrieves the S3 URIs of the uploaded files, and processes them using Spark. Configure the consumer using the JSON file:

Adjust parameters like queue_url, poll_interval, and batch_size to suit your requirements.

Step 3: Run the Consumer

Once configured, start the consumer to process incoming SQS events and load S3 files into Spark as DataFrames:

Key Functions Explained

1. main Function The main function orchestrates the ETL workflow:

• Initializes the SQS poller to fetch messages.
• Creates a Spark session for data processing.
• Continuously polls for new messages, processes them using the process_batch function, and deletes processed messages.
• Includes retry logic and periodic waits for efficient resource utilization.

1. process_batch Function The process_batch function processes a batch of SQS messages:

• Extracts S3 file paths from messages.
• Loads files into Spark as DataFrames.
• Writes processed data into the target table format (Hudi, Iceberg, or Delta Lake).
• Logs metrics and handles errors gracefully.

Benefits of the Approach

1. Cost Optimization: Avoids expensive S3 list() calls by using event-driven processing.
2. Scalability: Supports massive data ingestion workflows with minimal overhead.
3. Error Management: Ensures data integrity through DLQs for failed events.
4. Flexibility: Compatible with multiple AWS services and open-source table formats.

Conclusion

This Python template simplifies the process of ingesting data incrementally from S3 into your Lakehouse architecture. By combining S3 event notifications, SQS for message buffering, and Spark for data processing, you can create a robust, efficient, and scalable pipeline that supports modern table formats like Iceberg, Hudi, and Delta Lake.

For more details, explore the GitHub repository.