流量选品模块兼容店铺新流程dim_fb_asin_info

Pyspark_job/dwt/dwt_flow_asin.py

 """
-   @Author      : wangrui
+   @Author      : CT
    @Description : 流量选品
    @SourceTable :
                   dwd_asin_measure
                   dim_asin_detail
                   ods_bsr_end
                   dim_asin_bs_category
-                  dim_fd_asin_info
+                  dim_fb_asin_info
                   dim_asin_volume
 
 
@@ -73,7 +73,7 @@ class DwtFlowAsin(Templates):
         self.df_asin_measure = self.spark.sql(f"select 1+1;")
         self.df_bsr_end = self.spark.sql(f"select 1+1;")
         self.df_asin_bs_category = self.spark.sql(f"select 1+1;")
-        self.df_fd_asin_info = self.spark.sql(f"select 1+1;")
+        self.df_fb_asin_info = self.spark.sql(f"select 1+1;")
         self.df_flow_asin_last = self.spark.sql(f"select 1+1;")
         self.df_title_matching_degree = self.spark.sql(f"select 1+1;")
         self.df_flow_asin_last_year = self.spark.sql(f"select 1+1;")
@@ -234,22 +234,22 @@ class DwtFlowAsin(Templates):
         self.df_asin_bs_category = self.spark.sql(sqlQuery=sql)
         self.df_asin_bs_category = self.df_asin_bs_category.repartition(60).persist(StorageLevel.DISK_ONLY)
         self.df_asin_bs_category.show(10, truncate=False)
-        print("5.获取dim_fd_asin_info,得到卖家相关信息")
-        if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (self.date_type == '4_week' and self.date_info >= '2024-21'):
+        print("5.获取dim_fb_asin_info,得到卖家相关信息")
+        if self.date_type == 'month' and self.date_info >= '2026-06':
             sql = f"""
-            select fd_unique as account_id, fd_account_name as account_name, upper(fd_country_name) as seller_country_name, asin, updated_at
-            from dim_fd_asin_info where site_name='{self.site_name}' and fd_unique is not null"""
+            select seller_id as account_id, account_name, upper(fb_country_name) as seller_country_name, asin, fb_crawl_date
+            from dim_fb_asin_info
+            where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}' and seller_id is not null
+            """
         else:
             sql = f"""
-            select account_id, account_name, seller_country_name, asin 
-            from (select fd_unique as account_id, fd_account_name as account_name, upper(fd_country_name) as seller_country_name, asin, 
-            ROW_NUMBER() OVER (PARTITION BY asin ORDER BY updated_at DESC) AS t_rank 
-            from dim_fd_asin_info where site_name = '{self.site_name}' and fd_unique is not null) tmp
-            where tmp.t_rank = 1
+            select seller_id as account_id, account_name, upper(fb_country_name) as seller_country_name, asin, fb_crawl_date
+            from dim_fb_asin_info
+            where site_name='{self.site_name}' and date_type='month' and date_info='2026-06' and seller_id is not null
             """
-        self.df_fd_asin_info = self.spark.sql(sqlQuery=sql)
-        self.df_fd_asin_info = self.df_fd_asin_info.repartition(60).persist(StorageLevel.DISK_ONLY)
-        self.df_fd_asin_info.show(10, truncate=False)
+        self.df_fb_asin_info = self.spark.sql(sqlQuery=sql)
+        self.df_fb_asin_info = self.df_fb_asin_info.repartition(60).persist(StorageLevel.DISK_ONLY)
+        self.df_fb_asin_info.show(10, truncate=False)
         print("6.获取环比上期整合结果")
         sql = f"""
             select asin, round(asin_ao_val, 3) as previous_asin_ao_val, asin_price as previous_asin_price,
@@ -469,34 +469,26 @@ class DwtFlowAsin(Templates):
 
     # 处理配送方式、卖家所在地以及卖家所在地类型
     def handle_seller_country(self):
-        if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (self.date_type == '4_week' and self.date_info >= '2024-21'):
         # df1: account_id + seller_country_name 去重
-            df_seller_country = self.df_fd_asin_info.select('account_id', 'seller_country_name').dropDuplicates(['account_id'])
-
+        df_seller_country = self.df_fb_asin_info.select('account_id', 'seller_country_name').dropDuplicates(['account_id'])
         # df2: asin 去重，保留最新的 account_id + account_name
-            window = Window.partitionBy('asin').orderBy(F.col('updated_at').desc())
-            df_asin_account = self.df_fd_asin_info.select('asin', 'account_id', 'account_name', 'updated_at') \
+        window = Window.partitionBy('asin').orderBy(F.col('fb_crawl_date').desc())
+        df_asin_account = self.df_fb_asin_info.select('asin', 'account_id', 'account_name', 'fb_crawl_date') \
             .withColumn('rank', F.row_number().over(window)) \
             .filter(F.col('rank') == 1) \
-                .drop('rank', 'updated_at') \
+            .drop('rank', 'fb_crawl_date') \
             .withColumnRenamed('account_id', 'fd_account_id') \
             .withColumnRenamed('account_name', 'fd_account_name')
-
         # 1. 关联df2，用于填充dim表中account_id和account_name为空的情况
         self.df_asin_detail = self.df_asin_detail.join(df_asin_account, on=['asin'], how='left')
-
         # 2. 优先使用dim表中的数据（已从seller_json提取），为空则使用df2的数据
         self.df_asin_detail = self.df_asin_detail.withColumn(
             "account_id", F.coalesce(F.col("account_id"), F.col("fd_account_id"))
         ).withColumn(
             "account_name", F.coalesce(F.col("account_name"), F.col("fd_account_name"))
         ).drop("fd_account_id", "fd_account_name")
-
         # 3. 关联df1获取seller_country_name
         self.df_asin_detail = self.df_asin_detail.join(df_seller_country, on=['account_id'], how='left')
-        else:
-            self.df_asin_detail = self.df_asin_detail.drop("account_id", "account_name")
-            self.df_asin_detail = self.df_asin_detail.join(self.df_fd_asin_info, on=['asin'], how='left')
         self.df_asin_detail = self.df_asin_detail.withColumn("asin_site_name_type", F.expr("""
         CASE WHEN asin_buy_box_seller_type = 1  THEN 4
         WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%US%' THEN 1
