fix

Pyspark_job/sqoop_import/ods_keepa_asin_detail.py

@@ -16,6 +16,8 @@ if __name__ == '__main__':
     # date_info = date.today().strftime("%Y-%m-%d")
     # last_date_info = (date.today() - timedelta(days=1)).strftime("%Y-%m-%d")
     last_date_info = (datetime.strptime(date_info, "%Y-%m-%d").date() - timedelta(days=1)).strftime("%Y-%m-%d")
+    # 上限：程序运行当天 00:00:00（拉昨天及之前完整数据，今天还在写不拉）
+    upper_bound = datetime.now().strftime("%Y-%m-%d") + " 00:00:00"
 
     db_type = 'postgresql_cluster'
     import_table = f"{site_name}_keepa_last_detail"
@@ -31,11 +33,12 @@ if __name__ == '__main__':
         db_type=db_type
     )
 
-    # 不再限制 update_at 上限：拉取 last_date_info 之后所有更新数据
-    # 适配间隔几天才执行的场景（间隔多日时传更早的 date_info，自动覆盖间隔区间全部增量）
+    # 增量区间：[last_date_info, 程序运行时刻)
+    # 下限：last_date_info（前一日），间隔几天补跑时传更早 date_info 自动覆盖区间
+    # 上限：upper_bound（程序运行时刻），半开区间避免拉到 PG 正在写入的最新数据，且不固定为一天
     query = f"""
     SELECT asin, last_detail::text as last_detail, update_at FROM {import_table}
-    WHERE update_at >= '{last_date_info}' AND last_detail is not null AND \$CONDITIONS
+    WHERE update_at >= '{last_date_info}' AND update_at < '{upper_bound}' AND last_detail is not null AND \$CONDITIONS
     """
 
     engine.sqoop_raw_import(