店铺新增字段 优化导出

Pyspark_job/dwt/dwt_fb_base_report.py

@@ -32,6 +32,9 @@ from yswg_utils.common_udf import udf_new_asin_flag       # top20 新品判断
 
 class DwtFbBaseReport(object):
 
+    # 三个 Feedback 同比变化率列，handle_yoy_rate_padding 按这份清单批量填 null
+    YOY_COLS = ['count_30_day_yoy_rate', 'count_1_year_yoy_rate', 'count_life_time_yoy_rate']
+
     def __init__(self, site_name, date_type, date_info):
         self.site_name = site_name
         self.date_type = date_type
@@ -51,6 +54,10 @@ class DwtFbBaseReport(object):
         self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
         # last_month: 上个月（如 2023-08 -> 2023-07），用于计算 Feedback 环比变化率
         self.last_month = CommonUtil.get_month_offset(date_info, -1)
+        # last_year: 去年同月（如 2026-07 -> 2025-07），用于计算 Feedback 同比变化率
+        self.last_year = CommonUtil.get_month_offset(date_info, -12)
+        # 同比只从 2025-08 起算：更早月份的对比期分区没有验证过，读不到会让三列全落 +1000 假上涨
+        self.is_yoy_calc = (date_type == 'month' and date_info >= '2025-08')
 
         # 全局 df 初始化
         self.df_fb_feedback = self.spark.sql(f"select 1+1;")     # dim_fb_detail 当月+上月环比
@@ -158,6 +165,104 @@ class DwtFbBaseReport(object):
         df = df.drop('seller_address', 'seller_rating', 'feedback_histogram', 'metadata_json')
         return df
 
