AWS Glue：Redshift Upsert自动化实现方案咨询

Great question—this is one of the most common workarounds folks use when dealing with Redshift's lack of native MERGE/upsert support and stored procedures. Let’s break down how to automate the staging table-based upsert flow using Lambda and AWS Data Pipeline, step by step.

Core Upsert Logic (Staging Table Workflow) #

First, let’s recap the manual process we’ll automate—this is the foundation regardless of the tool:

• Stage the new data: Load your updated/insert records into a temporary or staging table (either a TEMP TABLE that auto-drops after the session, or a persistent table you truncate/reuse).
• Delete matching records: Remove any rows from your target table that have the same primary key as rows in the staging table (these are the "update" records).
• Insert new/updated rows: Add all records from the staging table into the target table.
• Clean up: Truncate or drop the staging table to free up space.

Option 1: Automate with AWS Lambda #

Lambda is perfect for event-driven or scheduled automation—great if you want to trigger upserts when new data lands in S3, or run them on a fixed schedule.

Step-by-Step Setup #

1. Configure IAM Permissions:

   • Create an IAM role for Lambda that has:
     • Redshift connect and execute permissions (to run SQL commands).
     • S3 GetObject permissions (if your staging data lives in S3).
     • CloudWatch Logs permissions (to log execution details).

2. Write the Lambda Function
   Use Python (with psycopg2-binary for Redshift connectivity) or Node.js to wrap the upsert logic. Here’s a Python example:

   import psycopg2
   import os
   
   def lambda_handler(event, context):
       # Pull Redshift credentials from environment variables (never hardcode!)
       redshift_config = {
           "host": os.environ["REDSHIFT_HOST"],
           "port": os.environ["REDSHIFT_PORT"],
           "dbname": os.environ["REDSHIFT_DB"],
           "user": os.environ["REDSHIFT_USER"],
           "password": os.environ["REDSHIFT_PASSWORD"]
       }
   
       conn = None
       try:
           # Connect to Redshift
           conn = psycopg2.connect(**redshift_config)
           cur = conn.cursor()
   
           # 1. Initialize staging table (truncate if reusing, create if missing)
           cur.execute("""
               CREATE TABLE IF NOT EXISTS staging_orders (
                   order_id INT PRIMARY KEY,
                   customer_id INT,
                   order_total NUMERIC(10,2),
                   order_date DATE
               );
               TRUNCATE TABLE staging_orders;
           """)
   
           # 2. Load new data into staging (example: copy from S3)
           cur.execute("""
               COPY staging_orders
               FROM 's3://your-data-bucket/orders/latest-orders.csv'
               IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3AccessRole'
               CSV HEADER;
           """)
   
           # 3. Delete matching rows from target table
           cur.execute("""
               DELETE FROM target_orders
               USING staging_orders
               WHERE target_orders.order_id = staging_orders.order_id;
           """)
   
           # 4. Insert all staging rows into target
           cur.execute("""
               INSERT INTO target_orders (order_id, customer_id, order_total, order_date)
               SELECT order_id, customer_id, order_total, order_date FROM staging_orders;
           """)
   
           # Commit changes and clean up
           conn.commit()
           cur.close()
           return {"statusCode": 200, "body": "Upsert completed successfully"}
   
       except Exception as e:
           print(f"Upsert failed: {str(e)}")
           if conn:
               conn.rollback()  # Undo any partial changes if something breaks
           raise e
       finally:
           if conn:
               conn.close()
   

3. Trigger the Function

   • S3 Trigger: Set up a trigger to run Lambda whenever new files are added to your S3 data bucket.
   • Scheduled Trigger: Use CloudWatch Events to run the function on a schedule (e.g., daily at 2 AM).

Option 2: Automate with AWS Data Pipeline #

Data Pipeline is better for complex workflows that depend on other tasks (e.g., ETL jobs finishing) or require more granular scheduling and error handling.

Step-by-Step Setup #

1. Create a New Pipeline:

   • Go to the AWS Data Pipeline console and start a new pipeline. Choose "Build using a template" or start from scratch.

2. Configure Redshift Connection:

   • Add a RedshiftDatabase resource to your pipeline, filling in your cluster endpoint, database name, credentials, and IAM role for access.

3. Add Upsert Activities:

   • Use a SqlActivity to run the upsert SQL steps. You can either write the SQL directly in the activity, or reference a script stored in S3.
   • Example SQL script (save to S3 and link to the activity):

     -- 1. Truncate staging table
     TRUNCATE TABLE staging_orders;
     
     -- 2. Load data from S3
     COPY staging_orders
     FROM 's3://your-data-bucket/orders/daily-orders.csv'
     IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3AccessRole'
     CSV HEADER;
     
     -- 3. Delete matching records
     DELETE FROM target_orders
     USING staging_orders
     WHERE target_orders.order_id = staging_orders.order_id;
     
     -- 4. Insert new/updated records
     INSERT INTO target_orders
     SELECT * FROM staging_orders;
     

4. Set Scheduling & Dependencies:

   • Configure the pipeline to run on your desired schedule (e.g., daily, hourly).
   • Add dependencies if needed (e.g., wait for a Glue ETL job to finish before running the upsert).

5. Monitor Execution:

   • Use the Data Pipeline console to check run status, view logs, and set up CloudWatch alarms for failures.

Key Best Practices #

• Use Transactions: Always wrap your upsert steps in a transaction (Redshift does this by default if you run multiple commands in one session, but explicitly using BEGIN/COMMIT is safer).
• Optimize Staging Tables: Use TEMP TABLE for one-off jobs (they’re faster and auto-cleanup) or persistent staging tables for repeated use (just truncate before each load).
• Handle Large Datasets: For big data volumes, use Redshift’s COPY command with parallel loading options, and consider splitting large upserts into batches to avoid locking the target table.
• Log Everything: Use CloudWatch Logs for Lambda, and Data Pipeline’s built-in logging to debug failures quickly.

内容的提问来源于stack exchange，提问作者maynard1978