流量选品月模块-兼容landing_asin字段做跳转处理

Pyspark_job/dim/dim_asin_detail.py

@@ -159,7 +159,8 @@ class DimAsinDetail(object):
             get_json_object(product_json, '$.Color') as product_json_color, 
             get_json_object(product_json, '$.Number of Items') as product_json_number_of_items, 
             current_asin,
-            nullif(coalesce(get_json_object(amazon_label, '$[0].badge_type'), get_json_object(amazon_label, '$.badge_type')), 'unknown') as amazon_label
+            nullif(coalesce(get_json_object(amazon_label, '$[0].badge_type'), get_json_object(amazon_label, '$.badge_type')), 'unknown') as amazon_label,
+            landing_asin
             from ods_asin_detail where site_name='{self.site_name}' {self.date_sql}"""
         print(sql)
         self.df_asin_detail = self.spark.sql(sqlQuery=sql)
@@ -533,7 +534,7 @@ class DimAsinDetail(object):
                     "asin_bought_month", "asin_length", "asin_width", "asin_height", "asin_is_self",
                     "customer_reviews_json", "img_list", "variat_list",
                     F.round("asin_fbm_price", 2).alias("asin_fbm_price"),
-                    "current_asin", "amazon_label", "keepa_tracking_since",
+                    "current_asin", "amazon_label", "keepa_tracking_since", "landing_asin",
                     F.lit(self.site_name).alias('site_name'),
                     F.lit(self.date_type).alias('date_type'),
                     F.lit(self.date_info).alias('date_info')).persist(StorageLevel.MEMORY_ONLY)

Pyspark_job/dwt/dwt_flow_asin.py

@@ -212,7 +212,7 @@ class DwtFlowAsin(Templates):
             date_format(created_time, 'yyyy-MM-dd HH:mm:ss') as asin_crawl_date, asin_bought_month, asin_image_view,  
             case when product_description is not null then 1 else 0 end as is_with_product_description, asin_describe, 
             category_id as top_category_id, category_first_id as top_category_first_id, customer_reviews_json, img_list as img_info, 
-            asin_follow_sellers as follow_sellers_count, asin_fbm_price, current_asin, amazon_label, variat_list
+            asin_follow_sellers as follow_sellers_count, asin_fbm_price, amazon_label, variat_list, landing_asin
             from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
         print("sql:" + sql)
         self.df_asin_detail = self.spark.sql(sqlQuery=sql)
@@ -888,12 +888,12 @@ class DwtFlowAsin(Templates):
                 print("不用导出旧数据到doris中")
 
     def handle_asin_redirect(self):
-        """asin跳转处理：将跳转asin替换为current_asin，按asin去重保留最新抓取时间"""
+        """asin跳转处理：将跳转asin替换为landing_asin，按asin去重保留最新抓取时间"""
         df_old = self.df_asin_detail
         window_redirect = Window.partitionBy('asin').orderBy(F.col('asin_crawl_date').desc_nulls_last())
         self.df_asin_detail = df_old \
-            .withColumn('asin', F.coalesce(F.col('current_asin'), F.col('asin'))) \
-            .drop('current_asin') \
+            .withColumn('asin', F.coalesce(F.col('landing_asin'), F.col('asin'))) \
+            .drop('landing_asin') \
             .withColumn('_rk', F.row_number().over(window_redirect)) \
             .filter(F.col('_rk') == 1) \
             .drop('_rk') \

Pyspark_job/sqoop_import/ods_asin_detail.py

@@ -42,7 +42,7 @@ if __name__ == '__main__':
            "sp_4stars_initial_seen_asins_json, sp_delivery_initial_seen_asins_json, compare_similar_asin_json, " \
            "together_asin_json, min_match_asin_json, variat_num, current_asin, img_list, variat_list, parent_asin, " \
            "bundles_this_asins_json, video_m3u8_url, result_list_json, bundle_asin_component_json, review_json_list, " \
-           "fbm_delivery_price, amazon_label"
+           "fbm_delivery_price, amazon_label, landing_asin"
 
     engine = get_remote_engine(
         site_name=site_name,
@@ -144,7 +144,8 @@ if __name__ == '__main__':
   REPLACE(REPLACE(REPLACE(bundle_asin_component_json, E'\n',' '), E'\r',' '), E'\t',' ') AS bundle_asin_component_json,
   REPLACE(REPLACE(REPLACE(review_json_list, E'\n',' '), E'\r',' '), E'\t',' ') AS review_json_list,
   fbm_delivery_price,
-  REPLACE(REPLACE(REPLACE(amazon_label, E'\n',' '), E'\r',' '), E'\t',' ') AS amazon_label
+  REPLACE(REPLACE(REPLACE(amazon_label, E'\n',' '), E'\r',' '), E'\t',' ') AS amazon_label,
+  REPLACE(REPLACE(REPLACE(landing_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS landing_asin
 FROM {import_table}
 WHERE  1=1 AND \$CONDITIONS
 """