+    def _read_last_year_feedback(self):
+        """读去年同月的三个评价数，当同比分母。dim_fb_detail 从 2026-06 起有分区，更早回退读 ODS"""
+        if self.last_year >= '2026-06':
+            print(f"获取 dim_fb_detail（去年同月 {self.last_year}，仅 count 字段）")
+            sql = f"""
+            select seller_id,
+                count_30_day_num as ly_30_day_num,
+                count_1_year_num as ly_1_year_num,
+                count_lifetime_num as ly_lifetime_num
+            from dim_fb_detail
+            where site_name = '{self.site_name}'
+              and date_type = '{self.date_type}'
+              and date_info = '{self.last_year}'
+            """
+            df = self.spark.sql(sqlQuery=sql)
+            print(sql)
+            return df
+
+        print(f"获取 ods_seller_account_feedback（去年同月 {self.last_year}，仅 count 字段）")
+        sql = f"""
+        select seller_id,
+            count_30_day as ly_30_day_num,
+            count_1_year as ly_1_year_num,
+            count_lifetime as ly_lifetime_num,
+            created_at
+        from ods_seller_account_feedback
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
+          and date_info = '{self.last_year}'
+          and length(seller_id) > 2
+        """
+        df = self.spark.sql(sqlQuery=sql)
+        print(sql)
+        # 同一 seller_id 一个月可能有多条爬取记录，按 created_at 倒序留最新一条
+        # 不去重的话下面 join 会 1:N 放大，行数暴涨且同比算错
+        w_ly = Window.partitionBy('seller_id').orderBy(F.col('created_at').desc())
+        df = df.withColumn('_rn', F.row_number().over(w_ly)) \
+               .filter(F.col('_rn') == 1) \
+               .drop('_rn', 'created_at')
+        return df
+
+    @staticmethod
+    def yoy_rate_expr(cur_col, cmp_col):
+        """
+        同比变化率 (本期-去年同月)/去年同月，写法对齐 dwt_aba_last_change_rate 的本期锚定简版。
+
+        取值:
+          null   本月该字段缺失，算不出同比
+          +1000  去年无该店(join 不上)或去年为 0，今年有 —— 新出现/上升
+          0      去年无/为 0 且今年也是 0 —— 无方向
+          其余   真实比率，今年归零时就是 -1.0（是真实值不是占位）
+
+        本月 count 缺失必须单独挡在第一个分支:
+        ods_seller_account_feedback 2024-08~2025-09 每月有 16~223 行三个 count 同时为 null，
+        漏判的话 `null == 0` 在 Spark 里是 null、第二个分支不成立，会直接掉进 +1000 记成假上升。
+        判 <= 0 而不是 == 0 是防御性的（实测各月 neg_cnt 均为 0，源表没有负数）。
+        """
+        cur = F.col(cur_col)
+        cmp_val = F.col(cmp_col)
+        return F.when(
+            cur.isNull(), F.lit(None).cast(DoubleType())
+        ).when(
+            (cmp_val.isNull() | (cmp_val <= 0)) & (cur == 0), F.lit(0.0)
+        ).when(
+            cmp_val.isNull() | (cmp_val <= 0), F.lit(1000.0)
+        ).otherwise(
+            F.round((cur - cmp_val) / cmp_val, 4)
+        )
+
+    def handle_yoy_rate_padding(self, df):
+        """
+        不在同比计算范围的月份：三列填 null，保证各分区 schema 一致、save_data 的 select 不报错。
+        比率类字段的门控占位项目里就是填 null（见 dwt_aba_last_change_rate.handle_rank_rate_padding），
+        不用 -1（-1 是同比真实值，评价数归零 = -100%）、也不用 -1000（真值里不出现，没有区分度）。
+        PG 侧这些月份 ALTER TABLE ADD COLUMN 之后本来就是 NULL，填别的值也改不了这个事实。
+        """
+        for col in self.YOY_COLS:
+            df = df.withColumn(col, F.lit(None).cast(DoubleType()))
+        return df
+
+    def handle_yoy_rate(self, df):
+        """当月 LEFT JOIN 去年同月，补三个同比变化率列；门控外走 padding"""
+        if not self.is_yoy_calc:
+            print(f"date_info={self.date_info} 不在同比计算范围（month 且 >= 2025-08），三列填 null")
+            return self.handle_yoy_rate_padding(df)
+
+        df_ly = self._read_last_year_feedback()
+        # 左连：本期店铺一个不丢，去年没有的店铺三个 ly_ 列为 null，由 yoy_rate_expr 兜成 +1000
+        df = df.join(df_ly, on='seller_id', how='left')
+        return df \
+            .withColumn('count_30_day_yoy_rate',
+                        self.yoy_rate_expr('count_30_day_num', 'ly_30_day_num')) \
+            .withColumn('count_1_year_yoy_rate',
+                        self.yoy_rate_expr('count_1_year_num', 'ly_1_year_num')) \
+            .withColumn('count_life_time_yoy_rate',
+                        self.yoy_rate_expr('count_lifetime_num', 'ly_lifetime_num')) \
+            .drop('ly_30_day_num', 'ly_1_year_num', 'ly_lifetime_num')
+
     def read_data(self):
         if self.date_info >= '2026-06':
             # 2026-06 起读 dim_fb_detail（已预处理 rating/histogram/metadata/business/is_self）
@@ -276,8 +381,11 @@ class DwtFbBaseReport(object):
                 F.round((F.col('count_1_year_num') - F.col('last_1_year_num')) / F.col('last_1_year_num'), 4)) \
             .withColumn('count_life_time_rate',
                 F.round((F.col('count_lifetime_num') - F.col('last_lifetime_num')) / F.col('last_lifetime_num'), 4)) \
-            .drop('last_30_day_num', 'last_1_year_num', 'last_lifetime_num') \
-            .cache()
+            .drop('last_30_day_num', 'last_1_year_num', 'last_lifetime_num')
+
+        # 当月 LEFT JOIN 去年同月，计算三个 Feedback 同比变化率
+        self.df_fb_feedback = self.handle_yoy_rate(self.df_fb_feedback)
+        self.df_fb_feedback = self.df_fb_feedback.cache()
 
         # 读取店铺-ASIN 关系表（dim_fb_asin_info 从 2026-06 开始有分区，历史月读 2026-06）
         asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
