流量选品月-同环比计算对齐30天流程

Pyspark_job/dwt/dwt_flow_asin.py

@@ -168,23 +168,23 @@ class DwtFlowAsin(Templates):
                         WHEN {col_name} < '{thirty_six_month}' THEN 7 ELSE 0 END"""
 
     @staticmethod
-    def calculate_change(current_col, previous_col, use_sentinel=False):
-        rise_col = F.col(current_col) - F.col(previous_col)
-        if use_sentinel:
-            change_col = F.when(
-                F.col(current_col).isNull() & F.col(previous_col).isNull(), F.lit(None)
-            ).when(
-                F.col(current_col).isNull(), F.lit(-1000.0)
-            ).when(
-                F.col(previous_col).isNull(), F.lit(1000.0)
-            ).otherwise(
-                F.round((F.col(current_col) - F.col(previous_col)) / F.col(previous_col), 4)
-            )
-        else:
-            change_col = F.when(
-                (F.col(previous_col).isNotNull()) & (F.col(previous_col) != 0),
-                F.round((F.col(current_col) - F.col(previous_col)) / F.col(previous_col), 4)
-            ).otherwise(None)
+    def calculate_change(current_col, previous_col):
+        curr = F.when(F.col(current_col) < 0, F.lit(None)).otherwise(F.col(current_col))
+        prev = F.when(F.col(previous_col) < 0, F.lit(None)).otherwise(F.col(previous_col))
+        rise_col = curr - prev
+        change_col = F.when(
+            curr.isNull() & prev.isNull(), F.lit(None)
+        ).when(
+            curr.isNull() & (prev == 0), F.lit(0.0)
+        ).when(
+            curr.isNull(), F.lit(-1000.0)
+        ).when(
+            (prev.isNull() | (prev == 0)) & (curr == 0), F.lit(0.0)
+        ).when(
+            prev.isNull() | (prev == 0), F.lit(1000.0)
+        ).otherwise(
+            F.round((curr - prev) / prev, 4)
+        )
         return rise_col, change_col
 
     def read_data(self):
@@ -615,22 +615,22 @@ class DwtFlowAsin(Templates):
     def handle_change_rate(self):
         self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last, on=['asin'], how='left')
         self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_year, on=['asin'], how='left')
-        # (current_col, prev_col, lastyear_col, suffix, rise_round, use_sentinel)
+        # (current_col, prev_col, lastyear_col, suffix, rise_round)
         # rise_round: None 表示不计算 _rise；整数表示保留小数位；'int' 表示转 IntegerType
         columns_to_change = [
-            ("first_category_rank", "previous_first_category_rank", "lastyear_first_category_rank", "asin_rank", "int", False),
-            ("bsr_orders", "previous_bsr_orders", "lastyear_bsr_orders", "asin_bsr_orders", "int", False),
-            ("asin_rating", "previous_asin_rating", "lastyear_asin_rating", "asin_rating", 1, False),
-            ("asin_total_comments", "previous_asin_total_comments", "lastyear_asin_total_comments", "asin_comments", "int", False),
-            ("variation_num", "previous_variation_num", "lastyear_variation_num", "asin_variation", "int", False),
-            ("asin_ao_val", "previous_asin_ao_val", "lastyear_asin_ao_val", "asin_ao", 3, False),
-            ("asin_price", "previous_asin_price", "lastyear_asin_price", "asin_price", 2, False),
-            ("sales", "previous_sales", "lastyear_sales", "asin_sales", 2, False),
-            ("asin_bought_month", "previous_asin_bought_month", "lastyear_asin_bought_month", "asin_bought", None, True),
+            ("first_category_rank", "previous_first_category_rank", "lastyear_first_category_rank", "asin_rank", "int"),
+            ("bsr_orders", "previous_bsr_orders", "lastyear_bsr_orders", "asin_bsr_orders", "int"),
+            ("asin_rating", "previous_asin_rating", "lastyear_asin_rating", "asin_rating", 1),
+            ("asin_total_comments", "previous_asin_total_comments", "lastyear_asin_total_comments", "asin_comments", "int"),
+            ("variation_num", "previous_variation_num", "lastyear_variation_num", "asin_variation", "int"),
+            ("asin_ao_val", "previous_asin_ao_val", "lastyear_asin_ao_val", "asin_ao", 4),
+            ("asin_price", "previous_asin_price", "lastyear_asin_price", "asin_price", 2),
+            ("sales", "previous_sales", "lastyear_sales", "asin_sales", 2),
+            ("asin_bought_month", "previous_asin_bought_month", "lastyear_asin_bought_month", "asin_bought", None),
         ]
-        for current_col, prev_col, lastyear_col, suffix, rise_round, use_sentinel in columns_to_change:
-            rise_col, mom_col = self.calculate_change(current_col, prev_col, use_sentinel=use_sentinel)
-            _, yoy_col = self.calculate_change(current_col, lastyear_col, use_sentinel=use_sentinel)
+        for current_col, prev_col, lastyear_col, suffix, rise_round in columns_to_change:
+            rise_col, mom_col = self.calculate_change(current_col, prev_col)
+            _, yoy_col = self.calculate_change(current_col, lastyear_col)
             if rise_round is not None:
                 if rise_round == "int":
                     self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))