流量选品30天-同环比计算逻辑优化

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -646,15 +646,21 @@ class KafkaFlowAsinDetail(Templates):
         ]
 
         def calculate_change(current_col, previous_col):
-            rise_col = F.col(current_col) - F.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(
-                F.col(current_col).isNull() & F.col(previous_col).isNull(), F.lit(None)
+                curr.isNull() & prev.isNull(), F.lit(None)
             ).when(
-                F.col(current_col).isNull(), F.lit(-1000.0)
+                curr.isNull() & (prev == 0), F.lit(0.0)
             ).when(
-                F.col(previous_col).isNull() | (F.col(previous_col) == 0), F.lit(1000.0)
+                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((F.col(current_col) - F.col(previous_col)) / F.col(previous_col), 4)
+                F.round((curr - prev) / prev, 4)
             )
             return rise_col, change_col