@@ -674,6 +782,14 @@ class DwtFbBaseReport(object):
             F.col('fb_star_3_pct'),
             F.col('fb_star_2_pct'),
             F.col('fb_star_1_pct'),
+
+            # 三个 Feedback 同比变化率（对比期=去年同月）
+            # 取值: +1000=去年无该店或为0、今年有(上升), 0=两期都是0(无方向),
+            #      null=本月字段缺失或该月不算同比, 其余为真实比率(今年归零就是 -1.0)
+            # 列顺序排在最后，与 Hive ALTER TABLE ADD COLUMNS 加在表末尾对齐
+            F.col('count_30_day_yoy_rate'),
+            F.col('count_1_year_yoy_rate'),
+            F.col('count_life_time_yoy_rate'),
             F.lit(self.site_name).alias('site_name'),
             F.lit(self.date_type).alias('date_type'),
             F.lit(self.date_info).alias('date_info')

Pyspark_job/dwt/dwt_fb_category_report.py

 """
    @Author      : hejiangming
    @Description : 店铺分类统计表
-                  数据源从 ODS 切换到 flow:
-                  - ods_seller_asin_account → dwt_flow_asin(seller-asin + asin_is_new + category)
-                  - ods_seller_account_feedback → 不再读取(以 flow 为主表)
-                  - dim_cal_asin_history_detail → 仅用于 bsr_asin_num 市场总量
+                  店铺集合(主表)恢复为 feedback 口径，不再以 dwt_flow_asin 为主表。
+                  flow 只覆盖流量选品有 asin 的店铺(约 77 万)，当主表会把 feedback 全量
+                  219 万店里的 2/3 挡在报表外，2025-10 起的分区就是这么变成 70 万行的。
+                  本次只换数据源(ODS -> DIM)，聚合与三个占比公式沿用原逻辑不动。
    @SourceTable :
-                  ①dwt_flow_asin
-                  ②dim_cal_asin_history_detail
-                  ③dim_bsr_category_tree
+                  1.dim_fb_detail / ods_seller_account_feedback  (店铺集合，按月门控)
+                  2.dim_fb_asin_info                             (店铺-ASIN 关系)
+                  3.dwt_flow_asin                                (asin_is_new，按 asin 取值)
+                  4.dim_cal_asin_history_detail                  (一级分类 + 市场总量分母)
+                  5.dim_bsr_category_tree                        (一级分类名称)
    @SinkTable   :
-                  ①dwt_fb_category_report
+                  1.dwt_fb_category_report
    @CreateTime  : 2023/07/18 17:33
-   @UpdateTime  : 2026/07/29
+   @UpdateTime  : 2026/08/17 17:59
 """
 
 import os
@@ -46,7 +48,8 @@ class DwtFbCategoryReport(object):
         self.spark = SparkUtil.get_spark_session(app_name)
 
         # 初始化全局df
-        self.df_fb_asin_info = self.spark.sql(f"select 1+1;")
+        self.df_fb_asin_info = self.spark.sql(f"select 1+1;")   # 主表店铺 x 店铺-ASIN 关系
+        self.df_flow_asin = self.spark.sql(f"select 1+1;")      # asin -> is_asin_new
         self.df_asin_history = self.spark.sql(f"select 1+1;")
         self.df_cate_name = self.spark.sql(f"select 1+1;")
         self.df_fb_cate_asin_cal = self.spark.sql(f"select 1+1;")
@@ -55,30 +58,78 @@ class DwtFbCategoryReport(object):
         self.df_result_cal = self.spark.sql(f"select 1+1;")
 
     def read_data(self):
