keepa无效数据补抓

Pyspark_job/script/export_asin_without_keepa.py

@@ -221,11 +221,17 @@ class ExportAsinWithoutKeepa(object):
             df = df.cache()
             print(f"筛选后数据量: {df.count()}")
 
-        # 排除 dim_keepa_asin_info 中已有 package_length 的ASIN
+        # 排除 dim_keepa_asin_info 中已有有效keepa数据的ASIN
+        # 若 package_length/width/height/weight 任意一个 < 0，视为数据异常，不排除（需重新抓取）
         print("7. 排除已有keepa数据的ASIN (dim_keepa_asin_info)")
-        df_keepa = self.spark.sql(
-            "select asin from dim_keepa_asin_info where package_length is not null"
-        ).repartition(40, 'asin')
+        df_keepa = self.spark.sql(f"""
+            select asin from dim_keepa_asin_info
+            where site_name      = '{self.site_name}'
+              and package_length >= 0
+              and package_width  >= 0
+              and package_height >= 0
+              and weight         >= 0
+        """).repartition(40, 'asin')
         df = df.join(df_keepa, on='asin', how='left_anti').cache()
         print(f"排除keepa后数据量: {df.count()}")