如何在不使用Boto的AWS Glue脚本中实现DynamoDB条件写入？

AWS Glue的write_dynamic_frame_from_options方法配合DynamoDB连接时，确实没有内置的连接选项支持条件写入（比如ConditionExpression这类参数）。如果不想使用Boto3，有一个简洁但存在局限性的实现方式——在Glue作业内预先筛选符合条件的记录，再执行写入。

方案：作业内预筛选符合条件的记录 #

核心思路是先读取目标DynamoDB表的现有数据，与待写入数据集做逻辑比对，只保留满足写入条件的记录，再写入目标表。适用于低并发、对一致性要求不高的场景，示例代码如下：

# 1. 读取目标DynamoDB表的现有数据
existing_data = glueContext.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={
        "dynamodb.input.tableName": args["OUTPUT_TABLE_NAME"]
    }
)

# 2. 将Dynamic Frame转换为Spark DataFrame，方便条件判断
from pyspark.sql import SparkSession
from awsglue.dynamicframe import DynamicFrame

spark = SparkSession.builder.getOrCreate()
write_df = SelectFromCollection_node1665510217343.toDF()
existing_df = existing_data.toDF()

# 3. 示例：仅写入目标表中不存在的主键记录（根据业务需求修改条件）
# 假设主键字段为`id`
filtered_df = write_df.join(
    existing_df,
    write_df.id == existing_df.id,
    "left_anti"  # 左反连接：保留待写入数据中未存在于目标表的记录
)

# 4. 转换回Dynamic Frame并写入DynamoDB
filtered_dyf = DynamicFrame.fromDF(filtered_df, glueContext, "filtered_dyf")
glueContext.write_dynamic_frame_from_options(
    frame=filtered_dyf,
    connection_type="dynamodb",
    connection_options={
        "dynamodb.output.tableName": args["OUTPUT_TABLE_NAME"]
    }
)

方案局限性 #

• 竞态条件风险：从读取目标表数据到完成写入的时间段内，目标表的数据可能被其他操作修改，导致最终结果不符合预期。
• 性能损耗：若目标表数据量较大，全表读取会占用大量计算资源，拖慢作业执行效率。可通过分区读取或仅读取主键等关键属性优化，但会增加实现复杂度。

补充说明 #

如果需要强一致性的原子条件写入（依赖DynamoDB服务器端的ConditionExpression做判断），目前没有绕开Boto3的简洁方式。这种场景下，需将Dynamic Frame转换为数据集，遍历每条记录调用Boto3的put_item或update_item方法，并传入对应的条件表达式参数。

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