品类调研改为node_id分类

Pyspark_job/dwt/dwt_bs_top100.py

@@ -44,6 +44,11 @@ class DwtBsTop100(Templates):
         self.df_cate_tree = self.spark.sql("select 1+1;")  # DF 对象占位符
         self.df_nodes_num_concat = self.spark.sql("select 1+1;")  # DF 对象占位符
         self.df_nodes_num_concat2 = self.spark.sql("select 1+1;")  # DF 对象占位符
+        # ===== 换 node_id 归属新增的占位符 =====
+        self.df_node = self.spark.sql("select 1+1;")        # asin→node_id 重定向感知映射（来自 dim_asin_detail）
+        self.df_path = self.spark.sql("select 1+1;")        # keepa_category_path：cat_id→node_id_path 整条祖先链
+        self.df_cate_name = self.spark.sql("select 1+1;")   # keepa_category：cat_id→name 分类展示名
+        self.virtual_node_ids = []                          # 虚拟节点 cat_id 列表（运行时算出，拆层时剔除）
         self.df_level_concat = self.spark.sql("select 1+1;")  # DF 对象占位符
         self.partitions_by = ['site_name', 'date_type', 'date_info']  # 三个分区
         self.reset_partitions(partitions_num=1)  # 分区数量
@@ -66,12 +71,83 @@ class DwtBsTop100(Templates):
             # 给 DataFrame 动态添加 iteritems 方法，映射到新版 items()
             pd.DataFrame.iteritems = pd.DataFrame.items
         engine = get_remote_engine(site_name="us", db_type="mysql")
