如何将CDC加载至Redshift数据库？求Redshift的SQL CDC增量加载方法

Great questions—let's break this down step by step for Redshift CDC and incremental loading alternatives to the standard upsert.

Loading CDC (Change Data Capture) data into Redshift follows a typical staging-to-target workflow, with a few Redshift-specific considerations:

• Capture CDC Data from Your Source
  First, you need to capture change logs from your source system (like PostgreSQL, MySQL, or SaaS tools). Common tools include AWS DMS, Debezium, or custom log parsers—these tools output change records with an operation type (e.g., 'I' for insert, 'U' for update, 'D' for delete) and full/partial row data.

• Stage CDC Data in S3 (or Kinesis)
  Redshift works best with data staged in S3 (or Kinesis Data Streams for real-time loads). Save your CDC records in a columnar format like Parquet (for better compression and performance) or JSON.

• Load Staged Data to a Redshift Staging Table
  Use Redshift's COPY command to load the CDC data into a temporary staging table. Example:

  COPY staging_cdc_changes
  FROM 's3://your-cdc-bucket/daily-changes/'
  IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3Access'
  FORMAT AS PARQUET;
  

• Apply Changes to the Target Table
  Finally, use SQL to sync the staging table's changes to your production target table. This is where methods like upsert or the alternatives we'll cover next come into play.

Since you're already familiar with the MERGE (upsert) command, here are other practical approaches tailored to different use cases:

Option 1: Insert First, Then Delete (For Handling Deletes & Updates) #

This method is useful when you need to explicitly handle delete operations and want to avoid some of MERGE's performance limitations with large datasets:

1. Insert new and updated records: First, add all insert ('I') and update ('U') records from the staging table to your target. This will create duplicate rows for updated records temporarily.

   INSERT INTO target_table
   SELECT * FROM staging_cdc_changes
   WHERE op IN ('I', 'U');
   

2. Remove outdated records: Delete the old versions of updated rows, plus any rows marked for deletion ('D'), using your table's primary key to match records.

   DELETE FROM target_table
   USING staging_cdc_changes
   WHERE target_table.primary_key = staging_cdc_changes.primary_key
     AND staging_cdc_changes.op IN ('U', 'D');
   

   Pro tip: Run these steps in a transaction to avoid data inconsistency if something fails mid-process.

Option 2: Soft Delete + Historical Tracking #

If you need to retain audit trails or historical data instead of physically deleting records, use this approach:

• Add two fields to your target table: is_deleted BOOLEAN DEFAULT FALSE and last_updated TIMESTAMP DEFAULT GETDATE().
• For delete operations: Instead of removing rows, mark them as deleted and update the timestamp.

  UPDATE target_table
  SET is_deleted = TRUE, last_updated = GETDATE()
  FROM staging_cdc_changes
  WHERE target_table.primary_key = staging_cdc_changes.primary_key
    AND staging_cdc_changes.op = 'D';
  

• For updates: Either overwrite the existing row (with timestamp update) or insert a new row for full historical tracking (depending on your needs).

Option 3: Create a Fresh Target Table (For Large Batches) #

For scenarios where you process large daily increments and want to avoid locking the production table:

1. Combine the existing target table data with new CDC changes, then keep only the latest version of each record.

   CREATE TABLE new_target_table AS
   SELECT DISTINCT ON (primary_key) *
   FROM (
       SELECT * FROM target_table
       UNION ALL
       SELECT * FROM staging_cdc_changes WHERE op IN ('I', 'U')
   ) combined_data
   ORDER BY primary_key, last_updated DESC;
   

2. Swap the old and new tables to minimize downtime:

   BEGIN TRANSACTION;
   ALTER TABLE target_table RENAME TO old_target_table;
   ALTER TABLE new_target_table RENAME TO target_table;
   DROP TABLE old_target_table;
   COMMIT;
   

   This works best for non-real-time loads where you can tolerate a short swap window.

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