AWS Glue DynamicFrame实现Not in逻辑避免ETL重复数据写入S3

完全可以实现你要的需求，核心是通过左连接+过滤空值来模拟「Not in」的逻辑，具体步骤如下：

实现步骤 #

1. 读取源数据与目标数据
   • 从S3读取JSON格式的源数据为DynamicFrame
   • 从Glue Catalog读取目标存储桶的Parquet数据为DynamicFrame
2. 左连接两个数据集
   基于唯一键（比如user_id）对源数据（左表）和目标数据（右表）做左连接，保留所有源数据记录，匹配不到目标的记录会填充空值
3. 过滤出目标中不存在的记录
   筛选连接结果中目标侧唯一键为空的记录，这些就是源端新增的、目标中没有的数据
4. 写入目标存储桶
   将过滤后的增量数据以Parquet格式写入目标S3

代码示例 #

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

# 初始化Glue上下文
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
job.init(args['JOB_NAME'], args)

# 1. 读取源数据（S3 JSON）
source_dyf = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-source-bucket/json-data/"]},
    format="json"
)

# 2. 读取目标数据（Glue Catalog中的Parquet表）
target_dyf = glueContext.create_dynamic_frame.from_catalog(
    database="your-glue-db",
    table_name="your-target-table"
)

# 3. 基于唯一键做左连接（假设唯一键是user_id）
joined_dyf = Join.apply(
    frame1=source_dyf,
    frame2=target_dyf,
    keys1=["user_id"],
    keys2=["user_id"],
    join_type="left"
)

# 4. 过滤出目标中不存在的记录（目标侧user_id为空）
filtered_dyf = Filter.apply(
    frame=joined_dyf,
    f=lambda x: x["user_id_right"] is None
)

# 5. 移除目标侧的冗余字段（可选，保持源数据结构）
final_dyf = DropFields.apply(
    frame=filtered_dyf,
    paths=["user_id_right", "other_target_fields"]  # 替换为目标表的其他字段
)

# 6. 写入目标S3（Parquet格式，追加模式）
glueContext.write_dynamic_frame.from_options(
    frame=final_dyf,
    connection_type="s3",
    connection_options={"path": "s3://your-target-bucket/parquet-data/", "partitionKeys": []},
    format="parquet",
    transformation_ctx="write_s3"
)

job.commit()

注意事项 #

• 如果唯一键是多个字段，keys1和keys2要传入所有唯一键字段的列表，比如["order_id", "order_date"]
• 若目标数据量较大，建议通过分区裁剪（比如按日期分区读取目标数据）来减少连接的数据量，提升性能
• 写入时可以根据需求选择追加模式或覆盖模式，增量场景一般用追加

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