修改 bsr nsr榜单导出方式

Pyspark_job/sqoop_export/dwt_bsr_nsr_asin_detail_new.py

@@ -5,6 +5,7 @@ sys.path.append(os.path.dirname(sys.path[0]))
 from utils.db_util import DBUtil, DbTypes
 from utils.ssh_util import SSHUtil
 from utils.common_util import CommonUtil
+from utils.secure_db_client import get_remote_engine
 
 
 def export_postgresql():
@@ -45,13 +46,8 @@ def export_postgresql_cluster():
     DBUtil.exec_sql(db_type=db_type, site_name=site_name, sql=sql, dispose_flag=True)
 
     #  Sqoop 导出需要一段时间，如果直接写正式分区表，导出期间业务查询可能读到一半的脏数据。用 copy 表先接收数据，全部就绪后再一次性切换到正式表，
-    # 导出表名 导出到copy表
-    sh = CommonUtil.build_export_sh(
-        site_name=site_name,
-        db_type=db_type,
-        hive_tb=f"dwt_{tb_type}_asin_detail",
-        export_tb=export_tb_copy,
-        col=[
+    # 导出字段，顺序必须和 Hive 表 dwt_{tb_type}_asin_detail 一致
+    cols_list = [
         "asin",
         "title",
         "img_url",
@@ -83,31 +79,76 @@ def export_postgresql_cluster():
         "asin_launch_time_type",
         "seller_country_type",
         "asin_bought_month",
-        ],
-        partition_dict={
+    ]
+
+    # 旧写法：本地拼 sqoop 命令再 SSH 到集群执行
+    # sh = CommonUtil.build_export_sh(
+    #     site_name=site_name,
+    #     db_type=db_type,
+    #     hive_tb=f"dwt_{tb_type}_asin_detail",
+    #     export_tb=export_tb_copy,
+    #     col=cols_list,
+    #     partition_dict={
+    #         "site_name": site_name,
+    #         "date_info": rel_date_info
+    #     }
+    # )
+    # client = SSHUtil.get_ssh_client()
+    # SSHUtil.exec_command_async(client, sh, ignore_err=False)
+    # client.close()
+
+    # 导出表名 导出到copy表，m=6 保持和原来一样的 map 数
+    remote_engine = get_remote_engine(
+        site_name=site_name,
+        db_type="postgresql_cluster"
+    )
+    remote_engine.sqoop_raw_export(
+        hive_table=f"dwt_{tb_type}_asin_detail",
+        import_table=export_tb_copy,
+        partitions={
             "site_name": site_name,
             "date_info": rel_date_info
-        }
+        },
+        m=6,
+        cols=",".join(cols_list)
     )
 
-    client = SSHUtil.get_ssh_client()
-    SSHUtil.exec_command_async(client, sh, ignore_err=False)
-    client.close()
-
     # 导出表 copy 表替换正式分区
-    engine = DBUtil.get_db_engine(db_type, site_name)
-    DBUtil.exchange_pg_part_distributed_tb(
-        engine=engine,
-        source_tb_name=export_tb_copy, # 数据来源：copy 表
-        part_master_tb=export_master_tb, # 主表
-        part_target_tb=export_tb, # 正式月分区子表
-        part_val={ #  part_val = {'from': ['2026-04'], 'to': ['2026-05']},  # 分区范围
-            "from": [rel_date_info],
-            "to": [next_month]
-        },
-        drop_old=True
-    )
-    engine.dispose()
+    # 旧写法：DBUtil 的 SQLAlchemy engine，5 条 DDL 连着跑在同一个连接的事务里，
+    # Citus 集群会在事务内攒下不同用户建立的 placement 连接，detach 分区时报
+    # cannot perform query on placements that were modified in this transaction by a different user
+    # engine = DBUtil.get_db_engine(db_type, site_name)
+    # DBUtil.exchange_pg_part_distributed_tb(
+    #     engine=engine,
+    #     source_tb_name=export_tb_copy, # 数据来源：copy 表
+    #     part_master_tb=export_master_tb, # 主表
+    #     part_target_tb=export_tb, # 正式月分区子表
+    #     part_val={ #  part_val = {'from': ['2026-04'], 'to': ['2026-05']},  # 分区范围
+    #         "from": [rel_date_info],
+    #         "to": [next_month]
+    #     },
+    #     drop_old=True
+    # )
+    # engine.dispose()
+    df_part = remote_engine.read_sql(f"""
+        select relid::varchar as relid from pg_partition_tree('{export_master_tb}') where relid::varchar = '{export_tb}'
+    """)
+    exist_flag = not df_part.empty
+    print(f"分区{export_tb}是否已存在：{exist_flag}")
+
+    if exist_flag:
+        # 旧分区从主表摘下来并改名成 _back，让出正式分区表名
+        remote_engine.execute(f"""alter table {export_master_tb} detach partition {export_tb}""")
+        remote_engine.execute(f"""alter table {export_tb} rename to {export_tb}_back""")
+
+    # 装满新数据的 copy 表改成正式分区名，再挂回主表，业务查询由此切到新数据
+    remote_engine.execute(f"""alter table {export_tb_copy} rename to {export_tb}""")
+    remote_engine.execute(f"""alter table {export_master_tb} attach partition {export_tb} for values from ('{rel_date_info}') to ('{next_month}')""")
+
+    # 挂载成功后旧分区数据不再需要
+    remote_engine.execute(f"""drop table if exists {export_tb}_back""")
+
+    print(f"==================表{export_tb_copy}加入成功==================================")
     print("success")