周趋势图计算与导出脚本

Pyspark_job/dwt/dwt_st_base_report_week.py

+"""
+   @Author      : hejiangming
+   @Description : 周趋势图独立计算：直读 ODS 按新口径算周趋势(排名+搜索量),写 Hive dwt_st_base_report(date_type=week)
+   @SourceTable :
+                  1.ods_brand_analytics          搜索词周报告(搜索词+排名+asin,按周分区)
+                  2.ods_st_key                   搜索词→st_key 映射(站点级)
+                  3.ods_rank_search_rate_repeat  排名→搜索量/转化率 映射(站点级)
+   @SinkTable   : dwt_st_base_report (date_type=week)
+   @CreateTime  : 2026/07/17 12:05
+   @UpdateTime  : 2026/07/17 12:05
+"""
+import os
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
+
+from pyspark.sql import functions as F
+from pyspark.sql.window import Window
+from utils.hdfs_utils import HdfsUtils
+from utils.common_util import CommonUtil
+from utils.spark_util import SparkUtil
+
+
+class DwtStBaseReportWeek(object):
+
+    def __init__(self, site_name, date_type, date_info):
+        self.site_name = site_name
+        self.date_type = date_type   # 固定 week
+        self.date_info = date_info    # 周: 2025-24
+        self.hive_tb = "dwt_st_base_report"
+        app_name = f"{self.hive_tb}_week:{site_name} {date_type} {date_info}"
+        self.spark = SparkUtil.get_spark_session(app_name)
+        self.partitions_num = CommonUtil.reset_partitions(site_name, 1)
+
+    def get_rank_map(self):
+        """
+        读 ods_rank_search_rate_repeat 产出 rank → (st_volume, st_orders) 映射。
+        照搬 dim/dim_st_detail_week.py 第146-152行 + 第293-297行。
+        """
+        # ============================================================
+        # 【为什么不用 WHERE 限日期】这张表站点级实际只有一个分区，靠「按 rank 分组 + date_info 倒序取第一行」
+        #   拿每个 rank 最新那份映射，等价于直接读那个分区(与 dim 一致)
+        # 【算法】周搜索量 = 月度 search_num / 4；周预估销量 = 周搜索量 × 转化率 rate(四舍五入)
+        # 【例子】rank=1 月度 search_num=2768855 → st_volume=round(2768855/4)=692214
+        # ============================================================
+        sql = f"""
+        select rank, search_num as search_volume, rate as st_search_rate, date_info
+        from ods_rank_search_rate_repeat
+        where site_name = '{self.site_name}'
+        """
+        df_map = self.spark.sql(sql)
+        window = Window.partitionBy(['rank']).orderBy(df_map.date_info.desc())
+        df_map = df_map.withColumn('rk', F.row_number().over(window)) \
+                       .filter('rk = 1') \
+                       .drop('rk', 'date_info')
+        df_map = df_map.withColumn('search_volume', F.round(F.col('search_volume') / 4)) \
+                       .withColumn('orders', F.round(F.col('search_volume') * F.col('st_search_rate'))) \
+                       .select('rank', 'search_volume', 'orders')
+        return df_map.repartition(40, 'rank').cache()
+
+    def run(self):
+        # ============================================================
+        # Step1：读本周 ods_brand_analytics 的「搜索词 + 排名 + top3 asin」
+        # 【为什么读 asin1/2/3】后面要照 dim 做 asin 长度过滤(剔 asin 异常词)，所以得把 asin 带出来
+        # 【为什么 search_term is not null】过滤掉搜索词为空的垃圾行(与 dim/backfill 一致)
+        # ============================================================
+        sql = f"""
+        select search_term, rank, updated_time, asin1, asin2, asin3
+        from ods_brand_analytics
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
+          and date_info = '{self.date_info}'
+          and search_term is not null
+        """
+        df_aba = self.spark.sql(sql)
+
+        # ============================================================
+        # Step2：按搜索词去重，保留 updated_time 最新的一条(照搬 dim 第96-99行)
+        # 【为什么去重】同一搜索词一周内可能有多条抓取记录，取最新那条防重复
+        # 【顺序对齐 dim】dim 是「先按原始 search_term 去重、再清 \x00」，本脚本保持同序
+        # ============================================================
+        dedup_win = Window.partitionBy(['search_term']).orderBy(F.col('updated_time').desc_nulls_last())
+        df_aba = df_aba.withColumn('dt_rank', F.row_number().over(dedup_win)) \
+                       .filter('dt_rank = 1') \
+                       .drop('dt_rank', 'updated_time')
+
+        # ============================================================
+        # Step3：清 \x00 脏字符(照搬 dim 第102-103行，对 search_term + asin1/2/3)
+        # 【为什么】ABA 早期数据带空字节，清完仍乱码的词后面 inner join st_key 时自然被丢
+        # ============================================================
+        for col in ['search_term', 'asin1', 'asin2', 'asin3']:
+            df_aba = df_aba.withColumn(col, F.regexp_replace(F.col(col), '\x00', ''))
+        df_aba = df_aba.repartition(40, 'search_term').cache()
+
+        # ============================================================
+        # Step4：asin 长度过滤(照搬 dim save_data 第359-360行)
+        # 【为什么】这是与 us 现有口径一模一样的关键一步：dim 落盘时剔掉 asin 长度>10 的异常词(约18个)，
+        #   不做这步 ODS 直读会比 us 多这18个词
+        # ============================================================
+        df_aba = df_aba.filter('length(asin1) <= 10 AND length(asin2) <= 10 AND length(asin3) <= 10')
+
+        # ============================================================
+        # Step5：拿 st_key(照搬 dim 第156-159行)，inner join 丢掉无 key 的词
+        # ============================================================
+        sql = f"""
+        select st_key, search_term from ods_st_key where site_name = '{self.site_name}'
+        """
+        df_st_key = self.spark.sql(sql).repartition(40, 'search_term').cache()
+
+        # rank → 搜索量/销量映射
+        df_map = self.get_rank_map()
+
+        # ============================================================
+        # Step6：拼装。search_term inner join st_key(与 dim save_data 一致)，再按 rank left join 映射
+        # ============================================================
+        df_save = df_aba.join(df_st_key, on='search_term', how='inner') \
+                        .join(df_map, on='rank', how='left')
+
+        # 补充非数据字段：年份取自 date_info，时间戳取当前
+        df_save = df_save.withColumn('years', F.lit(int(self.date_info.split('-')[0]))) \
+                         .withColumn('created_time', F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS')) \
+                         .withColumn('updated_time', F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS')) \
+                         .withColumn('site_name', F.lit(self.site_name)) \
+                         .withColumn('date_type', F.lit(self.date_type)) \
+                         .withColumn('date_info', F.lit(self.date_info))
+
+        # ============================================================
+        # Step7：空值兜底 -1(照搬 dim save_data 第430-434行)
+        # 【为什么】与 us 一模一样：rank 无映射时 search_volume/orders 会是 null，dim 填 -1；
+        #   实测每个 rank 都有映射、不触发，但补上保证逐字节一致
+        # 【为什么用 -1】下游 Java 读 DB NULL 有问题，用 -1 占位，Java 转 null 返前端
+        # ============================================================
+        df_save = df_save.select(
+            'st_key',
+            'search_term',
+            F.col('search_volume').alias('st_volume'),
+            F.col('orders').alias('st_orders'),
+            F.col('rank').alias('st_rank'),
+            'years',
+            'created_time',
+            'updated_time',
+            'site_name',
+            'date_type',
+            'date_info'
+        ).fillna({
+            'st_volume': -1,
+            'st_orders': -1,
+            'st_rank': -1
+        })
+
+        # ============================================================
+        # Step8：落盘。先物理删该分区目录再 append，保证幂等
+        # 【为什么先删】saveAsTable append 不覆盖同分区旧文件，重跑会叠加重复数据
+        # ============================================================
+        hdfs_path = f"/home/{SparkUtil.DEF_USE_DB}/dwt/{self.hive_tb}/site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}"
+        print(f"清除hdfs目录中数据:{hdfs_path}")
+        HdfsUtils.delete_hdfs_file(hdfs_path)
+
+        df_save = df_save.repartition(self.partitions_num)
+        partition_by = ["site_name", "date_type", "date_info"]
+        print(f"当前存储的表名为：{self.hive_tb}, 分区为{partition_by}")
+        df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
+        print("success")
+
+
+if __name__ == '__main__':
+    site_name = CommonUtil.get_sys_arg(1, None)   # 参数1：站点 us/uk/de
+    date_type = CommonUtil.get_sys_arg(2, None)   # 参数2：固定 week
+    date_info = CommonUtil.get_sys_arg(3, None)   # 参数3：周 2025-24
+    obj = DwtStBaseReportWeek(site_name, date_type, date_info)
+    obj.run()

