默认值fix

Pyspark_job/dwt/dwt_flow_asin_year.py

@@ -174,13 +174,20 @@ class DwtFlowAsinYear(object):
         self._handle_merge()
 
     def _handle_monthly_pivot(self):
-        """12个自然月月销透视 + 季度月销 + 总销量 + 全部出现月份数组"""
+        """12个自然月月销透视 + 季度月销 + 总销量 + 全部出现月份数组
+        （出现月份 = 该月在 dwt_flow_asin 有记录，跟 bought_month 是否有值/是否为0无关，
+        所以额外用 count(1) 透视一份"是否有记录"，而不是复用 bought_month 判空）"""
         df_pivot = self.df_last_12_month.groupBy('asin').pivot('date_info', self.last_12_month).agg(
-            F.first('bought_month').alias('bought_month')
+            F.first('bought_month').alias('bought_month'),
+            F.count(F.lit(1)).alias('appear_cnt'),
         )
         for month_str in self.last_12_month:
             month_num = int(month_str.split('-')[-1])
-            df_pivot = df_pivot.withColumnRenamed(month_str, f'bought_month_{month_num}')
+            df_pivot = df_pivot.withColumnRenamed(
+                f'{month_str}_bought_month', f'bought_month_{month_num}'
+            ).withColumnRenamed(
+                f'{month_str}_appear_cnt', f'appear_cnt_{month_num}'
+            )
 
         month_cols = [f'bought_month_{m}' for m in range(1, 13)]
         bought_month_total = reduce(
@@ -188,7 +195,7 @@ class DwtFlowAsinYear(object):
         )
         appear_month_arr = F.array_sort(F.expr(
             "filter(array(" +
-            ",".join([f"CASE WHEN bought_month_{m} IS NOT NULL THEN {m} END" for m in range(1, 13)]) +
+            ",".join([f"CASE WHEN appear_cnt_{m} IS NOT NULL AND appear_cnt_{m} > 0 THEN {m} END" for m in range(1, 13)]) +
             "), x -> x is not null)"
         ))
 
@@ -309,8 +316,10 @@ class DwtFlowAsinYear(object):
             F.col('bought_month_q4').cast(IntegerType()),
             F.concat_ws(',', F.col('total_appear_month')).alias('total_appear_month'),
             F.col('bought_month_peak').cast(IntegerType()),
-            F.concat_ws(',', F.col('peak_month_arr')).alias('peak_month_arr'),
-            F.col('is_periodic_flag').cast(IntegerType()),
+            # peak_month_arr 在该 asin 全年 bought_month 都为空时会是 NULL 数组，
+            # concat_ws 对 NULL 数组的结果是空字符串 ''，这里用 nullif 转回 NULL
+            F.expr("nullif(concat_ws(',', peak_month_arr), '')").alias('peak_month_arr'),
+            F.coalesce(F.col('is_periodic_flag'), F.lit(0)).cast(IntegerType()).alias('is_periodic_flag'),
             F.col('is_seasonal_flag').cast(IntegerType()),
             F.col('bsr_seen_count_total'),
             F.col('nsr_seen_count_total'),