流量选品-排名同环比默认值修复

Pyspark_job/dwt/dwt_flow_asin.py

@@ -171,7 +171,9 @@ class DwtFlowAsin(Templates):
                         WHEN {col_name} < '{thirty_six_month}' THEN 7 ELSE 0 END"""
 
     @staticmethod
-    def calculate_change(current_col, previous_col):
+    def calculate_change(current_col, previous_col, reverse=False):
+        # reverse=True：排名类指标（越小越好），哨兵值方向取反
+        up_val, down_val = (-1000.0, 1000.0) if reverse else (1000.0, -1000.0)
         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
@@ -180,11 +182,11 @@ class DwtFlowAsin(Templates):
         ).when(
             curr.isNull() & (prev == 0), F.lit(0.0)
         ).when(
-            curr.isNull(), F.lit(-1000.0)
+            curr.isNull(), F.lit(down_val)
         ).when(
             (prev.isNull() | (prev == 0)) & (curr == 0), F.lit(0.0)
         ).when(
-            prev.isNull() | (prev == 0), F.lit(1000.0)
+            prev.isNull() | (prev == 0), F.lit(up_val)
         ).otherwise(
             F.round((curr - prev) / prev, 4)
         )
@@ -686,8 +688,9 @@ class DwtFlowAsin(Templates):
             ("variation_bought_month", "previous_variation_bought_month", "lastyear_variation_bought_month", "variation_bought_month", None),
         ]
         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)
+            reverse = suffix == "asin_rank"
+            rise_col, mom_col = self.calculate_change(current_col, prev_col, reverse)
+            _, yoy_col = self.calculate_change(current_col, lastyear_col, reverse)
             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()))

Pyspark_job/dwt/dwt_flow_asin_week.py

@@ -123,7 +123,9 @@ class DwtFlowAsinWeek(Templates):
                         WHEN {col_name} < '{thirty_six_month}' THEN 7 ELSE 0 END"""
 
     @staticmethod
-    def calculate_change(current_col, previous_col):
+    def calculate_change(current_col, previous_col, reverse=False):
+        # reverse=True：排名类指标（越小越好），哨兵值方向取反
+        up_val, down_val = (-1000.0, 1000.0) if reverse else (1000.0, -1000.0)
         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
@@ -132,11 +134,11 @@ class DwtFlowAsinWeek(Templates):
         ).when(
             curr.isNull() & (prev == 0), F.lit(0.0)
         ).when(
-            curr.isNull(), F.lit(-1000.0)
+            curr.isNull(), F.lit(down_val)
         ).when(
             (prev.isNull() | (prev == 0)) & (curr == 0), F.lit(0.0)
         ).when(
-            prev.isNull() | (prev == 0), F.lit(1000.0)
+            prev.isNull() | (prev == 0), F.lit(up_val)
         ).otherwise(
             F.round((curr - prev) / prev, 4)
         )
@@ -626,8 +628,9 @@ class DwtFlowAsinWeek(Templates):
             ("variation_bought_month", "variation_bought_month", "variation_bought_month", "int"),
         ]
         for current_col, base_col, suffix, rise_round in columns_to_change:
+            reverse = suffix == "asin_rank"
             if self.can_wow:
-                rise_col, wow_col = self.calculate_change(current_col, f"lastweek_{base_col}")
+                rise_col, wow_col = self.calculate_change(current_col, f"lastweek_{base_col}", reverse)
                 if rise_round == "int":
                     rise_col = rise_col.cast(IntegerType())
                 elif rise_round is not None:
@@ -638,8 +641,8 @@ class DwtFlowAsinWeek(Templates):
                 wow_col = null_double
             if rise_round is not None:
                 self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", rise_col)
-            mom_col = F.round(self.calculate_change(current_col, f"previous_{base_col}")[1], 4) if self.can_mom else null_double
-            yoy_col = F.round(self.calculate_change(current_col, f"lastyear_{base_col}")[1], 4) if self.can_yoy else null_double
+            mom_col = F.round(self.calculate_change(current_col, f"previous_{base_col}", reverse)[1], 4) if self.can_mom else null_double
+            yoy_col = F.round(self.calculate_change(current_col, f"lastyear_{base_col}", reverse)[1], 4) if self.can_yoy else null_double
             self.df_asin_detail = self.df_asin_detail \
                 .withColumn(f"{suffix}_wow", wow_col) \
                 .withColumn(f"{suffix}_mom", mom_col) \

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -739,7 +739,9 @@ class KafkaFlowAsinDetail(Templates):
             ("asin_bought_month", "previous_asin_bought_month",   "lastyear_asin_bought_month",   "bought_month", None),
         ]
 
-        def calculate_change(current_col, previous_col):
+        def calculate_change(current_col, previous_col, reverse=False):
+            # reverse=True：排名类指标（越小越好），哨兵值方向取反
+            up_val, down_val = (-1000.0, 1000.0) if reverse else (1000.0, -1000.0)
             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
@@ -748,19 +750,20 @@ class KafkaFlowAsinDetail(Templates):
             ).when(
                 curr.isNull() & (prev == 0), F.lit(0.0)
             ).when(
-                curr.isNull(), F.lit(-1000.0)
+                curr.isNull(), F.lit(down_val)
             ).when(
                 (prev.isNull() | (prev == 0)) & (curr == 0), F.lit(0.0)
             ).when(
-                prev.isNull() | (prev == 0), F.lit(1000.0)
+                prev.isNull() | (prev == 0), F.lit(up_val)
             ).otherwise(
                 F.round((curr - prev) / prev, 4)
             )
             return rise_col, change_col
 
         for current_col, prev_col, lastyear_col, suffix, rise_round in columns_to_change:
-            rise_col, mom_col = calculate_change(current_col, prev_col)
-            _, yoy_col = calculate_change(current_col, lastyear_col)
+            reverse = suffix == "rank"
+            rise_col, mom_col = calculate_change(current_col, prev_col, reverse)
+            _, yoy_col = calculate_change(current_col, lastyear_col, reverse)
             if rise_round is not None:
                 if rise_round == "int":
                     df = df.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))