Pyspark_job/sqoop_export/dwt_st_base_report_week.py

+"""
+   @Author      : hejiangming
+   @Description : 周趋势图独立导出：Hive dwt_st_base_report(week) → PG集群 {site}_aba_last_total_week(自动建年分区)
+   @SourceTable : dwt_st_base_report (date_type=week)
+   @SinkTable   : {site}_aba_last_total_week (PG集群, 按年 RANGE 分区)
+   @CreateTime  : 2026/07/17 12:05
+   @UpdateTime  : 2026/07/17 12:05
+"""
+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
+from utils.db_util import DBUtil, DbTypes
+
+if __name__ == '__main__':
+    site_name = CommonUtil.get_sys_arg(1, None)   # 参数1：站点 us/uk/de
+    date_type = CommonUtil.get_sys_arg(2, None)   # 参数2：固定 week
+    date_info = CommonUtil.get_sys_arg(3, None)   # 参数3：周 2025-24
+    print(f"执行参数为{sys.argv}")
+
+
+    CommonUtil.judge_is_work_hours(
+        site_name=site_name, date_type=date_type, date_info=date_info,
+        principal='hejiangming', priority=1, export_tools_type=1, belonging_to_process='ABA周增长趋势图'
+    )
+
+    # ============================================================
+    # 趋势导 PG 集群(postgresql_cluster)。逻辑照搬 sqoop_export/dim_st_detail_week.py
+    #   (us 现有趋势导出)，抽出来三站共用。
+    # ============================================================
+    db_type = DbTypes.postgresql_cluster.name
+    year_str = CommonUtil.safeIndex(date_info.split("-"), 0, None)
+    year_next = str(int(year_str) + 1)
+
+    master_tb = f"{site_name}_aba_last_total_week"
+    export_tb = f"{master_tb}_{year_str}"
+
+    engine = DBUtil.get_db_engine(db_type, site_name)
+
+    # ============================================================
+    # 建当年分区子表 + st_key 索引 + 清本周
+    # 【为什么 create if not exists partition of】趋势表是「顶层母表 + 按年 RANGE 分区子表」，
+    #   新一年第一次导出自动建该年子表，不用手动预建
+    # 【为什么先 delete 本周】导出是「先删该周、再 sqoop append」，幂等；Hive 有数据重跑即补回
+    # ============================================================
+    sql = f"""
+        create table if not exists {export_tb} partition of {master_tb} for values from ('{year_str}') to ('{year_next}');
+        create index if not exists {export_tb}_st_key_idx on {export_tb} using btree (st_key);
+        delete from {export_tb} where date_info = '{date_info}';
+    """
+    print("================================执行sql================================")
+    print(sql)
+    DBUtil.engine_exec_sql(engine, sql)
+
+    # sqoop 导出：Hive dwt_st_base_report 对应分区 → PG 子表
+    columns = ["st_key", "search_term", "st_volume", "st_rank", "st_orders", "years",
+               "created_time", "updated_time", "date_type", "date_info"]
+    sh = CommonUtil.build_export_sh(
+        site_name=site_name,
+        db_type=db_type,
+        hive_tb="dwt_st_base_report",
+        export_tb=export_tb,
+        col=columns,
+        partition_dict={
+            "site_name": site_name,
+            "date_type": "week",
+            "date_info": date_info
+        }
+    )
+    client = SSHUtil.get_ssh_client()
+    SSHUtil.exec_command_async(client, sh, ignore_err=False)
+    client.close()
+    print("================================周趋势图导出完成================================")
+