品类调研 多父级按字典去最小路径 避免同父级数据算两次

Pyspark_job/dwt/dwt_bs_top100.py

@@ -457,7 +457,22 @@ class DwtBsTop100(Templates):
             df_node_num_list2.append(df_node_num)
 
         df_nodes_num_concat2 = reduce(lambda a, b: a.unionByName(b, allowMissingColumns=True), df_node_num_list2)
-        window_spec = Window.partitionBy("asin").orderBy(F.col("_loop_order").asc())
+        # ============================================================
+        # 按 asin 去重，每个 asin 只保留一条路径 —— 保证"一个 asin 只算一次、不在祖先层双重计数"。
+        # 【为什么排序键要加 parent_id_join】分类树是 DAG：同一个"最深分类"在同一一级下可经多个中间父级到达
+        #   （us_bs_category 里有 3243 个 category_id 挂了多个父级）。上面 tree join 只用 (一级, 最深层) 两列作键，
+        #   会把这种 asin 扇成多行、各带不同 parent_id_join；这些行来自同一个 num → _loop_order 相同。
+        #   只按 _loop_order 排序时它们是"平局"，row_number 由 shuffle 顺序随机留一条 → 每次跑留下的中间路径不同
+        #   → 中间层 group 集合忽有忽无 → 落表行数每次不一样（实测同代码两次跑 18912 / 18898）。
+        # 数据本身不带中间路径、判不出"正确"父级，这里补 parent_id_join 升序做【确定性 tie-break】：
+        #   平局时永远留字典序最小的那条路径 → 每次跑结果一致、可复现（只求稳定，不改"一个 asin 只算一次"的口径）。
+        # 例：Cables(C) 同挂在 P1、P2 下 → 扇出 (E,P1,C)/(E,P2,C) 两行，parent_id_join 为 "E&&&&P1"/"E&&&&P2"
+        #     → 升序永远留 "E&&&&P1" 那条。（一级等无扇出的 asin 只有一行，此键为 null、不影响结果）
+        # ============================================================
+        window_spec = Window.partitionBy("asin").orderBy(
+            F.col("_loop_order").asc(),
+            F.col("asin_bs_cate_parent_id_join").asc_nulls_last(),  # 确定性 tie-break：平局留字典序最小路径
+        )
         # 生成行号并过滤
         self.df_nodes_num_concat2 = df_nodes_num_concat2.withColumn("rn", F.row_number().over(window_spec)) \
             .filter(F.col("rn") == 1) \
@@ -579,47 +594,44 @@ class DwtBsTop100(Templates):
             # 三个 top10 求和列 _c_asin/_c_brand/_c_seller 先 join 回 df_res，占比在 calculate_column 里 ÷ 总月销。
             # ============================================================
             orders_d = F.col("asin_amazon_orders").cast("double")
-            # ===== 原集中度逻辑(单窗口全量排序，level=1 大类几百万行挤一个 task → 倾斜卡住)。保留备查，换成下面的加盐版 =====
-            # # 商品集中度：asin 直接按月销排序取 top10
-            # w_asin = Window.partitionBy(*group_keys).orderBy(orders_d.desc_nulls_last())
-            # df_c_asin = df_level.withColumn("_rn_c", F.row_number().over(w_asin)) \
-            #     .filter(F.col("_rn_c") <= 10) \
-            #     .groupBy(group_keys).agg(F.sum(orders_d).alias("_c_asin_sales"))
-            # df_res = df_res.join(df_c_asin, on=group_keys, how="left")
-            # # 品牌集中度：先按(路径,品牌)汇总月销(排除无品牌)，再排序取 top10 品牌求和
-            # df_brand_sales = df_level.filter(F.col("asin_brand_name").isNotNull() & (F.col("asin_brand_name") != "")) \
-            #     .groupBy(*group_keys, "asin_brand_name").agg(F.sum(orders_d).alias("_brand_sales"))
-            # w_brand = Window.partitionBy(*group_keys).orderBy(F.col("_brand_sales").desc_nulls_last())
-            # df_c_brand = df_brand_sales.withColumn("_rn_c", F.row_number().over(w_brand)) \
-            #     .filter(F.col("_rn_c") <= 10) \
-            #     .groupBy(group_keys).agg(F.sum("_brand_sales").alias("_c_brand_sales"))
-            # df_res = df_res.join(df_c_brand, on=group_keys, how="left")
-            # # 卖家集中度：同品牌，把 asin_brand_name 换成 account_id(排除无卖家)
-            # df_seller_sales = df_level.filter(F.col("account_id").isNotNull() & (F.col("account_id") != "")) \
-            #     .groupBy(*group_keys, "account_id").agg(F.sum(orders_d).alias("_seller_sales"))
-            # w_seller = Window.partitionBy(*group_keys).orderBy(F.col("_seller_sales").desc_nulls_last())
-            # df_c_seller = df_seller_sales.withColumn("_rn_c", F.row_number().over(w_seller)) \
-            #     .filter(F.col("_rn_c") <= 10) \
-            #     .groupBy(group_keys).agg(F.sum("_seller_sales").alias("_c_seller_sales"))
-            # df_res = df_res.join(df_c_seller, on=group_keys, how="left")
-
-            # ===== 新：两段式加盐取 top10(见 _cr10_top_sum)，结果等价、消除倾斜 =====
-            # 先把月销 cast 成一列，三处集中度复用
-            df_level_o = df_level.withColumn("_orders_d", orders_d)
-            # 商品集中度：每行一个 asin，按月销取 top10 asin 求和
-            df_c_asin = self._cr10_top_sum(df_level_o, group_keys, "_orders_d", "_c_asin_sales")
+            # ===== 集中度 CR10：单窗口全量排序取 top10（正式版）。曾试两段式加盐版(见下方注释)消除 level=1 大类倾斜，
+            #   但实测加盐前后总耗时基本一致(~1h40m，瓶颈在拆层+8层 groupBy 不在这一步)，故保留更简单的单窗口版。 =====
+            # 商品集中度：asin 直接按月销排序取 top10
+            w_asin = Window.partitionBy(*group_keys).orderBy(orders_d.desc_nulls_last())
+            df_c_asin = df_level.withColumn("_rn_c", F.row_number().over(w_asin)) \
+                .filter(F.col("_rn_c") <= 10) \
+                .groupBy(group_keys).agg(F.sum(orders_d).alias("_c_asin_sales"))
             df_res = df_res.join(df_c_asin, on=group_keys, how="left")
