AWS Glue基于时间戳的数据加载及过滤方案咨询

一、加载阶段实现数据过滤（推荐） #

可以直接在Glue的连接选项中使用query参数替代dbtable，通过SQL语句在RDS端完成数据过滤，无需全表加载。这种方式能减少数据传输量，提升Pipeline效率，完全适配CDC基于时间戳增量同步的需求。

修改后的代码示例：

connection_options = {
    "connectionName": "myconnection",
    "database": "dbname"
}

# 替换为实际的上次同步时间戳，可从Glue书签或外部存储获取
last_sync_time = "2024-01-01 00:00:00"

for table in tables_list:
    # 构造带时间戳过滤的查询语句
    filter_query = f"""
        SELECT * FROM {table} 
        WHERE creation_date > '{last_sync_time}' 
           OR updation_date > '{last_sync_time}'
    """
    # 配置query参数，移除dbtable避免冲突
    connection_options['query'] = filter_query
    connection_options.pop('dbtable', None)

    datasource = glueContext.create_dynamic_frame.from_options(
        connection_type="postgresql",
        connection_options=connection_options
    )
    output_path = f"s3://your-landing-zone/{table}/"
    glueContext.write_dynamic_frame.from_options(
            frame=datasource,
            connection_type="s3",
            connection_options={"path": output_path},
            format="csv"
    )

核心逻辑：利用Glue JDBC连接支持query参数的特性，直接在RDS端执行过滤查询，只拉取符合CDC条件的增量数据，无需依赖外部VPN连接。

二、加载后过滤（可行但不推荐） #

如果因特殊原因无法在加载阶段过滤，也可以先将全表数据加载到Glue动态帧，再通过代码过滤。但这种方式会加载全量数据，占用更多VPC带宽和Glue计算资源，大表场景下性能和成本都不占优。

示例代码：

from pyspark.sql.functions import col

connection_options = {
    "connectionName": "myconnection",
    "database": "dbname"
}

last_sync_time = "2024-01-01 00:00:00"

for table in tables_list:
    connection_options['dbtable'] = table
    # 加载全表数据
    datasource = glueContext.create_dynamic_frame.from_options(
        connection_type="postgresql",
        connection_options=connection_options
    )
    # 转换为Spark DataFrame执行过滤（也可直接用DynamicFrame的filter方法）
    filtered_df = datasource.toDF().filter(
        (col("creation_date") > last_sync_time) | (col("updation_date") > last_sync_time)
    )
    # 转回DynamicFrame后写入S3
    filtered_dynamic_frame = glueContext.create_dynamic_frame.fromDF(filtered_df, glueContext, "filtered_frame")
    output_path = f"s3://your-landing-zone/{table}/"
    glueContext.write_dynamic_frame.from_options(
            frame=filtered_dynamic_frame,
            connection_type="s3",
            connection_options={"path": output_path},
            format="csv"
    )

总结 #

优先选择加载阶段通过query参数过滤的方案，既符合CDC增量同步的需求，又能优化资源使用和性能；加载后过滤仅作为特殊场景下的备选方案。

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