如何将Glue中两个write_dynamic_frame写入操作合并为单个事务？

你遇到的TypeError是因为glueContext.write_dynamic_frame.from_options()方法本身不支持transactionId直接参数，事务ID需要通过其他方式传递，或换用Spark原生事务实现原子性操作。以下是两种可行解决方案：

方案一：使用Spark原生事务（适用于多数外部数据源场景） #

Glue底层基于Spark，直接利用Spark事务API可实现多写入操作的原子性，前提是目标数据源支持ACID事务（如MySQL、PostgreSQL、Delta Lake、Iceberg等）。

代码示例 #

# 获取Glue绑定的SparkSession
spark = glueContext.spark_session

# 配置Spark事务管理器（适配V2数据源API）
spark.sql("SET spark.sql.transactionManager=org.apache.spark.sql.execution.datasources.v2.V2TransactionManager")

try:
    # 启动Spark事务
    spark.sql("START TRANSACTION")
    
    # 将DynamicFrame转为DataFrame，执行第一个写入
    df1.write.format("your_format") \
        .options(**combinedConf) \
        .mode("append")  # 根据需求选择写入模式（append/overwrite等）
        .save()
    
    # 执行第二个写入操作
    df2.write.format("your_format") \
        .options(**combinedConf) \
        .mode("append") \
        .save()
    
    # 提交事务
    spark.sql("COMMIT")
    print("所有写入操作完成，事务提交成功")
except Exception as e:
    # 回滚事务
    spark.sql("ROLLBACK")
    print(f"写入失败，事务已回滚，错误信息：{str(e)}")
    raise

方案二：针对Glue Catalog事务表传递事务ID #

如果目标是Glue Catalog管理的支持事务的表（如Glue原生ACID表、Hudi、Iceberg），可将事务ID加入connection_options字典传递，而非直接作为from_options的参数。

代码示例 #

# 启动Glue事务
tx_id = glueContext.start_transaction(read_only=False)

# 为两个写入配置添加事务ID
config1_with_tx = config1.copy()
config1_with_tx["transactionId"] = tx_id

config2_with_tx = config2.copy()
config2_with_tx["transactionId"] = tx_id

try:
    # 第一个写入操作
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(df1, glueContext, "df1"),
        connection_type="custom.spark",
        connection_options=config1_with_tx
    )
    
    # 第二个写入操作
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(df2, glueContext, "df2"),
        connection_type="custom.spark",
        connection_options=config2_with_tx
    )
    
    # 提交Glue事务
    glueContext.commit_transaction(tx_id)
    print("事务提交成功，所有写入操作完成")
except Exception as e:
    # 取消事务（回滚）
    glueContext.cancel_transaction(tx_id)
    print(f"写入失败，事务已取消，错误信息：{str(e)}")
    raise

注意事项 #

• 方案二仅适用于与Glue Catalog集成的事务型数据源，普通外部数据源（如JDBC）建议使用方案一。
• 确保你的Glue版本支持事务功能（Glue 2.0及以上版本对事务的支持更完善）。

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