如何通过WHERE子句限制AWS Glue拉取PostgreSQL数据避免全量同步

完全可以通过以下步骤实现增量拉取，避免全量重复同步：

1. 建立作业运行时间元数据存储 #

首先需要记录每次Glue作业成功运行的时间，作为下次拉取的起始点。推荐在Redshift中创建一张元数据表：

CREATE TABLE job_last_run (
    job_name VARCHAR(100) PRIMARY KEY,
    last_success_timestamp TIMESTAMP NOT NULL
);
-- 初始化作业记录（首次运行时执行）
INSERT INTO job_last_run (job_name, last_success_timestamp)
VALUES ('your-glue-job-name', '1970-01-01 00:00:00');

2. 在Glue作业中获取上次运行时间 #

在Glue的Python脚本中，从Redshift元表读取上次成功运行的时间：

from pyspark.sql import SparkSession
import datetime

spark = SparkSession.builder.getOrCreate()

# 读取上次运行时间
last_run_df = spark.read.format("jdbc") \
    .option("url", "jdbc:redshift://your-redshift-host:5439/your-db") \
    .option("dbtable", "job_last_run") \
    .option("user", "redshift-username") \
    .option("password", "redshift-password") \
    .load()

last_run_time = last_run_df.filter(last_run_df.job_name == 'your-glue-job-name') \
    .select("last_success_timestamp") \
    .collect()[0][0]

# 转换为PostgreSQL兼容的时间字符串
last_run_str = last_run_time.strftime("%Y-%m-%d %H:%M:%S")
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

3. 构造增量查询拉取数据 #

利用Glue的JDBC连接选项，直接构造带WHERE子句的查询语句，从PostgreSQL视图中拉取上次运行后的新增/更新数据：

# 配置PostgreSQL连接参数，使用增量查询
postgres_df = spark.read.format("jdbc") \
    .option("url", "jdbc:postgresql://your-postgres-host:5432/your-db") \
    .option("query", "SELECT * FROM your_target_view WHERE update_time > ?") \
    .option("user", "postgres-username") \
    .option("password", "postgres-password") \
    .option("numPartitions", 4)  # 根据数据量调整并行度
    .option("partitionColumn", "update_time") \
    .option("lowerBound", last_run_str) \
    .option("upperBound", current_time) \
    .load()

# 转换为DynamicFrame（如果需要使用Glue的其他功能）
from awsglue.dynamicframe import DynamicFrame
incremental_dyf = DynamicFrame.fromDF(postgres_df, glueContext, "incremental_dyf")

注意：确保你的PostgreSQL视图包含可靠的时间字段（如update_time或create_time），该字段需在数据新增/更新时自动更新，否则会导致数据遗漏。如果视图没有这类字段，需修改视图逻辑，关联源表的时间字段。

4. 将增量数据写入Redshift #

使用Glue的写入逻辑将增量数据插入Redshift（选择append模式避免覆盖已有数据）：

glueContext.write_dynamic_frame.from_options(
    frame=incremental_dyf,
    connection_type="redshift",
    connection_options={
        "url": "jdbc:redshift://your-redshift-host:5439/your-db",
        "dbtable": "your-target-redshift-table",
        "user": "redshift-username",
        "password": "redshift-password",
        "redshiftTmpDir": "s3://your-temp-bucket/tmp/",
        "mode": "append"
    }
)

5. 更新作业运行时间元数据 #

仅当作业成功完成后，更新Redshift元表中的上次运行时间，避免失败导致的重复拉取或数据遗漏：

# 构造更新语句
update_query = f"""
    UPDATE job_last_run 
    SET last_success_timestamp = '{current_time}' 
    WHERE job_name = 'your-glue-job-name'
"""

# 执行更新
spark.read.format("jdbc") \
    .option("url", "jdbc:redshift://your-redshift-host:5439/your-db") \
    .option("query", update_query) \
    .option("user", "redshift-username") \
    .option("password", "redshift-password") \
    .load()

额外优化建议 #

• 避免SQL注入：优先使用参数化查询（如上述示例中的?占位符），若用字符串拼接需确保时间格式正确。
• 并行拉取优化：针对超大数据集，调整numPartitions参数，结合时间字段作为partitionColumn实现并行拉取，提升作业效率。
• 失败重试处理：结合Glue的作业重试机制，确保只有在作业完全成功时才更新元数据，避免数据不一致。

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