搜索词更换店铺dim数据源

Pyspark_job/dwt/dwt_aba_st_analytics.py

@@ -363,11 +363,11 @@ class DwtAbaStAnalytics(Templates):
 
         # 仅获取 asin和country_name,对country_name进行了聚合处理
         sql = f"""
-        select 
+        select
             asin,
-            concat_ws(\",\",collect_list(cast(fd_country_name as string))) as country_name 
-        from dim_fd_asin_info  
-        where site_name = '{self.site_name}'  
+            concat_ws(\",\",collect_list(cast(fb_country_name as string))) as country_name
+        from dim_fb_asin_info
+        where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
         group by asin;
         """
         self.df_seller_asin_country = self.spark.sql(sqlQuery=sql)
@@ -377,12 +377,12 @@ class DwtAbaStAnalytics(Templates):
 
         # 获取 dim_fd_asin_info 表
         sql = f"""
-        select 
+        select
             asin,
-            fd_unique as account_id,
-            fd_country_name as country_name 
-        from dim_fd_asin_info 
-        where site_name = '{self.site_name}' 
+            seller_id as account_id,
+            fb_country_name as country_name
+        from dim_fb_asin_info
+        where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
         """
         self.df_seller_asin_info = self.spark.sql(sqlQuery=sql)
         self.df_seller_asin_info = self.df_seller_asin_info.drop_duplicates(['asin']).repartition(80, 'asin').cache()

Pyspark_job/dwt/dwt_aba_st_analytics_report.py

@@ -326,14 +326,14 @@ class DwtAbaStAnalyticsReport(Templates):
 
         print("1.7 读取dim_seller_asin_history_info系列表")
         sql = f"""
-            select asin, 
-            case when upper(fd_country_name) = 'US' then 'US' 
-            when upper(fd_country_name) = 'CN' then 'CN' 
-            when upper(fd_country_name) = 'HK' then 'HK' 
-            when upper(fd_country_name) = 'FR' then 'FR' 
-            when upper(fd_country_name) = 'DE' then 'DE' 
+            select asin,
+            case when upper(fb_country_name) = 'US' then 'US'
+            when upper(fb_country_name) = 'CN' then 'CN'
+            when upper(fb_country_name) = 'HK' then 'HK'
+            when upper(fb_country_name) = 'FR' then 'FR'
+            when upper(fb_country_name) = 'DE' then 'DE'
             else 'OTHER' end as seller_name
-            from dim_fd_asin_info where site_name = '{self.site_name}';
+            from dim_fb_asin_info where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}';
         """
         print("sql:", sql)
         self.df_seller_asin = self.spark.sql(sql).repartition(80, 'asin').cache()