PySpark实现Aurora RDS表Upsert（插入更新）方案咨询

Got it, here's a practical approach to implement the upsert (update + insert) logic using a staging table with PySpark/Glue, tailored for your Aurora RDS setup:

Core Approach #

We'll leverage database-side staging tables + SQL MERGE/INSERT ON CONFLICT (depending on your Aurora engine: MySQL vs PostgreSQL) to handle the upsert efficiently. This avoids rewriting your entire ETL pipeline and leverages the database's native capabilities for atomic updates.

Step-by-Step Implementation #

1. Convert DynamicFrames to Spark DataFrames #

Glue DynamicFrames are great for catalog integration, but Spark DataFrames offer more flexibility for writing to JDBC sources:

# Convert Glue DynamicFrames to Spark DataFrames
zipterr_new_df = zipterr_new_dynamicframe.toDF()

2. Configure Aurora JDBC Connection #

Set up your database connection details (replace placeholders with your actual credentials):

# Aurora RDS JDBC configuration
jdbc_url = "jdbc:mysql://your-aurora-cluster-endpoint:3306/terr_db"  # Adjust port/engine if using PostgreSQL
jdbc_properties = {
    "user": "your_db_username",
    "password": "your_db_password",
    "driver": "com.mysql.cj.jdbc.Driver"  # Use "org.postgresql.Driver" for PostgreSQL
}

3. Write New Data to a Staging Table #

We'll create a temporary staging table in Aurora to hold the new data temporarily:

# Write new data to the staging table (overwrite if it exists)
zipterr_new_df.write \
    .mode("overwrite") \
    .jdbc(
        url=jdbc_url,
        table="zip_terr_staging",  # Temporary staging table name
        properties=jdbc_properties
    )

Note: For MySQL, temporary tables are session-scoped and auto-deleted when the connection closes. For PostgreSQL, you can explicitly create a temp table first if needed.

4. Execute Upsert Logic via SQL #

Run a native SQL statement to merge the staging table data into your target table (zip_terr_txt). Use the appropriate syntax for your Aurora engine:

Option A: For Aurora MySQL (8.0+)

Use the MERGE statement:

# Define the MERGE SQL query
merge_sql = """
MERGE INTO zip_terr_txt target
USING zip_terr_staging source
ON target.zip_code = source.zip_code  # Match on primary key
WHEN MATCHED THEN
    UPDATE SET
        territory_code = source.territory_code,
        territory_name = source.territory_name,
        state = source.state,
        business_unit_code = source.business_unit_code,
        field_force_name = source.field_force_name,
        is_active = source.is_active,
        is_deleted = source.is_deleted
WHEN NOT MATCHED THEN
    INSERT (zip_code, territory_code, territory_name, state, business_unit_code, field_force_name, is_active, is_deleted)
    VALUES (source.zip_code, source.territory_code, source.territory_name, source.state, source.business_unit_code, source.field_force_name, source.is_active, source.is_deleted);
"""

# Execute the SQL statement
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

# Get a JDBC connection to execute the merge
conn = spark._jvm.java.sql.DriverManager.getConnection(
    jdbc_url,
    jdbc_properties["user"],
    jdbc_properties["password"]
)
stmt = conn.createStatement()
stmt.execute(merge_sql)

# Clean up
stmt.close()
conn.close()

Option B: For Aurora PostgreSQL

Use INSERT ... ON CONFLICT (PostgreSQL's equivalent of upsert):

# Define the upsert SQL query
upsert_sql = """
INSERT INTO zip_terr_txt (zip_code, territory_code, territory_name, state, business_unit_code, field_force_name, is_active, is_deleted)
SELECT zip_code, territory_code, territory_name, state, business_unit_code, field_force_name, is_active, is_deleted
FROM zip_terr_staging
ON CONFLICT (zip_code) DO UPDATE SET
    territory_code = EXCLUDED.territory_code,
    territory_name = EXCLUDED.territory_name,
    state = EXCLUDED.state,
    business_unit_code = EXCLUDED.business_unit_code,
    field_force_name = EXCLUDED.field_force_name,
    is_active = EXCLUDED.is_active,
    is_deleted = EXCLUDED.is_deleted;
"""

# Execute the SQL statement (same connection code as above)
spark = SparkSession.builder.getOrCreate()
conn = spark._jvm.java.sql.DriverManager.getConnection(jdbc_url, jdbc_properties["user"], jdbc_properties["password"])
stmt = conn.createStatement()
stmt.execute(upsert_sql)
stmt.close()
conn.close()

5. Optional: Clean Up Staging Table #

If needed, manually drop the staging table to free up resources:

cleanup_sql = "DROP TABLE IF EXISTS zip_terr_staging;"
stmt = conn.createStatement()
stmt.execute(cleanup_sql)

Key Notes #

• Atomicity: The MERGE/INSERT ON CONFLICT statements are atomic, so either all updates/inserts succeed or none do—critical for production ETL.
• Performance: Using database-side operations avoids shuffling large datasets in Spark, which is more efficient for big data.
• Schema Consistency: Ensure the staging table has the exact same schema (field names, data types) as the target table to avoid errors.

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