从AWS Glue到Amazon Redshift实现Upsert：能否在Glue脚本内用临时表方案？

当然可以！这个临时表方案是Glue向Redshift实现UPSERT的常用思路，完全能在Glue脚本里端到端完成。我给你拆解下具体流程和代码示例：

核心流程概述 #

整个流程完全匹配你的期望：创建临时表 → 加载待更新数据到临时表 → 执行MERGE（合并临时表与目标表） → 清理临时表（可选，Redshift会话级临时表会自动销毁）

具体实现步骤（Python脚本示例） #

1. 初始化Glue上下文与Redshift连接配置 #

首先在脚本里初始化Glue相关上下文，同时准备好Redshift的连接信息（可以用Glue数据目录里的预配置连接，也可以手动指定）：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

# 初始化上下文
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

# Redshift连接配置示例（用Glue数据目录中的连接）
redshift_conn_name = "your-redshift-connection-name"
redshift_db = "your-database-name"
redshift_target_table = "public.your_target_table"
# 用会话级临时表的话，表名加#前缀，会话结束自动销毁
redshift_temp_table = "#staging_upsert_temp"

2. 加载待UPSERT的数据到临时表 #

不管你的数据来自S3、Glue表还是其他数据源，都可以转换成DataFrame/DynamicFrame后写入Redshift临时表：

# 示例：从S3加载待处理的UPSERT数据
source_dyf = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/upsert-dataset/"]},
    format="parquet"
)

# 转为DataFrame方便后续操作
source_df = source_dyf.toDF()

# 写入Redshift临时表
source_df.write \
    .format("jdbc") \
    .option("url", f"jdbc:redshift://your-redshift-endpoint:5439/{redshift_db}") \
    .option("dbtable", redshift_temp_table) \
    .option("user", "your-redshift-username") \
    .option("password", "your-redshift-password") \
    .option("tempdir", "s3://your-bucket/redshift-temp/")  # Redshift写入依赖的临时S3目录
    .mode("overwrite") \
    .save()

3. 执行MERGE（UPSERT）逻辑 #

通过Glue运行Redshift的MERGE语句，将临时表的数据合并到目标表：

# 构造MERGE SQL语句，替换成你的主键和字段
merge_sql = f"""
MERGE INTO {redshift_target_table} AS target
USING {redshift_temp_table} AS source
ON target.id = source.id  # 这里是主键匹配条件，按需替换
WHEN MATCHED THEN
    UPDATE SET 
        target.column1 = source.column1,
        target.column2 = source.column2  # 列出需要更新的字段
WHEN NOT MATCHED THEN
    INSERT (id, column1, column2)  # 目标表的字段列表
    VALUES (source.id, source.column1, source.column2)
"""

# 通过Spark JDBC执行MERGE语句
spark.sql(merge_sql)

4. 清理临时表（可选） #

如果没用会话级临时表（没加#前缀），记得手动删除避免残留：

drop_sql = f"DROP TABLE IF EXISTS {redshift_temp_table}"
spark.sql(drop_sql)

关键注意事项 #

• 权限配置：确保Glue执行角色拥有Redshift的读写权限，以及临时S3目录的读写权限。
• 临时表类型：优先用Redshift会话级临时表（带#前缀），无需手动清理，更安全高效。
• MERGE语法限制：Redshift的MERGE要求目标表有主键或唯一约束，且语句不能包含过于复杂的嵌套逻辑，日常UPSERT场景基本都能满足。

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