如何在AWS Glue PySpark任务中先清空PostgreSQL表再写入数据？

针对替换PostgreSQL目标表全部数据的需求，有两种常用解决方案，具体如下：

方案1：执行TRUNCATE语句清空表（推荐保留表结构） #

该方式仅清空表内数据，保留表的结构、索引、约束等信息，适合需要维持原表结构的场景。

在读取S3数据之后、写入数据库之前，添加执行TRUNCATE的代码。如果目标表已在Glue Data Catalog中注册，可直接通过Spark SQL执行：

# 清空目标表数据（替换为你的数据库和表名）
spark.sql("TRUNCATE TABLE database.table")

若需通过JDBC直接连接PostgreSQL执行（适合未在Data Catalog注册表的场景），可添加以下代码：

# 配置PostgreSQL连接参数
jdbc_url = "jdbc:postgresql://你的PostgreSQL端点:5432/数据库名"
conn_props = {
    "user": "用户名",
    "password": "密码",
    "driver": "org.postgresql.Driver"
}

# 执行TRUNCATE语句
spark.read.jdbc(url=jdbc_url, table="(TRUNCATE TABLE 目标表名) AS tmp", properties=conn_props)

修改后的完整代码示例：

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

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# 读取S3中的CSV数据
AmazonS3_node1673448010093 = glueContext.create_dynamic_frame.from_options(
    format_options={
        "quoteChar": '-1',
        "withHeader": True,
        "separator": ";",
        "optimizePerformance": False,
    },
    connection_type="s3",
    format="csv",
    connection_options={
        "paths": ["s3://folder/data.csv"],
        "recurse": True,
    },
    transformation_ctx="AmazonS3_node1673448010093"
)

# 清空目标表数据
spark.sql("TRUNCATE TABLE database.table")

# 将数据写入PostgreSQL
AWSGlueDataCatalog_node1673448012790 = glueContext.write_dynamic_frame.from_catalog(
    frame=AmazonS3_node1673448010093,
    database="database",
    table_name="table",
    transformation_ctx="AWSGlueDataCatalog_node1673448012790",
)

job.commit()

方案2：使用Overwrite模式替换全表 #

该方式会删除原表并重新创建，再写入新数据。优点是代码简洁，但会丢失原表的索引、约束、注释等结构信息，仅适合无需保留原表结构的场景。

修改写入数据的代码，添加additional_options设置writeMode为overwrite：

AWSGlueDataCatalog_node1673448012790 = glueContext.write_dynamic_frame.from_catalog(
    frame=AmazonS3_node1673448010093,
    database="database",
    table_name="table",
    additional_options={"writeMode": "overwrite"},
    transformation_ctx="AWSGlueDataCatalog_node1673448012790",
)

注意事项 #

• 权限配置：确保Glue任务的IAM角色拥有PostgreSQL的访问权限，以及执行TRUNCATE或表删除操作的权限。
• JDBC驱动：若使用JDBC直接连接PostgreSQL，需确保Glue作业已配置PostgreSQL JDBC驱动（可通过Glue作业的依赖库添加）。
• 原子性：若需要保证数据替换的原子性，可先将新数据写入临时表，再清空原表并从临时表导入数据，避免清空后写入失败导致数据丢失。

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