品牌调研近一年峰值计算

Pyspark_job/dwt/dwt_bs_top100_last365_peak.py

+"""
+   @Author      : hejiangming
+   @Description : 品类调研 近一年(滚动12月)分类月销峰值月计算。
+   @SourceTable : dwt_bs_top100 (最近 12 个月分区)
+   @SinkTable   : dwt_bs_top100_last365_peak
+   @CreateTime  : 2026/08/03 11:29
+   @UpdateTime  : 2026/08/03 11:29
+"""
+import os
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))
+
+from utils.hdfs_utils import HdfsUtils
+from utils.common_util import CommonUtil
+from utils.spark_util import SparkUtil
+from pyspark.sql import functions as F
+
+
+class DwtBsTop100Last365Peak(object):
+    """
+    在整个品类调研链路里的位置：
+      dwt_flow_asin → dwt_bs_top100（各月落表，本表的上游）→ 本任务（近12月滚动挑峰值月）→ 导出 PG 给页面 join。
+    页面不随查询月份变化：不管前端查哪个月，展示的都是最新这一份"近一年峰值月"。
+    """
+
+    # 分类跨月对齐主键，与 dwt_bs_top100_change_rate 保持一致。
+    # current_id 是当前分类节点，parent_id_join 是根→直接父级的完整路径；
+    # 实测近12月每个 current_id 只有一条路径（keepa 树里 cat_id 唯一→一条 node_id_path），
+    # 仍带上路径，保持和 change_rate 同键，将来 keepa 树若变也能兜住。
+    JOIN_KEYS = ['asin_node_current_id', 'asin_node_parent_id_join']
+
+    # 峰值取哪个指标：分类总月销（来自前台"已售"徽章 asin_bought_month，
+    # 进 dwt_bs_top100 时改名 asin_amazon_orders、分类层聚合成 _sum）
+    PEAK_COL = 'asin_amazon_orders_sum'
+
+    # 近一年窗口长度（含本期月），即 date_info 往前推 11 个月
+    WINDOW_MONTHS = 12
+
+    def __init__(self, site_name, date_type, date_info):
+        self.site_name = site_name
+        self.date_type = date_type
+        self.date_info = date_info
+        self.hive_tb = "dwt_bs_top100_last365_peak"
+        self.source_tb = "dwt_bs_top100"
+        app_name = f"{self.hive_tb}:{site_name}:{date_type}:{date_info}"
+        self.spark = SparkUtil.get_spark_session(app_name)
+        self.partitions_num = CommonUtil.reset_partitions(site_name, 1)
+
+        # 窗口起点 = 本期往前推 11 个月，如本期 2026-06 → 窗口 2025-07 ~ 2026-06
+        self.start_month = CommonUtil.get_month_offset(self.date_info, -(self.WINDOW_MONTHS - 1))
+        print(f"本期={self.date_info}，近一年窗口={self.start_month} ~ {self.date_info}")
+
+        # 落表前清 HDFS 分区目录（幂等重跑，先删后写不残留旧文件）
+        hdfs_path = CommonUtil.build_hdfs_path(
+            self.hive_tb,
+            partition_dict={"site_name": site_name, "date_type": date_type, "date_info": date_info},
+        )
+        print(f"清除 hdfs 目录：{hdfs_path}")
+        HdfsUtils.delete_hdfs_file(hdfs_path)
+
+        self.df_base = self.spark.sql("select 1+1;")
+        self.df_save = self.spark.sql("select 1+1;")
+
+    def read_data(self):
+        # 一次读满窗口内的 12 个月分区，只取对齐键 + 月份 + 月销三样。
+        # date_info 是 string 分区列，'YYYY-MM' 的字典序就是时间序，直接用 >= / <= 比较，分区裁剪正常生效。
+        sql = f"""
+            select {', '.join(self.JOIN_KEYS)}, date_info, {self.PEAK_COL}
+            from {self.source_tb}
+            where site_name='{self.site_name}' and date_type='{self.date_type}'
+              and date_info >= '{self.start_month}' and date_info <= '{self.date_info}'
+        """
+        print(f"读取 {self.source_tb} 近一年分区: sql -- {sql}")
+        self.df_base = self.spark.sql(sqlQuery=sql)
+
+    def handle_data(self):
+        # 只有"当月确实有销量"的行才有资格当峰值月，先滤掉两类：
+        #   null = 该分类当月没有月销数据（前台徽章覆盖率低，整组 asin 月销全空时 sum 返 null）
+        #   0    = 当月一件没卖动
+        # 都不是"卖得最好的月"。一整年一条都不剩的分类不会出现在本表里，
+        #   页面 join 不到 = 该分类近一年无月销数据（前端按空处理）。
+        # 缺月本身不用管：那个月压根没有行，进不了下面的聚合，天然只在有数据的月里比。
+        peak_col = F.col(self.PEAK_COL)
+        df = self.df_base.filter(peak_col.isNotNull() & (peak_col > 0))
+
+        # 每个分类收两样：近一年的最高月销，以及 (月销, 月份) 的全部明细
+        df_agg = df.groupBy(*self.JOIN_KEYS).agg(
+            F.max(self.PEAK_COL).alias("max_orders"),
+            F.collect_list(F.struct(self.PEAK_COL, "date_info")).alias("orders_arr"),
+        )
+
+        # 峰值月：从明细里挑出月销 = 最高值的项 → 取月份 → 升序 → 拼成 'YYYY-MM' 逗号串。
+        # 并列峰值全保留，格式与 ABA 峰值月(dwt_aba_last365.peak_month)一致，导出后转 PG 数组。
+        # 例：25-11 和 26-04 都是 3000 且并列最高 → "2025-11,2026-04"
+        # nullif 是防御：上面已滤掉 null/0，filter 理论上不会为空，真空了也转 null 交给 save_data 兜成空串。
+        df_agg = df_agg.withColumn(
+            "peak_month",
+            F.expr(f"""
+                nullif(
+                    concat_ws(',',
+                        array_sort(transform(
+                            filter(orders_arr, x -> x.{self.PEAK_COL} = max_orders),
+                            x -> x.date_info
+                        ))
+                    ),
+                    ''
+                )
+            """)
+        )
+        self.df_save = df_agg
+
+    def save_data(self):
+        # 落表 = 分类键2 + peak_month + 3 分区列
+        out_cols = self.JOIN_KEYS + ["peak_month"]
+        self.df_save = self.df_save.select(*out_cols) \
+            .na.fill({"peak_month": ""}) \
+            .withColumn("site_name", F.lit(self.site_name)) \
+            .withColumn("date_type", F.lit(self.date_type)) \
+            .withColumn("date_info", F.lit(self.date_info))
+
+        self.df_save = self.df_save.repartition(self.partitions_num)
+        partition_by = ["site_name", "date_type", "date_info"]
+        print(f"当前存储的表名为：{self.hive_tb}，分区为 {partition_by}")
+        self.df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
+        print("success")
+
+    def run(self):
+        self.read_data()
+        self.handle_data()
+        self.save_data()
+
+
+if __name__ == '__main__':
+    site_name = CommonUtil.get_sys_arg(1, None)   # 站点 us
+    date_type = CommonUtil.get_sys_arg(2, None)   # month
+    date_info = CommonUtil.get_sys_arg(3, None)   # 窗口最后一个月，如 2026-06
+    obj = DwtBsTop100Last365Peak(site_name, date_type, date_info)
+    obj.run()

