单独导出 近一年 峰值月/常年可卖 给到月搜索词页面查询

Pyspark_job/sqoop_export/dwt_aba_last365_peak.py

+"""
+   @Author      : hejiangming
+   @Description : ABA搜索词"峰值月/常年可卖"窄表导出。从 Hive dwt_aba_last365 只取
+                  id/search_term/peak_month/all_year_text_flag 四列，单独落一张窄表，
+                  供月表 dwt_aba_st_analytics 页面按 id join 展示/筛选这两个字段——
+                  避免直接 join 宽表 us_aba_last_365_day(130+ 列)带来的字段名冲突和查询慢。
+   @SourceTable : dwt_aba_last365 (date_type='month' 近12月滚动结果)
+   @SinkTable   : {site_name}_aba_last_365_day_peak (PG集群，正式表需手动创建，见文件底部 DDL)
+   @CreateTime  : 2026/07/13 20:22
+   @UpdateTime  : 2026/07/15 11:50
+"""
+import os
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))
+
+from utils.db_util import DBUtil
+from utils.ssh_util import SSHUtil
+from utils.common_util import CommonUtil, DateTypes
+
+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}")
+
+    if test_flag == 'test':
+        db_type = 'postgresql_test'
+        print("导出到测试库中")
+    else:
+        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'新ABA流程_{date_type}'
+        )
+        db_type = "postgresql_cluster"
+        print("导出到PG集群中")
+
+    # 仅 month(近12月滚动)流程建窄表：月表页面 join 的就是这一份"最新近一年"结果
+    assert date_type == DateTypes.month.name, "dwt_aba_last365_peak 仅支持 date_type=month(近一年滚动结果)"
+
+    # 正式表；每次导出建 copy(like 正式表) → sqoop → 换表
+    export_tb_before = f"{site_name}_aba_last_365_day_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[](见下方，转回后 exchange_tb 才能连 gin 索引一起复制)。
+    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(200);
+        """
+        print("================================执行sql================================")
+        print(sql)
+        connection.execute(sql)
+
+    # 导出脚本：从 Hive dwt_aba_last365 当前分区只取 4 列，写入 copy 表
+    sh = CommonUtil.build_export_sh(
+        site_name=site_name,
+        db_type=db_type,
+        hive_tb="dwt_aba_last365",
+        export_tb=export_tb_rel,
+        col=[
+            "id",
+            "search_term",
+            "peak_month",
+            "all_year_text_flag",
+        ],
+        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) 会把正式表的索引复制到 copy 上，
+    #   其中 peak_month 是 gin 索引——只有 copy 的 peak_month 已经是数组类型，gin 索引才能被正确复制。
+    #   (dwt_aba_last365.py 那张宽表 peak_month 没有索引，所以它是换表后再转；本窄表 peak_month 带 gin，必须换表前转)
+    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：exchange_tb 自动把正式表现有的 4 个索引
+    #   (id / search_term / all_year_text_flag / peak_month gin) 复制到 copy 后再换入。
+    #   索引靠"正式表(手动建表时带好索引) → 每次换表自动传递"维护，无需手动重建（与其他导出脚本一致）。
+    DBUtil.exchange_tb(engine,
+                       source_tb_name=export_tb_rel,
+                       target_tb_name=export_tb_before,
+                       cp_index_flag=True)
+
+    # 更新 workflow_everyday：本窄表被前端另一个页面直接查询，需上报导出完成，供该页面"数据就绪"监控感知。
+    # page='AbaWordYearPeak'(新增页面标识，需同步给后端)；date_type 恒为 month；沿用 status_val=14 / '导出pg完成'。
+    # 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}', 'AbaWordYearPeak', '是',
+                    'ABA峰值月/常年可卖窄表', 'postgresql_cluster'
+                );
+            """
+            print("================================更新workflow_everyday================================")
+            print(sql)
+            connection.execute(sql)
+
+    print("success")
+