-        sql = f"SELECT nodes_num, category_id, category_parent_id, category_first_id, en_name as cur_category_name from us_bs_category where delete_time is null;"
-        df_cate = engine.read_sql(sql)
-        self.df_cate_bs_data = self.spark.createDataFrame(df_cate)
-        self.df_cate_bs_data.cache()
-        # print(self.df_cate_bs_data.show())
-        print("df_cate_bs_data数量", self.df_cate_bs_data.count())
+        # ============================================================
+        # 【旧口径：BSR 自拼树】原来读 us_bs_category（BSR 分类树），配合 handle_bs_hierarchy_to_row 自拼层级。
+        #   归属换成页面 node_id 后不再用 BSR 树，整段注释保留备查。
+        # sql = f"SELECT nodes_num, category_id, category_parent_id, category_first_id, en_name as cur_category_name from us_bs_category where delete_time is null;"
+        # df_cate = engine.read_sql(sql)
+        # self.df_cate_bs_data = self.spark.createDataFrame(df_cate)
+        # self.df_cate_bs_data.cache()
+        # print("df_cate_bs_data数量", self.df_cate_bs_data.count())
+        # ============================================================
+
+        # ============================================================
+        # 【新口径：keepa 分类树】用 node_id 做归属。两张 MySQL 新表（后端维护）：
+        #   keepa_category_path：每个 cat_id 一条，node_id_path=根到叶的 id 链（'>' 分隔，实测样本
+        #                        如 "16310091>16310161"），full_path=名称面包屑（'>' 分隔），leaf_node 1叶0非叶。
+        #   keepa_category：cat_id→name（分类展示名）。
+        # ============================================================
+        sql_path = "SELECT cat_id, node_id_path, full_path, leaf_node FROM keepa_category_path;"
+        df_path_pd = engine.read_sql(sql_path)
+        self.df_path = self.spark.createDataFrame(df_path_pd)
+        # cat_id / node_id_path 统一转 string，防 MySQL 数值型与 Spark join 时类型不一致
+        self.df_path = self.df_path.withColumn("cat_id", F.col("cat_id").cast("string")) \
+                                   .withColumn("node_id_path", F.col("node_id_path").cast("string"))
+        self.df_path.cache()
+        print("df_path(keepa_category_path)数量", self.df_path.count())
+
+        sql_name = "SELECT cat_id, name FROM keepa_category;"
+        df_name_pd = engine.read_sql(sql_name)
+        self.df_cate_name = self.spark.createDataFrame(df_name_pd)
+        self.df_cate_name = self.df_cate_name.withColumn("cat_id", F.col("cat_id").cast("string"))
+        self.df_cate_name.cache()
+        print("df_cate_name(keepa_category)数量", self.df_cate_name.count())
+
+        # ============================================================
+        # 【虚拟节点识别】node_id_path 里含约 44 个"结构性中转节点"（各部门下的
+        #   Categories/Styles/Subjects/Departments 等），Amazon 面包屑不展示。
+        # 【识别规则】某 cat_id 的 name ≠ 它自己 full_path 的末段 → 是虚拟节点
+        #   （规则与名字无关，别按 'Categories' 字面写死；虚拟节点的名字被 full_path 清掉了，
+        #    所以 name 和 full_path 末段对不上）。
+        # 【怎么用】算出虚拟 cat_id 列表 → 拆 node_id_path 时用 F.filter 保序剔掉，只留可见层级。
+        # ============================================================
+        df_vn = self.df_path.select("cat_id", "full_path").join(self.df_cate_name, on="cat_id", how="inner")
+        # full_path 末段：按 '>' 切、去两端空格，取最后一段
+        df_vn = df_vn.withColumn("_last_name",
+                                 F.trim(F.element_at(F.split(F.col("full_path"), ">"), -1)))
+        df_vn = df_vn.withColumn("_name_trim", F.trim(F.col("name")))
+        df_vn = df_vn.filter(
+            F.col("full_path").isNotNull() & (F.col("_name_trim") != F.col("_last_name"))
+        )
+        self.virtual_node_ids = [r["cat_id"] for r in df_vn.select("cat_id").distinct().collect()]
+        print(f"虚拟节点数量={len(self.virtual_node_ids)}，示例={self.virtual_node_ids[:10]}")
+
+        # ============================================================
+        # 【取 node_id：重定向感知映射，不改 dwt_flow】
+        # 为什么不直接用 dwt_flow 里的 node：dwt_flow 不带 node 列；若拿 dwt_flow 的 asin 回 join
+        #   dim_asin_detail 会踩重定向坑——dwt_flow_asin.py 的 handle_asin_redirect 把 asin 换成了
+        #   current_asin（跳转目标），而 dim_asin_detail 存的是原始 asin，直接 join 会大面积假"缺 node"。
+        # 做法：读 dim 四小列，复刻 dwt_flow 的重定向变换：key=coalesce(current_asin, asin)，
+        #   同 key 按 created_time 降序留一条，得到"重定向后 asin → node_id"，键就能和 dwt_flow 的 asin 对上。
+        # ============================================================
+        sql_node = f"""select asin, current_asin, node_id, created_time
+                       from dim_asin_detail
+                       where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
+        print(f"3. 读取 dim_asin_detail 取 node_id: sql -- {sql_node}")
+        df_node = self.spark.sql(sql_node)
+        df_node = df_node.withColumn("asin_eff", F.coalesce(F.col("current_asin"), F.col("asin")))
+        w_node = Window.partitionBy("asin_eff").orderBy(F.col("created_time").desc_nulls_last())
+        df_node = df_node.withColumn("_rk", F.row_number().over(w_node)).filter(F.col("_rk") == 1)
+        # 输出成 asin（重定向后）→ node_id，node_id 转 string 与 keepa cat_id 对齐
+        self.df_node = df_node.select(
+            F.col("asin_eff").alias("asin"),
+            F.col("node_id").cast("string").alias("node_id")
+        )
+        # 【加固】node_id 来自 dim(Hive)，若该列被推断成 double，cast string 会带尾巴 ".0"（如 "123.0"），
+        #   与 keepa 的 cat_id "123" join 不上 → 假缺 node 被误排除。这里 strip 掉 ".0"（本月无 .0，纯为未来上保险）。
+        self.df_node = self.clean_column(self.df_node, "node_id")
+        self.df_node.cache()
+        print("df_node(asin→node_id)数量", self.df_node.count())
 
 
         # 读利润率数据（dim_asin_profit_rate_info）——给每个 ASIN 补"海运/空运利润率"，
@@ -154,24 +230,31 @@ class DwtBsTop100(Templates):
 
     def handle_bse_column_rename(self):
         print("handle_bse_column_rename调用 重命名列名")
+        # ============================================================
+        # 只保留"非分类归属"的列改名：销量(bsr_orders)、BSR一级排名(first_category_rank，废弃保留字段)、
+        #   前台月销(buy_data_bought_month)。这些下游还要用。
+        # 【换 node 后注释掉的】原来还把 BSR 的 category_id/first/parent 改名成 asin_bs_cate_current_id/1_id/parent_id
+        #   当分类归属——现在归属改由 node_id → keepa 路径产出各级 id 列（见 handle_bs_node_path_to_row），
+        #   故 BSR 这三列不再参与归属，改名整段注释保留。
+        # ============================================================
         rename_dict = {
             'bsr_orders': 'asin_bsr_orders',
-            'category_id': 'asin_bs_cate_current_id',
-            'category_first_id': 'asin_bs_cate_1_id',
-            'category_parent_id': 'asin_bs_cate_parent_id',
+            # 'category_id': 'asin_bs_cate_current_id',        # 旧：BSR current 当归属，换 node 后弃用
+            # 'category_first_id': 'asin_bs_cate_1_id',        # 旧：BSR 一级当归属，换 node 后弃用
+            # 'category_parent_id': 'asin_bs_cate_parent_id',  # 旧：BSR 父级当归属，换 node 后弃用
             'first_category_rank': 'asin_bs_cate_1_rank',
             'buy_data_bought_month': 'asin_amazon_orders'
         }
         for old_col, new_col in rename_dict.items():
             self.df_bs_data = self.df_bs_data.withColumnRenamed(old_col, new_col)
-        rename_dict2 = {
-            'category_id': 'asin_bs_cate_current_id',
-            'category_first_id': 'asin_bs_cate_1_id',
-            'category_parent_id': 'asin_bs_cate_parent_id'
-        }
-        # 循环批量重命名，赋值给原变量=原数据修改
-        for old, new in rename_dict2.items():
-            self.df_cate_bs_data = self.df_cate_bs_data.withColumnRenamed(old, new)
+        # 旧：给 BSR 树表 df_cate_bs_data 改名——该表已停用，整段注释保留。
+        # rename_dict2 = {
+        #     'category_id': 'asin_bs_cate_current_id',
+        #     'category_first_id': 'asin_bs_cate_1_id',
+        #     'category_parent_id': 'asin_bs_cate_parent_id'
+        # }
+        # for old, new in rename_dict2.items():
+        #     self.df_cate_bs_data = self.df_cate_bs_data.withColumnRenamed(old, new)
 
     def handle_bs_categorical_data(self):
         print("handle_bs_categorical_data 调用")
@@ -269,6 +352,45 @@ class DwtBsTop100(Templates):
         # print(self.df_deduplicated.columns)
         # print(self.df_deduplicated.count())
 
+    def handle_bs_node_attach(self):
+        """
+        【替换 handle_bs_asin_deduplicated】给每个 asin 挂 node_id + node_id_path（换 node 归属新逻辑）。
+        产出 self.df_deduplicated：一个 asin 一行 + 它的完整祖先链 node_id_path，供 handle_bs_add_column / 拆层用。
+        """
+        print("handle_bs_node_attach 调用（node_id 归属）")
+        # ============================================================
+        # Step 1：给 asin 挂 node_id
+        # 【为什么 join on asin】df_node 的 key 已按 coalesce(current_asin, asin) 复刻了 dwt_flow 的重定向，
+        #   而 df_bs_data 的 asin 也是 dwt_flow 重定向后的 asin，两边键对得上。
+        # 【dwt_flow 是 asin 粒度】一个 asin 一行，join 后仍一个 asin 一行（df_node 也已按 asin 去重）。
+        # ============================================================
+        self.df_deduplicated = self.df_bs_data.join(self.df_node, on="asin", how="left")
+
+        # ============================================================
+        # Step 2：node_id 为空的直接排除
+        # node 空基本是图书/搜索词行（只抓了搜索结果没抓详情页 → 没页面 node、没面包屑），
+        # ============================================================
+        self.df_deduplicated = self.df_deduplicated.filter(
+            F.col("node_id").isNotNull() & (F.trim(F.col("node_id")) != "")
+        )
+
+        # ============================================================
+        # Step 3：node_id join keepa_category_path 拿整条祖先链 node_id_path
+        # ode 命中不了 keepa 树（如 keepa 未收录的冷门 node）也无法归属，一并排除。
+        # keepa_category_path 里 cat_id 唯一 → 一个 node 一条链，不会把 asin 扇成多行。
+        # 按列名等值 join，避免跨 DF 对象引用列（cat_id）触发列歧义。
+        # ============================================================
+        df_path_sel = self.df_path.select(
+            F.col("cat_id").alias("node_id"), "node_id_path", "leaf_node"
+        )
+        self.df_deduplicated = self.df_deduplicated.join(df_path_sel, on="node_id", how="left")
+        self.df_deduplicated = self.df_deduplicated.filter(
+            F.col("node_id_path").isNotNull() & (F.trim(F.col("node_id_path")) != "")
+        )
+
+        self.df_deduplicated.cache()
+        print("df_deduplicated(挂 node 后)数量", self.df_deduplicated.count())
+
     def handle_bs_add_column(self):
         print("handle_bs_add_column 调用")
         # 【利润率 join · 在 ASIN 层做一次】此时 df_deduplicated 是"一个 asin 一行
@@ -294,7 +416,9 @@ class DwtBsTop100(Templates):
                                                                       self.df_deduplicated.asin_bsr_orders).otherwise(
                                                                    F.lit(0)))
 
-        self.df_deduplicated = self.df_deduplicated.withColumn("level", self.df_deduplicated.nodes_num - 1)
+        # 旧：level = nodes_num - 1（nodes_num 来自 BSR 树表）。换 node 后无 nodes_num，且该 level 列
+        #     未进 cols_select_list、下游未用（handle_bs_level 自己按列名派生 level），整行注释保留。
+        # self.df_deduplicated = self.df_deduplicated.withColumn("level", self.df_deduplicated.nodes_num - 1)
 
         #     销售额相关字段计算
         self.df_deduplicated = self.df_deduplicated.withColumn("asin_bsr_orders_sales",
@@ -375,6 +499,72 @@ class DwtBsTop100(Templates):
     #     用 dropDuplicates 去重不确定，且结果 df_nodes_num_concat 从未被下游消费（空跑浪费）。正式逻辑见下方 concat2。
     #     如需找回：开发母本 spark_hjm/bs_top100/dwt_bs_top100.py 或 git 历史。
 
+    def handle_bs_node_path_to_row(self):
+        """
+        【替换 handle_bs_nodes_num_concat2】把 node_id_path 拆成各级分类 id 列，产出下游拆层要的宽表。
+        产出 self.df_nodes_num_concat2：一个 asin 一行 + asin_bs_cate_1_id..N_id（各级祖先 id）+ 各指标列，
+        列结构与老 concat2 完全一致，故 handle_bs_level 起下游一行不用改。
+        cat_id 唯一→无 DAG 扇出，老 concat2 里按深度分桶、tree join 回补、parent_id_join tie-break 全不需要。
+        """
+        print("handle_bs_node_path_to_row 调用（拆 node_id_path 为各级 id 列）")
+
+        # ============================================================
+        # Step 1：node_id_path 按 '>' 切成 id 数组，并剔掉空串 + 虚拟节点
+        # 【为什么剔虚拟节点】node_id_path 含 Categories/Styles 等结构性中转节点，Amazon 面包屑不展示，
+        #   剔掉后剩下的才是可见层级；剔完的 id 数组应与 full_path（可见面包屑）逐段对齐。
+        # 【为什么用 F.filter 不用 array_except】F.filter 保证保留原顺序（根→叶），array_except 不保证序。
+        # 例：node_id_path="1000>2000>9999>3000"（9999=虚拟）→ ["1000","2000","3000"]
+        # ============================================================
+        arr_raw = F.split(F.col("node_id_path"), ">")
+        if self.virtual_node_ids:
+            cate_arr = F.filter(
+                arr_raw,
+                lambda x: (F.trim(x) != "") & (~x.isin(self.virtual_node_ids))
+            )
+        else:
+            cate_arr = F.filter(arr_raw, lambda x: F.trim(x) != "")
+        df = self.df_deduplicated.withColumn("_cate_arr", cate_arr)
+
+        # ============================================================
+        # Step 2：算出本月最大层级深度，据此把数组按位置摊成 asin_bs_cate_1_id..N_id
+        # 【为什么要先求 maxd】各 asin 深度不一（有的 3 级有的 6 级），要建到全月最深的那级，浅的 asin 深层列自动为 null。
+        # 【element_at 1 基、越界返 null】asin_bs_cate_i_id = 数组第 i 个；i 超过该 asin 深度 → null（正是上卷要的：深层为空）。
+        # 【asin_bs_cate_1_id = 数组第1个 = 部门（可见一级）】剔虚拟节点后首段即部门，对齐老口径的一级。
+        # ============================================================
+        maxd = df.agg(F.max(F.size("_cate_arr")).alias("m")).first()["m"]
+        maxd = int(maxd) if maxd else 0
+        print(f"本月最大分类层级深度 maxd={maxd}")
+        for i in range(1, maxd + 1):
+            df = df.withColumn(f"asin_bs_cate_{i}_id", F.element_at(F.col("_cate_arr"), i).cast("string"))
+            # 复用老逻辑的清洗：空串/nan/null 归 None、去末尾 .0，防后续 join/分组出问题
+            df = self.clean_column(df, f"asin_bs_cate_{i}_id")
+
+        # ============================================================
+        # Step 3：只保留下游要的列（指标列 + 拆出来的各级 id 列），对齐老 concat2 的输出结构
+        # 【与老 cols_select_list 的差异】去掉了 BSR 的 asin_bs_cate_current_id / asin_bs_cate_1_id / nodes_num
+        #   （current_id/parent_id 由 handle_bs_level 每层现场派生，1_id 已由上面拆层产出）。其余指标列原样保留。
+        # ============================================================
+        metric_cols = ['asin', 'asin_amazon_orders', 'asin_bsr_orders', 'asin_bsr_orders_new',
+                       'asin_bsr_orders_brand',
+                       'asin_bsr_orders_sales', 'asin_bsr_orders_sales_new', 'asin_bsr_orders_sales_brand',
+                       'asin_bs_cate_1_rank', 'asin_price', 'asin_rating', 'asin_is_new', 'is_brand_label',
+                       'asin_total_comments',
+                       'launch_time_type0_rate', 'launch_time_type1_rate', 'launch_time_type2_rate',
+                       'launch_time_type3_rate', 'launch_time_type4_rate',
+                       'launch_time_type5_rate', 'launch_time_type6_rate', 'launch_time_type7_rate',
+                       'asin_buy_box_seller_type0_rate', 'asin_buy_box_seller_type1_rate',
+                       'asin_buy_box_seller_type2_rate', 'asin_buy_box_seller_type3_rate',
+                       'asin_buy_box_seller_type4_rate',
+                       'seller_country_us_rate', 'seller_country_cn_rate', 'seller_country_other_rate',
+                       'asin_orders_ge_50_rate', 'asin_orders_ge_100_rate', 'asin_orders_ge_200_rate',
+                       'asin_brand_name', 'account_id',
+                       'asin_ocean_freight_gross_margin', 'asin_air_freight_gross_margin']
+        metric_cols = metric_cols + self.band_margin_cols + self.pkg_band_cols
+        id_cols = [f"asin_bs_cate_{i}_id" for i in range(1, maxd + 1)]
+        self.df_nodes_num_concat2 = df.select(*(metric_cols + id_cols))
+        self.df_nodes_num_concat2.cache()
+        print("df_nodes_num_concat2(拆层后)列名", self.df_nodes_num_concat2.columns)
+
     def handle_bs_nodes_num_concat2(self):
         print("handle_bs_nodes_num_concat2 打印")
         self.df_deduplicated2 = self.df_deduplicated.fillna({"nodes_num": -1})
@@ -632,20 +822,22 @@ class DwtBsTop100(Templates):
             # 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")
+            # 换 node 后输出列名从 asin_bs_cate_* 改为 asin_node_*（值是页面 node id，不再是 BSR 分类 id）。
+            #   注意：内部级列 asin_bs_cate_{i}_id 仍沿用旧名（只是中间分组列，不落表），故这里只改输出的 3 个列名。
+            df_res = df_res.withColumnRenamed(curr_col, "asin_node_current_id")
             if level == 1:
-                df_res = df_res.withColumn("asin_bs_cate_parent_id", F.lit("0"))
-                df_res = df_res.withColumn("asin_bs_cate_parent_id_join", F.lit("0"))
+                df_res = df_res.withColumn("asin_node_parent_id", F.lit("0"))
+                df_res = df_res.withColumn("asin_node_parent_id_join", F.lit("0"))
             else:
                 # 面包屑 = 0&&&&一级&&&&…&&&&直接父级，用"一级..上一层"祖先列现场拼。
                 # 必须在 rename parent_col 之前拼（此时 asin_bs_cate_{level-1}_id 还在），拼完再 rename。
                 ancestor_join_cols = [f"asin_bs_cate_{i}_id" for i in range(1, level)]  # 1..level-1
                 df_res = df_res.withColumn(
-                    "asin_bs_cate_parent_id_join",
+                    "asin_node_parent_id_join",
                     F.concat(F.lit("0&&&&"),
                              F.concat_ws("&&&&", *[F.col(c).cast("string") for c in ancestor_join_cols]))
                 )
-                df_res = df_res.withColumnRenamed(parent_col, "asin_bs_cate_parent_id")
+                df_res = df_res.withColumnRenamed(parent_col, "asin_node_parent_id")
                 # 删掉分组用剩的中间祖先列（1..level-2），只保留 current_id/parent_id/parent_id_join，
                 # 否则它们会随 unionByName 混进最终结果，污染列结构。
                 for _c in [f"asin_bs_cate_{i}_id" for i in range(1, level - 1)]:
@@ -663,15 +855,34 @@ class DwtBsTop100(Templates):
         # print("self.df_level_concat",self.df_level_concat.count())
         # print(self.df_level_concat.columns)
         #     增加分类名称
-        #  # 筛选self.df_cate_bs_data指定两列的数据 以对asin_bs_cate_current_id去重
-        print("增加分类名称")
-        df_cate_name =  self.df_cate_bs_data.select(*['asin_bs_cate_current_id', 'cur_category_name','_cate_order'])
-        window_spec = Window.partitionBy('asin_bs_cate_current_id').orderBy(F.col("_cate_order").asc_nulls_last())
-        # _cate_order 仅用于上面 window 去重排序，用完 drop
-        df_cate_name = df_cate_name.withColumn("_rn",F.row_number().over(window_spec)).filter(F.col("_rn") == 1).drop("_rn", "_cate_order")
-        # inner join 补分类名：current_id 不在 us_bs_category 的行会被丢。实测仅丢 2 个非标准一级
-        # (private-brands / book-special-features 之类，树里无对应 category_id、页面也不查) → 保留 inner，无害。
-        self.df_level_concat  =self.df_level_concat.join(df_cate_name, on='asin_bs_cate_current_id',how="inner")
+        print("增加分类名称（keepa_category.name）")
+        # ============================================================
+        # 换 node 后分类名从 keepa_category 取：当前层 id(asin_node_current_id)=keepa cat_id → name。
+        # 【旧逻辑】原从 BSR 树表 df_cate_bs_data 取 cur_category_name，inner join（会丢 2 个非标准一级）。整段替换。
+        # 【为什么改 inner 为 left】现在 current_id 都来自 keepa 树，正常都能在 keepa_category 查到名；
+        #   用 left 兜底：万一个别 id 查不到名也不丢整条分类行（cur_category_name 是废弃展示字段、后端不查，
+        #   为空无害；但分类行本身的各项统计不能因缺名被丢）。
+        # ============================================================
+        df_cate_name = self.df_cate_name.select(
+            F.col("cat_id").alias("asin_node_current_id"),
+            F.col("name").alias("cur_category_name")
+        ).dropDuplicates(["asin_node_current_id"])
+        self.df_level_concat = self.df_level_concat.join(df_cate_name, on='asin_node_current_id', how="left")
+
+        # ============================================================
+        # 增加 leaf_node（是否叶子分类）：直接取 keepa_category_path.leaf_node（1=叶子/最细分类，0=还有子分类）。
+        # 【用途】前端"默认显示最小节点(叶子)分类"时按 leaf_node=1 筛。
+        # 【为什么 left + coalesce 0】current_id 都来自 keepa 树、正常都能匹配上；万一没匹配到（理论不会）
+        #   补 0（当非叶子），避免落 null（Java 侧读 NULL 有问题，见项目约定）。
+        # ============================================================
+        df_leaf = self.df_path.select(
+            F.col("cat_id").alias("asin_node_current_id"),
+            F.col("leaf_node")
+        ).dropDuplicates(["asin_node_current_id"])
+        self.df_level_concat = self.df_level_concat.join(df_leaf, on='asin_node_current_id', how="left")
+        self.df_level_concat = self.df_level_concat.withColumn(
+            "leaf_node", F.coalesce(F.col("leaf_node").cast("int"), F.lit(0))
+        )
         print("增加分类名称 join后的数据")
         # self.df_level_concat.show(10,truncate=False )
         # print(self.df_level_concat.count())
@@ -898,7 +1109,7 @@ class DwtBsTop100(Templates):
         #   parent_id_join 形如 "0&&&&electronics&&&&Cables"，唯一标识"某条路径下的这个父级"，只有同路径
         #   的兄弟才进同一分母。level=1(值"0")、level=2(值"0&&&&一级")的分组与按直接父级 id 完全等价，
         #   只有 level≥3 的多归属分类被修正——和 handle_bs_level 的拆行改动对齐。
-        window_spec = Window.partitionBy("asin_bs_cate_parent_id_join")
+        window_spec = Window.partitionBy("asin_node_parent_id_join")
         # 3. 计算父节点总订单数
         # 这一步相当于原来的: agg(sum("asin_bsr_orders_sum")) + join
         # 效果：每一行都会多出一列 'asin_bsr_orders_sum_parent'，
@@ -950,13 +1161,22 @@ class DwtBsTop100(Templates):
             self.df_save = self.df_save.withColumn(col_name, F.round(F.col(col_name) / F.col("asin_count"), 4))
 
     def handle_data(self):
-        self.handle_bs_redirect_flag()
-        self.handle_bs_hierarchy_to_row()
+        # ============================================================
+        # 归属换 node_id 后的编排：
+        #   旧的两步（BSR 树去重 + 自拼树）不再需要，注释保留；
+        #   asin_deduplicated（BSR 树 join 取最深）换成 node_attach（挂 node_id + node_id_path、空 node 排除）；
+        #   nodes_num_concat2（按深度分桶 + tree join 回补）换成 node_path_to_row（拆 node_id_path 为各级 id 列）。
+        #   handle_bs_level 及其后全不动。
+        # ============================================================
+        # self.handle_bs_redirect_flag()     # 旧：BSR 树去重，换 node 后无树，停用（方法保留备查）
+        # self.handle_bs_hierarchy_to_row()  # 旧：自拼 BSR 层级树，换 node_id_path 后不需要，停用（方法保留备查）
         self.handle_bse_column_rename()
         self.handle_bs_categorical_data()
-        self.handle_bs_asin_deduplicated()
+        # self.handle_bs_asin_deduplicated() # 旧：BSR (一级,current) join us_bs_category 取最深，停用（方法保留备查）
+        self.handle_bs_node_attach()         # 新：挂 node_id + node_id_path，空 node 排除
         self.handle_bs_add_column()
-        self.handle_bs_nodes_num_concat2()
+        # self.handle_bs_nodes_num_concat2() # 旧：按 nodes_num 分桶 + tree join 回补祖先，停用（方法保留备查）
+        self.handle_bs_node_path_to_row()    # 新：拆 node_id_path → asin_bs_cate_1_id..N_id
         self.handle_bs_level()
         self.handle_bs_level_calculate_column()
         self.handle_bs_level_proportion2()
@@ -970,7 +1190,7 @@ class DwtBsTop100(Templates):
         #   + 打包占比7 + 卖家地占比3 + 月销阈值占比3 + 集中度3 = 86 列，再 + 3 分区列。
         # ============================================================
         business_cols = [
-            "asin_bs_cate_current_id",
+            "asin_node_current_id",
             "asin_amazon_orders_sum", "asin_bsr_orders_sum", "asin_bsr_orders_sum_new", "asin_bsr_orders_sum_brand",
             "asin_bsr_orders_sales_sum", "asin_bsr_orders_sales_sum_new", "asin_bsr_orders_sales_sum_brand",
             "asin_count_new", "asin_count_brand", "asin_count",
@@ -979,7 +1199,7 @@ class DwtBsTop100(Templates):
             "launch_time_type4_rate", "launch_time_type5_rate", "launch_time_type6_rate", "launch_time_type7_rate",
             "asin_buy_box_seller_type0_rate", "asin_buy_box_seller_type1_rate", "asin_buy_box_seller_type2_rate",
             "asin_buy_box_seller_type3_rate", "asin_buy_box_seller_type4_rate",
-            "asin_bs_cate_parent_id", "asin_bs_cate_parent_id_join", "level", "cur_category_name",
+            "asin_node_parent_id", "asin_node_parent_id_join", "level", "cur_category_name", "leaf_node",
             "asin_count_rate_new", "asin_bsr_orders_sum_rate_new", "asin_count_rate_brand", "asin_bsr_orders_sum_rate_brand",
             "asin_bsr_orders_sales_mean", "asin_bsr_orders_sales_mean_new", "asin_bsr_orders_sales_mean_brand",
             "asin_bsr_orders_mean", "asin_bsr_orders_mean_new", "asin_bsr_orders_mean_brand",

Pyspark_job/dwt/dwt_bs_top100_change_rate.py

@@ -52,7 +52,7 @@ class DwtBsTop100ChangeRate(object):
 
     # 跨月对齐主键：分类多归属（同 current_id 挂多个父级路径=多行），只用 current_id 会 1:N 串行。
     # current_id + parent_id_join（根→直接父级的完整路径）= 每行唯一（已实测 2026-05 分区无重复）。
-    JOIN_KEYS = ['asin_bs_cate_current_id', 'asin_bs_cate_parent_id_join']
+    JOIN_KEYS = ['asin_node_current_id', 'asin_node_parent_id_join']
 
     def __init__(self, site_name, date_type, date_info):
         self.site_name = site_name
@@ -101,10 +101,10 @@ class DwtBsTop100ChangeRate(object):
         return df
 
     def read_data(self):
-        # 本期额外带 level（落表用，便于后端/排查），对比期只要指标值
+        # 本期额外带 level + leaf_node（落表用，便于后端/前端按叶子筛），对比期只要指标值
         metric_cols = [m for m, _ in self.METRICS]
         base_sql = f"""