-        # 从 flow 读 seller-asin 关系 + asin_is_new + category_first_id, 以 flow 为主表
-        # 替代原来的 ods_seller_account_feedback(seller集合) + ods_seller_asin_account(seller-asin)
-        # + dim_cal_asin_history_detail(asin_launch_time+category)
-        # asin_is_new 直接用 flow 已算好的值, 不再 UDF 重算
-        print("获取 dwt_flow_asin (seller-asin + asin_is_new + category)")
+        # 旧代码：以 flow 为主表，seller 集合直接取 dwt_flow_asin.account_id
+        # 换掉的原因见文件头 —— flow 只覆盖有流量选品 asin 的店铺，当主表会丢掉 2/3 的店
+        # sql = f"""
+        # select account_id as seller_id, asin,
+        #        asin_is_new as is_asin_new,
+        #        category_first_id as bsr_cate_1_id
+        # from dwt_flow_asin
+        # where site_name = '{self.site_name}' and date_type = '{self.date_type}'
+        #   and date_info = '{self.date_info}' and account_id is not null
+        # """
+
+        # 店铺集合(主表)：2026-06 起读 dim_fb_detail，之前的月份回退读 ODS
+        # 门控与取数口径对齐 dwt_fb_base_report，保证两张报表的店铺集合一致
+        if self.date_info >= '2026-06':
+            print("获取 dim_fb_detail（店铺集合）")
             sql = f"""
-        select account_id as seller_id, asin,
-               asin_is_new as is_asin_new,
-               category_first_id as bsr_cate_1_id
-        from dwt_flow_asin
+            select seller_id
+            from dim_fb_detail
+            where site_name = '{self.site_name}'
+              and date_type = '{self.date_type}'
+              and date_info = '{self.date_info}'
+            """
+        else:
+            # length(seller_id) > 2 是脏数据过滤，与 dim_fb_detail 内部口径一致
+            print("获取 ods_seller_account_feedback（店铺集合）")
+            sql = f"""
+            select distinct seller_id
+            from ods_seller_account_feedback
             where site_name = '{self.site_name}'
               and date_type = '{self.date_type}'
               and date_info = '{self.date_info}'
-          and account_id is not null
+              and length(seller_id) > 2
+            """
+        df_fb_seller = self.spark.sql(sqlQuery=sql)
+        print(sql)
+
+        # 店铺-ASIN 关系：dim_fb_asin_info 从 2026-06 起有分区，历史月读 2026-06
+        # 替代原来的 ods_seller_asin_account —— 那张表读取时带 created_at <= cal_date 过滤，
+        # 爬虫先删后增会刷新 created_at，跨月重跑时活跃店铺被误筛成 0 商品；dim 层已去掉该过滤
+        asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
+        print(f"获取 dim_fb_asin_info（店铺-ASIN 关系，date_info={asin_info_date}）")
+        sql = f"""
+        select seller_id, asin
+        from dim_fb_asin_info
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
+          and date_info = '{asin_info_date}'
         """
-        # 原来从 ods_seller_asin_account 读, 有 created_at 过滤(爬虫先删后增导致活跃店铺被误筛)
-        # 现在从 flow 读, 不存在该问题
-        self.df_fb_asin_info = self.spark.sql(sqlQuery=sql)
+        df_fb_asin = self.spark.sql(sqlQuery=sql)
+        print(sql)
+
+        # 主表店铺 left join 店铺-ASIN 关系：店铺一个不丢，没有 asin 的店铺 asin 为 null
+        self.df_fb_asin_info = df_fb_seller.join(df_fb_asin, on='seller_id', how='left')
         self.df_fb_asin_info = self.df_fb_asin_info.drop_duplicates(['seller_id', 'asin']).cache()
+
+        # asin_is_new 从 flow 按 asin 取值（不是按店铺，不影响店铺集合）
+        # 用 flow 已算好的值而不是 UDF 重算，口径与 dwt_fb_base_report.fb_new_asin_num 对齐
+        print("获取 dwt_flow_asin（asin_is_new）")
+        sql = f"""
+        select asin, asin_is_new as is_asin_new
+        from dwt_flow_asin
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
+          and date_info = '{self.date_info}'
+        """
+        self.df_flow_asin = self.spark.sql(sqlQuery=sql).drop_duplicates(['asin'])
         print(sql)
 
