年度流量选品-月销峰值月调整

Pyspark_job/dwt/dwt_flow_asin_year.py

@@ -263,6 +263,10 @@ class DwtFlowAsinYear(object):
                 ).otherwise(
                     self.udf_is_periodic(F.col('peak_month_arr'), F.col('peak_month_arr_before'))
                 )
+            ).withColumn(
+                # 峰值月超过6个月视为没有明显峰值，置为null（is_periodic_flag已用原始数组判断过，不受影响）
+                'peak_month_arr',
+                F.when(F.size('peak_month_arr') > 6, F.lit(None)).otherwise(F.col('peak_month_arr'))
             ).select(
                 'asin', 'bought_month_peak', 'peak_month_arr', 'is_periodic_flag', 'launch_time'
             ).persist(StorageLevel.DISK_ONLY)