流量选品-关联asin填充母体ao值、母体zr占比

Pyspark_job/dwt/dwt_flow_asin.py

@@ -337,7 +337,16 @@ class DwtFlowAsin(Templates):
         ).withColumn("asin_ao_val_type", F.expr("""CASE WHEN asin_ao_val BETWEEN 0 AND 0.1 THEN 1 
         WHEN asin_ao_val BETWEEN 0.1 AND 0.2 THEN 2 WHEN asin_ao_val BETWEEN 0.2 AND 0.4 THEN 3 
         WHEN asin_ao_val BETWEEN 0.4 AND 0.8 THEN 4 WHEN asin_ao_val BETWEEN 0.8 AND 1.2 THEN 5 
-        WHEN asin_ao_val BETWEEN 1.2 AND 2 THEN 6 WHEN asin_ao_val >= 2 THEN 7 ELSE 0 END""")).drop("asin_amazon_orders")
+        WHEN asin_ao_val BETWEEN 1.2 AND 2 THEN 6 WHEN asin_ao_val >= 2 THEN 7 ELSE 0 END""")).drop("asin_amazon_orders").cache()
+        # 关联流量asin由于来自广告位，无法计算ao值，填充母体ao值和母体zr占比
+        df_parent_asin_ao_val = self.df_asin_detail.select('parent_asin', 'matrix_ao_val', 'matrix_flow_proportion').groupby(['parent_asin']).agg(
+            F.first('matrix_ao_val').alias('parent_matrix_ao_val'), F.first('matrix_flow_proportion').alias('parent_matrix_flow_proportion')
+        )
+        self.df_asin_detail = self.df_asin_detail.join(df_parent_asin_ao_val, on=['parent_asin'], how='left').withColumn(
+            "matrix_ao_val", F.coalesce(F.col("matrix_ao_val"), F.col("parent_matrix_ao_val"))
+        ).withColumn(
+            "matrix_flow_proportion", F.coalesce(F.col("matrix_flow_proportion"), F.col("parent_matrix_flow_proportion"))
+        ).drop("parent_matrix_ao_val", "parent_matrix_flow_proportion")
         self.df_asin_measure.unpersist()
 
     # 处理配送方式、卖家所在地以及卖家所在地类型