流量选品-变体销量趋势图

Pyspark_job/dwt/dwt_amazon_report.py

@@ -33,16 +33,17 @@ class DwtAmazonReport(Templates):
     def read_data(self):
         # 读取流量选品本月月销数据
         sql1 = f"""
-        select 
-            asin, 
-            asin_bought_month as new_monthly_sales, 
-            asin_zr_counts as new_zr_count, 
-            asin_sp_counts as new_sp_count, 
-            asin_st_counts as new_total_count, 
-            date_info as new_date_info_list 
-        from dwt_flow_asin 
-        where site_name = '{self.site_name}' 
-          and date_type = '{self.date_type}' 
+        select
+            asin,
+            asin_bought_month as new_monthly_sales,
+            asin_zr_counts as new_zr_count,
+            asin_sp_counts as new_sp_count,
+            asin_st_counts as new_total_count,
+            date_info as new_date_info_list,
+            variation_bought_month as new_variation_monthly_sales
+        from dwt_flow_asin
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
           and date_info = '{self.date_info}';
         """
         self.df_asin_detail_new = self.spark.sql(sqlQuery=sql1).repartition(15, 'asin').fillna('-1').cache()
@@ -50,24 +51,26 @@ class DwtAmazonReport(Templates):
 
         # 读历史数据
         sql2 = f"""
-        select 
-            asin, 
-            monthly_sales as old_monthly_sales, 
-            zr_count as old_zr_count, 
-            sp_count as old_sp_count, 
-            total_count as old_total_count, 
-            date_info_list as old_date_info_list 
-        from dwt_amazon_report 
-        where site_name = '{self.site_name}' 
-          and date_type = '{self.date_type}' 
+        select
+            asin,
+            monthly_sales as old_monthly_sales,
+            zr_count as old_zr_count,
+            sp_count as old_sp_count,
+            total_count as old_total_count,
+            date_info_list as old_date_info_list,
+            variation_monthly_sales as old_variation_monthly_sales
+        from dwt_amazon_report
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
           and date_info = '{self.date_info_pre}';
         """
         self.df_asin_detail_old = self.spark.sql(sqlQuery=sql2).repartition(15, 'asin').cache()
         self.df_asin_detail_old.show(10, truncate=True)
 
     def handle_data(self):
-        # 关联后的列名
-        join_columns = ['monthly_sales', 'zr_count', 'sp_count', 'total_count', 'date_info_list']
+        # 关联后的列名（新增字段放末尾）
+        join_columns = ['monthly_sales', 'zr_count', 'sp_count', 'total_count', 'date_info_list',
+                         'variation_monthly_sales']
         # 获取历史df对象中，date_info的数量，用来确定关联不到的历史asin填充多少个 -1
         old_date_info_first = self.df_asin_detail_old.select('old_date_info_list').distinct().first()
         if old_date_info_first is None: