字段类型兼容fix

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -790,8 +790,6 @@ def sync_extra_table(spark, site_name, date_info):
     依赖：dwt/dwt_flow_asin_year.py 必须已经算好并写入该分区"""
     extra_tb = f"{site_name}_flow_asin_365day_extra"
 
-    # total_appear_month / peak_month_arr 在 Hive 那边存的是逗号拼接的 STRING
-    # （dwt/dwt_flow_asin_year.py 里的说明），这里读出来转回 ARRAY<INT> 给 Doris 用
     sql = f"""
         SELECT
             asin, bought_month_total,
@@ -799,14 +797,32 @@ def sync_extra_table(spark, site_name, date_info):
             bought_month_5, bought_month_6, bought_month_7, bought_month_8,
             bought_month_9, bought_month_10, bought_month_11, bought_month_12,
             bought_month_q1, bought_month_q2, bought_month_q3, bought_month_q4,
-            CAST(SPLIT(total_appear_month, ',') AS ARRAY<INT>) AS total_appear_month,
+            total_appear_month,
             bought_month_peak,
-            CAST(SPLIT(peak_month_arr, ',') AS ARRAY<INT>) AS peak_month_arr,
+            peak_month_arr,
             is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total
         FROM dwt_flow_asin_year
         WHERE site_name = '{site_name}' AND date_info = '{date_info}'
     """
-    df_extra = spark.sql(sqlQuery=sql).repartition(20, 'asin').cache()
+    df_raw = spark.sql(sqlQuery=sql)
+
+    # total_appear_month / peak_month_arr 在 Hive 那边存的是逗号拼接的 STRING
+    # （dwt/dwt_flow_asin_year.py 里的说明），这里用 Spark 算子转回 ARRAY<INT> 再 to_json
+    # 成 "[1,2,3]"（与 img_type 处理方式一致，Doris StreamLoad ARRAY<INT> 要求）
+    df_extra = df_raw.withColumn(
+        'total_appear_month', F.split(F.col('total_appear_month'), ',')
+    ).withColumn(
+        'total_appear_month', F.expr("transform(total_appear_month, x -> cast(x as int))")
+    ).withColumn(
+        'total_appear_month', F.to_json(F.col('total_appear_month'))
+    ).withColumn(
+        'peak_month_arr', F.split(F.col('peak_month_arr'), ',')
+    ).withColumn(
+        'peak_month_arr', F.expr("transform(peak_month_arr, x -> cast(x as int))")
+    ).withColumn(
+        'peak_month_arr', F.to_json(F.col('peak_month_arr'))
+    ).repartition(20, 'asin').cache()
+
     row_count = df_extra.count()
     print(f"[同步365day_extra] 读取 Hive dwt_flow_asin_year[site_name={site_name}, date_info={date_info}]："
           f"{row_count} 条")