@@ -504,7 +496,7 @@ class DwtFlowAsin(Templates):
         WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%HK%' THEN 5
         WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%TW%' THEN 6
         ELSE 3 END"""))
-        self.df_fd_asin_info.unpersist()
+        self.df_fb_asin_info.unpersist()
 
     # 处理asin的lqs评分
     def handle_asin_lqs_rating(self):

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -866,14 +866,14 @@ class KafkaFlowAsinDetail(Templates):
         self.df_previous_flow_asin_lastyear.show(10, truncate=False)
         print("2. 获取店铺相关信息")
         sql = f"""
-            select fd_unique as seller_id, fd_account_name as account_name, upper(fd_country_name) as seller_country_name, asin, updated_at
-            from dim_fd_asin_info_30day where site_name='{self.site_name}' and date_type = '30day' and fd_unique is not null"""
+            select seller_id, account_name, upper(fb_country_name) as seller_country_name, asin, fb_crawl_date
+            from dim_fb_asin_info where site_name='{self.site_name}' and date_type = '30day' and seller_id is not null"""
         print("sql=", sql)
         self.df_seller_info = self.spark.sql(sqlQuery=sql)
         self.df_seller_info = self.df_seller_info.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
         self.df_seller_info.show(10, truncate=False)
         # df_seller_country: 按 seller_id 去重，保留 seller_country_name 非空的最新记录
-        window_seller = Window.partitionBy('seller_id').orderBy(F.col('updated_at').desc())
+        window_seller = Window.partitionBy('seller_id').orderBy(F.col('fb_crawl_date').desc())
         self.df_seller_country = self.df_seller_info \
             .filter(F.col('seller_country_name').isNotNull()) \
             .withColumn('rank', F.row_number().over(window_seller)) \
@@ -881,7 +881,7 @@ class KafkaFlowAsinDetail(Templates):
             .select('seller_id', 'seller_country_name') \
             .persist(StorageLevel.DISK_ONLY)
         # df_asin_seller: 按 asin 去重，保留最新记录
-        window_asin = Window.partitionBy('asin').orderBy(F.col('updated_at').desc())
+        window_asin = Window.partitionBy('asin').orderBy(F.col('fb_crawl_date').desc())
         self.df_asin_seller = self.df_seller_info \
             .withColumn('rank', F.row_number().over(window_asin)) \
             .filter(F.col('rank') == 1) \

Pyspark_job/script/copy_file_to_30day.py

@@ -12,16 +12,9 @@ if __name__ == '__main__':
     date_type = sys.argv[2]
     date_info = sys.argv[3]
 
-    table_name_list = ['dwt_flow_asin', 'dim_fd_asin_info', 'ods_other_search_term_data', 
-                       'dwd_asin_measure', 'ods_one_category_report', 'ods_asin_keep_date']
+    table_name_list = ['dwt_flow_asin', 'dim_fb_asin_info', 'ods_other_search_term_data',
+                       'dwd_asin_measure', 'ods_one_category_report']
     for table_name in table_name_list:
-        if table_name in ['dim_fd_asin_info', 'ods_asin_keep_date']:
-            hdfs_month = CommonUtil.build_hdfs_path(table_name, {"site_name": site_name})
-            # 复制到copy表
-            table_name = f"{table_name}_30day"
-            hdfs_30day = CommonUtil.build_hdfs_path(table_name, {"site_name": site_name, "date_type": "30day"})
-            hdfs_30day_copy = CommonUtil.build_hdfs_path(table_name, {"site_name": site_name, "date_type": "30day_copy"})
-        else:
         hdfs_month = CommonUtil.build_hdfs_path(table_name, {"site_name": site_name, "date_type": date_type, "date_info": date_info})
         hdfs_30day = CommonUtil.build_hdfs_path(table_name, {"site_name": site_name, "date_type": "30day", "date_info": "1970-01"})
         hdfs_30day_copy = CommonUtil.build_hdfs_path(table_name, {"site_name": site_name, "date_type": "30day_copy", "date_info": "1970-01"})