-        # dim_cal_asin_history_detail: 仅用于 bsr_asin_num(按一级分类统计全量 asin 数, 作为市场占比分母)
-        # seller 的 asin_is_new 和 category 已从 flow 取, 这里不再取 asin_launch_time
-        print("获取 dim_cal_asin_history_detail (bsr_asin_num)")
+        # dim_cal_asin_history_detail: 两个用途
+        # 1. 给上面的 seller-asin 补一级分类（asin 全历史维表，覆盖率高于 flow 当月快照）
+        # 2. 按一级分类统计全量 asin 数，作为市场占比 fb_market_rate 的分母
+        print("获取 dim_cal_asin_history_detail (bsr_cate_1_id + bsr_asin_num)")
         sql = f"""
             select asin, category_first_id as bsr_cate_1_id
             from dim_cal_asin_history_detail
@@ -107,9 +158,15 @@ class DwtFbCategoryReport(object):
         self.sava_data()
 
     def handle_fb_agg(self):
-        # df_fb_asin_info 已含 is_asin_new 和 bsr_cate_1_id(从 flow 读取)
-        # 不需要 join asin_history 取 category/launch_time, 也不需要 UDF 算新品
-        self.df_fb_cate_asin_cal = self.df_fb_asin_info
+        # 旧代码：df_fb_asin_info 直接来自 flow，已自带 is_asin_new 和 bsr_cate_1_id，不用再 join
+        # self.df_fb_cate_asin_cal = self.df_fb_asin_info
+
+        # 现在 df_fb_asin_info 只有 seller_id + asin，按 asin 补一级分类和新品标记
+        self.df_fb_cate_asin_cal = self.df_fb_asin_info \
+            .join(self.df_asin_history, on='asin', how='left') \
+            .join(self.df_flow_asin, on='asin', how='left')
+
+        # 分类取不到的 asin 归到 '无'，与原逻辑一致
         self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.na.fill({'bsr_cate_1_id': '无'})
 
         # 按 seller_id + 一级分类聚合: 分类下 asin 数 + 新品数
@@ -118,7 +175,8 @@ class DwtFbCategoryReport(object):
             F.sum("is_asin_new").alias("fb_cate_new_asin_num"),
         )
 
-        # 店铺总 asin 数(flow 口径)
+        # 店铺总 asin 数：数 dim_fb_asin_info 的关系条数，与 dwt_fb_base_report.fb_asin_total 同口径
+        # count 会跳过 null，所以没有任何 asin 的店铺这里是 0
         self.df_fb_asin_cal = self.df_fb_asin_info.groupby(['seller_id']).agg(
             F.count("asin").alias("fb_asin_num"))
 

Pyspark_job/sqoop_export/dwt_fb_asin_info.py

@@ -34,6 +34,8 @@ if __name__ == '__main__':
     suffix = str(date_info).replace("-", "_")
     base_tb = f"{site_name}_fb_asin_info"
     export_master_tb = f"{base_tb}_{year_str}"
+    # 上一年母表，跨年时用它 like 出当年母表（见下面建表 sql）
+    export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
     export_tb = f"{base_tb}_{suffix}"
     next_month = CommonUtil.get_next_val(date_type, date_info)
 
@@ -41,8 +43,14 @@ if __name__ == '__main__':
     engine = DBUtil.get_db_engine(db_type, site_name)
 
 
+    # 跨年自动建当年母表：用上一年母表 like 出来（including all 带上索引/约束/默认值/注释），按 date_info 分区
+    # 不建的话跨年第一个月（如 2027-01）下面 like {export_master_tb} 会报表不存在
+    # if not exists 幂等，已存在直接跳过、只加不删
     # 保证幂等性，先删除原始表同周期的数据
     sql = f"""