Pyspark_job/sqoop_export/dwt_bs_top100_last365_peak.py

+"""
+   @Author      : hejiangming
+   @Description : 品类调研"近一年分类月销峰值月"窄表导出：Hive dwt_bs_top100_last365_peak
+                  → PG {site}_category_top_analysis_last365_peak。
+                  PG 侧是普通表(不按月分区)：页面查任何月都 join 同一份最新结果，
+                  每次导出用 copy 表 + exchange_tb 整表换入。
+   @SourceTable : dwt_bs_top100_last365_peak (Hive, 分区 site_name/date_type/date_info)
+   @SinkTable   : {site}_category_top_analysis_last365_peak (PG 集群, 正式表需手动建, DDL 见文件底部)
+   @CreateTime  : 2026/08/03 11:29
+   @UpdateTime  : 2026/08/03 11:29
+"""
+import os
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))
+from utils.ssh_util import SSHUtil
+from utils.common_util import CommonUtil, DateTypes
+from utils.db_util import DBUtil
+from utils.hdfs_utils import HdfsUtils
+
+if __name__ == '__main__':
+    site_name = CommonUtil.get_sys_arg(1, None)
+    date_type = CommonUtil.get_sys_arg(2, None)
+    date_info = CommonUtil.get_sys_arg(3, None)
+    # 最后一个参数判断是否导测试库
+    test_flag = CommonUtil.get_sys_arg(len(sys.argv) - 1, None)
+    print(f"执行参数为{sys.argv}")
+
+    # 近一年滚动结果只有月粒度
+    assert date_type == DateTypes.month.name, f"仅支持 month, 传入 date_type={date_type}"
+
+    if test_flag == 'test':
+        db_type = 'postgresql_test'
+        print("导出到PG测试库中")
+    else:
+        # belonging_to_process 单独用"品类调研峰值月流程"，不并进"品类调研流程_month"：
+        #   主表 + 同比环比两个导出走 modify_export_workflow_status，靠"同组未完成数=0"判断页面能否展示；
+        #   本窄表若挂进同一组，会让那两个跑完时未完成数永远不为 0，主流程状态行写不出来。
+        CommonUtil.judge_is_work_hours(site_name=site_name, date_type=date_type, date_info=date_info,
+                                       principal='hejiangming', priority=2, export_tools_type=1,
+                                       belonging_to_process=f'品类调研峰值月流程_{date_type}')
+        db_type = 'postgresql_cluster'
+        print("导出到PG集群库中")
+
+    # 导出前校验 Hive 分区有数据：空分区会灌出一张空 copy 表，换表后线上直接没数据
+    hive_path = CommonUtil.build_hdfs_path(
+        "dwt_bs_top100_last365_peak",
+        partition_dict={"site_name": site_name, "date_type": date_type, "date_info": date_info},
+    )
+    if not HdfsUtils.read_list(hive_path):
+        print(f"[ERROR] Hive 分区无数据文件：{hive_path}，跳过导出，请先检查 DWT 计算任务！")
+        sys.exit(1)
+    print(f"Hive 分区有数据：{hive_path}，继续导出")
+
+    export_tb_before = f"{site_name}_category_top_analysis_last365_peak"
+    export_tb_rel = f"{export_tb_before}_copy"
+    engine = DBUtil.get_db_engine(db_type, site_name)
+
+    # 建 copy 表：结构 like 正式表(含 peak_month VARCHAR[])。
+    # peak_month 正式表是 VARCHAR[]，sqoop 不能直写数组类型，先把 copy 的该列临时改成 VARCHAR 让 sqoop 写逗号串，
+    # sqoop 完、换表前再 ALTER 回 VARCHAR[]（机制同 dwt_aba_last365_peak.py）。
+    with engine.connect() as connection:
+        sql = f"""
+            drop table if exists {export_tb_rel};
+            create table if not exists {export_tb_rel}
+            (
+                like {export_tb_before} including comments
+            );
+            ALTER TABLE {export_tb_rel} ALTER COLUMN peak_month TYPE VARCHAR(500);
+        """
+        print("================================执行sql================================")
+        print(sql)
+        connection.execute(sql)
+
+    # 从 Hive 当前分区取 3 列写进 copy 表（列名与 Hive 表一致, sqoop 按名映射 PG 同名列）
+    sh = CommonUtil.build_export_sh(
+        site_name=site_name,
+        db_type=db_type,
+        hive_tb="dwt_bs_top100_last365_peak",
+        export_tb=export_tb_rel,
+        col=[
+            "asin_node_current_id",
+            "asin_node_parent_id_join",
+            "peak_month",
+        ],
+        partition_dict={
+            "site_name": site_name,
+            "date_type": date_type,
+            "date_info": date_info
+        }
+    )
+
+    client = SSHUtil.get_ssh_client()
+    SSHUtil.exec_command_async(client, sh, ignore_err=False)
+    client.close()
+
+    # 换表【前】把 copy 的 peak_month 从 VARCHAR 转回 VARCHAR[]（sqoop 写的逗号串拆成数组，
+    #   如 "2025-11,2026-04" → {2025-11,2026-04}）。
+    # 放在换表前：exchange_tb(cp_index_flag=True) 复制正式表索引时，列类型必须已经对上。
+    with engine.connect() as connection:
+        sql = f"""
+            ALTER TABLE {export_tb_rel}
+            ALTER COLUMN peak_month TYPE VARCHAR[]
+            USING string_to_array(coalesce(peak_month, ''), ',')::varchar[];
+        """
+        print("================================执行sql================================")
+        print(sql)
+        connection.execute(sql)
+
+    # 换表名：copy → 正式。cp_index_flag=True 自动把正式表现有索引复制到 copy 后再换入，
+    #   索引靠"正式表建表时带好 → 每次换表自动传递"维护，无需手动重建。
+    DBUtil.exchange_tb(engine,
+                       source_tb_name=export_tb_rel,
+                       target_tb_name=export_tb_before,
+                       cp_index_flag=True)
+    engine.dispose()
+
+    # 更新 workflow_everyday：本窄表被页面直接查询，需单独上报导出完成供"数据就绪"监控感知。
+    # 不走 modify_export_workflow_status（那是主表+同比环比"两个都完成才写"的机制，见上面 process 分组说明）。
+    # page='CategoryTopPeak'(新增页面标识, 需同步给后端)；test 模式不写，避免污染监控。
+    if test_flag != 'test':
+        engine = DBUtil.get_db_engine("mysql", "us")
+        with engine.connect() as connection:
+            sql = f"""
+                replace into workflow_everyday (
+                    site_name, report_date, status, status_val, table_name, date_type, page, is_end, remark, export_db_type
+                )
+                values (
+                    '{site_name}', '{date_info}', '导出pg完成', 14,
+                    '{export_tb_before}', '{date_type}', '品类调研峰值月', '是',
+                    '品类调研近一年月销峰值月窄表', 'postgresql_cluster'
+                );
+            """
+            print("================================更新workflow_everyday================================")
+            print(sql)
+            connection.execute(sql)
+        engine.dispose()
+
+    print("success")
+