部分异常数据兜底处理

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -738,10 +738,10 @@ class KafkaFlowAsinDetail(Templates):
             "zr_flow_proportion",      # df_asin_measure，自然搜索流量占比
             F.round(F.col("matrix_flow_proportion"), 4).alias("matrix_flow_proportion"),  # get_asin_variant_attribute，母体自然流量占比
             F.round(F.col("matrix_ao_val"), 4).alias("matrix_ao_val"),                 # get_asin_variant_attribute，母体 ao_val
-            F.round(self._safe_double("one_two_val"),    4).alias("one_two_val"),    # Kafka，1-2 位流量占比
-            F.round(self._safe_double("three_four_val"), 4).alias("three_four_val"), # Kafka，3-4 位流量占比
-            F.round(self._safe_double("five_six_val"),   4).alias("five_six_val"),   # Kafka，5-6 位流量占比
-            F.round(self._safe_double("eight_val"),      4).alias("eight_val"),      # Kafka，8 位以上流量占比
+            F.round(F.col("one_two_val").cast("double"),    4).cast("decimal(20,4)").alias("one_two_val"),    # Kafka，1-2 位流量占比
+            F.round(F.col("three_four_val").cast("double"), 4).cast("decimal(20,4)").alias("three_four_val"), # Kafka，3-4 位流量占比
+            F.round(F.col("five_six_val").cast("double"),   4).cast("decimal(20,4)").alias("five_six_val"),   # Kafka，5-6 位流量占比
+            F.round(F.col("eight_val").cast("double"),      4).cast("decimal(20,4)").alias("eight_val"),      # Kafka，8 位以上流量占比
             # ── 分类 ───────────────────────────────────────────────────────
             "category_first_id",      # BSR 解析 + node_id 补充，一级分类 ID
             "category_id",            # BSR 解析 + node_id 补充，当前分类 ID
@@ -751,7 +751,7 @@ class KafkaFlowAsinDetail(Templates):
             # ── 商品属性 ───────────────────────────────────────────────────
             F.round(F.col("weight").cast("double"), 4).alias("weight"),  # 派生：weight_str 解析（磅）
             "volume",            # Kafka，体积原始字符串
-            "asin_weight_ratio", # 派生：体积重 / 实重比率
+            F.round(F.col("asin_weight_ratio").cast("double"), 4).cast("decimal(20,4)").alias("asin_weight_ratio"), # 派生：体积重 / 实重比率，cast decimal与月流程一致，超限/NaN自动置空
             "color",             # 变体属性 + product_json Color 补充
             "size",              # 变体属性
             "style",             # 变体属性
@@ -977,14 +977,6 @@ class KafkaFlowAsinDetail(Templates):
         df_save = self.handle_column_name(df)
         return df_save
 
-    # 字符串 → double 安全转换：拦截 'nan'/'inf'/'infinity' 等字符串
-    # cast('double') 对这类字符串会得到 Double.NaN / Double.Infinity，
-    # Doris JSON 写入时无法解析为 DECIMAL，置 null 避免整行被过滤
-    @staticmethod
-    def _safe_double(col_name):
-        c = F.col(col_name).cast("double")
-        return F.when(F.isnan(c) | (F.abs(c) == F.lit(float('inf'))), F.lit(None)).otherwise(c)
-
     @staticmethod
     def udf_rank_and_category(best_sellers_rank, pattern_str, top100_prefix):
         import re