-            # 品牌集中度：先按(路径,品牌)汇总月销(排除无品牌)，再取 top10 品牌求和
-            df_brand_sales = df_level_o.filter(F.col("asin_brand_name").isNotNull() & (F.col("asin_brand_name") != "")) \
-                .groupBy(*group_keys, "asin_brand_name").agg(F.sum("_orders_d").alias("_brand_sales"))
-            df_c_brand = self._cr10_top_sum(df_brand_sales, group_keys, "_brand_sales", "_c_brand_sales")
+            # 品牌集中度：先按(路径,品牌)汇总月销(排除无品牌)，再排序取 top10 品牌求和
+            df_brand_sales = df_level.filter(F.col("asin_brand_name").isNotNull() & (F.col("asin_brand_name") != "")) \
+                .groupBy(*group_keys, "asin_brand_name").agg(F.sum(orders_d).alias("_brand_sales"))
+            w_brand = Window.partitionBy(*group_keys).orderBy(F.col("_brand_sales").desc_nulls_last())
+            df_c_brand = df_brand_sales.withColumn("_rn_c", F.row_number().over(w_brand)) \
+                .filter(F.col("_rn_c") <= 10) \
+                .groupBy(group_keys).agg(F.sum("_brand_sales").alias("_c_brand_sales"))
             df_res = df_res.join(df_c_brand, on=group_keys, how="left")
-            # 卖家集中度：把品牌换成 account_id(排除无卖家)
-            df_seller_sales = df_level_o.filter(F.col("account_id").isNotNull() & (F.col("account_id") != "")) \
-                .groupBy(*group_keys, "account_id").agg(F.sum("_orders_d").alias("_seller_sales"))
-            df_c_seller = self._cr10_top_sum(df_seller_sales, group_keys, "_seller_sales", "_c_seller_sales")
+            # 卖家集中度：同品牌，把 asin_brand_name 换成 account_id(排除无卖家)
+            df_seller_sales = df_level.filter(F.col("account_id").isNotNull() & (F.col("account_id") != "")) \
+                .groupBy(*group_keys, "account_id").agg(F.sum(orders_d).alias("_seller_sales"))
+            w_seller = Window.partitionBy(*group_keys).orderBy(F.col("_seller_sales").desc_nulls_last())
+            df_c_seller = df_seller_sales.withColumn("_rn_c", F.row_number().over(w_seller)) \
+                .filter(F.col("_rn_c") <= 10) \
+                .groupBy(group_keys).agg(F.sum("_seller_sales").alias("_c_seller_sales"))
             df_res = df_res.join(df_c_seller, on=group_keys, how="left")
 
+            # ===== 加盐版(两段式取 top10，结果与单窗口等价、消除大类倾斜)保留备查；耗时无优势故不启用 =====
+            # df_level_o = df_level.withColumn("_orders_d", orders_d)
+            # df_c_asin = self._cr10_top_sum(df_level_o, group_keys, "_orders_d", "_c_asin_sales")
+            # df_res = df_res.join(df_c_asin, on=group_keys, how="left")
+            # df_brand_sales = df_level_o.filter(F.col("asin_brand_name").isNotNull() & (F.col("asin_brand_name") != "")) \
+            #     .groupBy(*group_keys, "asin_brand_name").agg(F.sum("_orders_d").alias("_brand_sales"))
+            # df_c_brand = self._cr10_top_sum(df_brand_sales, group_keys, "_brand_sales", "_c_brand_sales")
+            # df_res = df_res.join(df_c_brand, on=group_keys, how="left")
+            # df_seller_sales = df_level_o.filter(F.col("account_id").isNotNull() & (F.col("account_id") != "")) \
+            #     .groupBy(*group_keys, "account_id").agg(F.sum("_orders_d").alias("_seller_sales"))
+            # df_c_seller = self._cr10_top_sum(df_seller_sales, group_keys, "_seller_sales", "_c_seller_sales")
+            # df_res = df_res.join(df_c_seller, on=group_keys, how="left")
+
             df_res = df_res.withColumnRenamed(curr_col, "asin_bs_cate_current_id")
             if level == 1:
                 df_res = df_res.withColumn("asin_bs_cate_parent_id", F.lit("0"))