AWS Glue使用DynamicFrame更新数据库报错，求catalog_connection创建及方案

一、报错直接解决：补充catalog_connection参数 #

你遇到的TypeError是因为from_jdbc_conf()方法必须传入Glue Data Catalog中已创建的连接名称作为catalog_connection参数，而非直接在connection_options中填写所有JDBC信息。

1. 创建Glue Catalog连接步骤 #

• 登录AWS控制台，进入Glue服务
• 左侧导航栏选择「连接」→「添加连接」
• 选择连接类型为「JDBC」，自定义连接名称（比如my-sql-connection），点击下一步
• 填写数据库连接详情：JDBC URL、用户名、密码、驱动类（如MySQL的com.mysql.cj.jdbc.Driver），点击「测试连接」确认连通后保存

2. 修改后的代码示例 #

glueContext.write_dynamic_frame.from_jdbc_conf(
    frame=updated_dynamic_frame,
    catalog_connection="my-sql-connection",  # 填入你刚创建的连接名称
    connection_options={
        "dbtable": db_table,
        "mode": "upsert",  # 注意：Glue JDBC不支持"update"模式，用upsert替代（需数据库支持）
        "hashfield": "主键字段名",  # 指定用于判断重复的主键
        "hashpartitions": "4"  # 可选，设置分区数提升性能
    }
)

提示：Glue JDBC写入支持的模式仅为append、overwrite、upsert，其中upsert需配合主键字段实现「存在则更新，不存在则插入」的逻辑。

二、更优实现方案（针对SQL/MongoDB更新需求） #

由于Glue本身不支持直接执行UPDATE语句，以下是几种更可靠的实现方式：

方案1：用Spark原生JDBC写入（灵活度更高） #

将DynamicFrame转为Spark DataFrame，利用Spark原生JDBC支持自定义更新逻辑（以MySQL为例）：

# 转换DynamicFrame为DataFrame
updated_df = updated_dynamic_frame.toDF()

# 执行upsert逻辑
updated_df.write \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", db_table) \
    .option("user", jdbc_user) \
    .option("password", jdbc_password) \
    .option("driver", jdbc_driver) \
    .option("mode", "append") \
    .option("properties", "rewriteBatchedStatements=true") \
    .option("onDuplicateKeyUpdate", "column1=values(column1), column2=values(column2)")  # 自定义需更新的字段
    .save()

方案2：MongoDB更新专属方案 #

针对MongoDB，推荐使用MongoDB Spark Connector直接实现upsert逻辑：

# 转换为DataFrame
updated_df = updated_dynamic_frame.toDF()

# 写入MongoDB并指定upsert规则
updated_df.write \
    .format("mongodb") \
    .option("spark.mongodb.connection.uri", "mongodb://user:password@host:port/") \
    .option("spark.mongodb.database", "db_name") \
    .option("spark.mongodb.collection", "collection_name") \
    .option("spark.mongodb.write.operationType", "upsert") \
    .option("spark.mongodb.write.updateDocument", "{'$set': {'field1': ?, 'field2': ?}}") \
    .mode("append") \
    .save()

提示：需在Glue作业的「作业参数」中添加依赖，比如--additional-python-modules pymongo。

方案3：直接执行SQL更新语句 #

如果需要复杂的UPDATE逻辑，可使用Python JDBC驱动（如pyodbc）直接连接数据库执行语句：

import pyodbc

# 建立数据库连接
conn = pyodbc.connect(f"DRIVER={jdbc_driver};SERVER={server};DATABASE={db_name};UID={jdbc_user};PWD={jdbc_password}")
cursor = conn.cursor()

# 执行更新语句
update_sql = "UPDATE your_table SET column1 = ? WHERE id = ?"
cursor.execute(update_sql, (new_value, target_id))
conn.commit()

# 关闭连接
cursor.close()
conn.close()

提示：需在Glue作业中通过--additional-python-modules pyodbc安装依赖。

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