+         create table if not exists {export_master_tb}
+                (like {export_master_tb_before} including all)
+         partition by range (date_info);
          drop table if exists {export_tb};
          create table if not exists {export_tb}
                 (

Pyspark_job/sqoop_export/dwt_fb_base_report.py

@@ -30,6 +30,8 @@ if __name__ == '__main__':
     suffix = str(date_info).replace("-", "_")
     base_tb = f"{site_name}_fb_base_report"
     export_master_tb = f"{base_tb}_{year_str}"
+    # 上一年母表，跨年时用它 like 出当年母表（见下面建表 sql）
+    export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
     export_tb = f"{base_tb}_{suffix}"
     next_month = CommonUtil.get_next_val(date_type, date_info)
 
@@ -45,8 +47,14 @@ if __name__ == '__main__':
         sys.exit(1)
     print(f"Hive 分区文件数：{len(hive_files)}，路径：{hive_partition_path}，继续导出")
 
+    # 跨年自动建当年母表：用上一年母表 like 出来（including all 带上索引/约束/默认值/注释），按 date_info 分区
+    # 不建的话跨年第一个月（如 2027-01）下面 like {export_master_tb} 会报表不存在
+    # if not exists 幂等，已存在直接跳过、只加不删
     # 保证幂等性，先删除原始表同周期的数据
     sql = f"""
+         create table if not exists {export_master_tb}
+                (like {export_master_tb_before} including all)
+         partition by range (date_info);
          drop table if exists {export_tb};
          create table if not exists {export_tb}
                 (
@@ -130,7 +138,11 @@ if __name__ == '__main__':
             "fb_star_4_pct",
             "fb_star_3_pct",
             "fb_star_2_pct",
-            "fb_star_1_pct"
+            "fb_star_1_pct",
+            # Feedback 同比变化率（对比去年同月）
+            "count_30_day_yoy_rate",
+            "count_1_year_yoy_rate",
+            "count_life_time_yoy_rate"
         ],
         partition_dict={
             "site_name": site_name,

Pyspark_job/sqoop_export/dwt_fb_category_report.py

@@ -31,6 +31,8 @@ if __name__ == '__main__':
     suffix = str(date_info).replace("-", "_")
     base_tb = f"{site_name}_fb_category_report"
     export_master_tb = f"{base_tb}_{year_str}"
+    # 上一年母表，跨年时用它 like 出当年母表（见下面建表 sql）
+    export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
     export_tb = f"{base_tb}_{suffix}"
     next_month = CommonUtil.get_next_val(date_type, date_info)
 
@@ -38,8 +40,14 @@ if __name__ == '__main__':
     engine = DBUtil.get_db_engine(db_type, site_name)
 
 
+    # 跨年自动建当年母表：用上一年母表 like 出来（including all 带上索引/约束/默认值/注释），按 date_info 分区
+    # 不建的话跨年第一个月（如 2027-01）下面 like {export_master_tb} 会报表不存在
+    # if not exists 幂等，已存在直接跳过、只加不删
     # 保证幂等性，先删除原始表同周期的数据
     sql = f"""
+         create table if not exists {export_master_tb}
+                (like {export_master_tb_before} including all)
+         partition by range (date_info);
          drop table if exists {export_tb};
          create table if not exists {export_tb}
                 (

Pyspark_job/sqoop_export/dwt_fb_top20_asin_info.py

@@ -32,6 +32,8 @@ if __name__ == '__main__':
     suffix = str(date_info).replace("-", "_")
     base_tb = f"{site_name}_fb_top20_asin_info"
     export_master_tb = f"{base_tb}_{year_str}"
+    # 上一年母表，跨年时用它 like 出当年母表（见下面建表 sql）
+    export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
     export_tb = f"{base_tb}_{suffix}"
     next_month = CommonUtil.get_next_val(date_type, date_info)
 
@@ -39,8 +41,14 @@ if __name__ == '__main__':
     engine = DBUtil.get_db_engine(db_type, site_name)
 
 
+    # 跨年自动建当年母表：用上一年母表 like 出来（including all 带上索引/约束/默认值/注释），按 date_info 分区
+    # 不建的话跨年第一个月（如 2027-01）下面 like {export_master_tb} 会报表不存在
+    # if not exists 幂等，已存在直接跳过、只加不删
     # 保证幂等性，先删除原始表同周期的数据
     sql = f"""
+         create table if not exists {export_master_tb}
+                (like {export_master_tb_before} including all)
+         partition by range (date_info);
          drop table if exists {export_tb};
          create table if not exists {export_tb}
                 (