-            select {', '.join(self.JOIN_KEYS + ['level'] + metric_cols)}
+            select {', '.join(self.JOIN_KEYS + ['level', 'leaf_node'] + metric_cols)}
             from {self.source_tb}
             where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'
         """
@@ -181,7 +181,7 @@ class DwtBsTop100ChangeRate(object):
         for m, _ in self.METRICS:
             rate_cols.append(f"{m}_ym_rate")
             rate_cols.append(f"{m}_lm_rate")
-        out_cols = self.JOIN_KEYS + ['level'] + rate_cols
+        out_cols = self.JOIN_KEYS + ['level', 'leaf_node'] + rate_cols
 
         self.df_save = self.df_save.select(*out_cols) \
             .withColumn("site_name", F.lit(self.site_name)) \

Pyspark_job/sqoop_export/dwt_bs_top100.py

@@ -78,8 +78,8 @@ if __name__ == '__main__':
     # 导出列 = 86 业务列 + 3 分区列（列名与 Hive dwt_bs_top100 一致, sqoop 按名映射 PG 同名列）
     export_cols = [
         # 维度 / 键
-        "asin_bs_cate_current_id", "asin_bs_cate_parent_id", "asin_bs_cate_parent_id_join",
-        "level", "cur_category_name",
+        "asin_node_current_id", "asin_node_parent_id", "asin_node_parent_id_join",
+        "level", "cur_category_name", "leaf_node",
         # 销量类
         "asin_amazon_orders_sum", "asin_bsr_orders_sum", "asin_bsr_orders_sum_new", "asin_bsr_orders_sum_brand",
         "asin_bsr_orders_mean", "asin_bsr_orders_mean_new", "asin_bsr_orders_mean_brand", "asin_bsr_orders_sum_parent",

Pyspark_job/sqoop_export/dwt_bs_top100_change_rate.py

@@ -78,7 +78,7 @@ if __name__ == '__main__':
 
     # 导出列 = join键2 + level + 28 变化率列 + 3 分区列（列名与 Hive dwt_bs_top100_change_rate 一致）
     export_cols = [
-        "asin_bs_cate_current_id", "asin_bs_cate_parent_id_join", "level",
+        "asin_node_current_id", "asin_node_parent_id_join", "level", "leaf_node",
         "asin_amazon_orders_sum_ym_rate", "asin_amazon_orders_sum_lm_rate",
         "asin_bsr_orders_sum_ym_rate", "asin_bsr_orders_sum_lm_rate",
         "asin_bsr_orders_mean_ym_rate", "asin_bsr_orders_mean_lm_rate",