标题匹配度计算优化

Pyspark_job/dwt/dwt_flow_asin.py

@@ -600,16 +600,9 @@ class DwtFlowAsin(Templates):
 
     # 处理匹配度
     def handle_title_matching_degree(self):
-        window = Window.partitionBy("asin")
-        self.df_title_matching_degree = self.df_title_matching_degree.withColumn(
-            "asin_count", F.count("asin").over(window))
-        # 统计每个asin中，flag为1的数量
-        self.df_title_matching_degree = self.df_title_matching_degree.withColumn(
-            "contains_count", F.sum("contains_flag").over(window))
-        self.df_title_matching_degree = self.df_title_matching_degree.withColumn(
-            "title_matching_degree", F.round(F.col("contains_count") / F.col("asin_count"), 4))
-        self.df_title_matching_degree = self.df_title_matching_degree.drop("asin_count", "contains_count", "contains_flag")
-        self.df_title_matching_degree = self.df_title_matching_degree.drop_duplicates(['asin'])
+        self.df_title_matching_degree = self.df_title_matching_degree.groupBy('asin').agg(
+            F.round(F.sum('contains_flag') / F.count('asin'), 4).alias('title_matching_degree')
+        )
         self.df_asin_detail = self.df_asin_detail.join(self.df_title_matching_degree, on=['asin'], how='left')
         self.df_title_matching_degree.unpersist()