在Glue中写入DynamicFrame时如何设置session_replication_role

你遇到的问题是sampleQuery参数仅适用于数据读取场景，写入操作不会执行该参数中的SQL语句，所以无法通过它设置session_replication_role。以下是两种可行的解决方案：

方案一：通过JDBC连接属性直接设置会话参数 #

PostgreSQL的JDBC驱动支持通过options参数传递会话级配置，你可以在Glue的连接选项中添加properties字段来实现：

source_connection = {
    "dbtable": table_name,
    "useConnectionProperties": "true",
    "connectionName": connection_name,
    "stringtype": "unspecified",
    "properties": {"options": "session_replication_role=replica"}
}

# 写入RDS
glue_context.write_dynamic_frame.from_options(
    frame = dyf,
    connection_type = "postgresql",
    connection_options = source_connection
)

方案二：先执行前置SQL语句再写入 #

如果方案一不生效，你可以通过Spark的JDBC连接先执行SET语句，再进行数据写入：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 从Glue连接中提取JDBC配置
connection_conf = glue_context.extract_jdbc_conf(connection_name)
jdbc_url = connection_conf["url"]
jdbc_user = connection_conf["user"]
jdbc_password = connection_conf["password"]

# 执行SET会话角色语句
spark.read.format("jdbc") \
    .option("url", jdbc_url) \
    .option("user", jdbc_user) \
    .option("password", jdbc_password) \
    .option("query", "SET session_replication_role = replica;") \
    .load()

# 执行数据写入
source_connection = {
    "dbtable": table_name,
    "useConnectionProperties": "true",
    "connectionName": connection_name,
    "stringtype": "unspecified"
}

glue_context.write_dynamic_frame.from_options(
    frame = dyf,
    connection_type = "postgresql",
    connection_options = source_connection
)

注意：第二种方法中，Spark会复用同一个JDBC连接会话，所以执行完SET语句后的写入操作会继承该会话的配置。

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