分布式表修改导出方式

Pyspark_job/sqoop_export/dwt_aba_st_analytics_report_pg.py

@@ -276,6 +276,16 @@ if __name__ == '__main__':
                                       """
     DBUtil.engine_exec_sql(engine, sql)
 
+    # 新一年母表是 like 上一年建的,like 不带分布式属性;上一年已是分布式而本年不是时,把本年母表(刚建为空)转成分布式,否则数据又落回主节点
+    if date_type == DateTypes.month.name and db_type == 'postgresql_cluster':
+        with engine.connect() as conn:
+            dist_tbs = [r[0] for r in conn.execute(f"select logicalrelid::text from pg_dist_partition where logicalrelid::text in ('{export_tb_before}', '{export_master_tb}')").fetchall()]
+        if export_tb_before in dist_tbs and export_master_tb not in dist_tbs:
+            DBUtil.engine_exec_sql(engine, f"""
+                set local citus.shard_replication_factor = 1;
+                select create_distributed_table('{export_master_tb}', 'search_term_id');
+            """)
+
     if test_flag == "month_append":
         partition_dict = {
             "site_name": site_name,

Pyspark_job/sqoop_export/dwt_fb_asin_info.py

@@ -46,16 +46,35 @@ if __name__ == '__main__':
     # 跨年自动建当年母表：用上一年母表 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}
-                (
-                like {export_master_tb}  including comments
-         );
+    """
+    DBUtil.engine_exec_sql(engine, sql)
+
+    # 新一年母表是 like 上一年建的,like 不带分布式属性;上一年已是分布式而本年不是时,把本年母表(刚建为空)转成分布式,否则数据又落回主节点
+    if db_type == 'postgresql_cluster':
+        with engine.connect() as conn:
+            dist_tbs = [r[0] for r in conn.execute(f"select logicalrelid::text from pg_dist_partition where logicalrelid::text in ('{export_master_tb_before}', '{export_master_tb}')").fetchall()]
+        if export_master_tb_before in dist_tbs and export_master_tb not in dist_tbs:
+            DBUtil.engine_exec_sql(engine, f"""
+                set local citus.shard_replication_factor = 1;
+                select create_distributed_table('{export_master_tb}', 'seller_id');
+            """)
+
+    # 当月子表直接建在母表下再清空,保证幂等:母表是分布式时子表自动分布式,sqoop 数据直接写到 worker,主节点不留数据
+    # 原写法建普通表 sqoop 后再 attach,Citus 复制到分片但主节点本地那份不删(实测 nsr 每张月表残留约 1GB),已改
+    # sql = f"""
+    #      drop table if exists {export_tb};
+    #      create table if not exists {export_tb}
+    #             (
+    #             like {export_master_tb}  including comments
+    #      );
+    # """
+    sql = f"""
+         create table if not exists {export_tb} partition of {export_master_tb} for values from ('{date_info}') to ('{next_month}');
+         truncate table {export_tb};
     """
     DBUtil.engine_exec_sql(engine, sql)
 
@@ -97,17 +116,17 @@ if __name__ == '__main__':
 
 
     client.close()
-    # 创建索引并交换分区
-    DBUtil.add_pg_part(
-        engine,
-        source_tb_name=export_tb,
-        part_master_tb=export_master_tb,
-        part_val={
-            "from": [date_info],
-            "to": [next_month]
-        },
-        cp_index_flag=False,
-    )
+    # 子表已直接建在母表下,不再需要挂载分区
+    # DBUtil.add_pg_part(
+    #     engine,
+    #     source_tb_name=export_tb,
+    #     part_master_tb=export_master_tb,
+    #     part_val={
+    #         "from": [date_info],
+    #         "to": [next_month]
+    #     },
+    #     cp_index_flag=False,
+    # )
 
     # 关闭链接
     engine.dispose()

Pyspark_job/sqoop_export/dwt_fb_top20_asin_info.py

@@ -44,16 +44,35 @@ if __name__ == '__main__':
     # 跨年自动建当年母表：用上一年母表 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}
-                (
-                like {export_master_tb}  including comments
-         );
+    """
+    DBUtil.engine_exec_sql(engine, sql)
+
+    # 新一年母表是 like 上一年建的,like 不带分布式属性;上一年已是分布式而本年不是时,把本年母表(刚建为空)转成分布式,否则数据又落回主节点
+    if db_type == 'postgresql_cluster':
+        with engine.connect() as conn:
+            dist_tbs = [r[0] for r in conn.execute(f"select logicalrelid::text from pg_dist_partition where logicalrelid::text in ('{export_master_tb_before}', '{export_master_tb}')").fetchall()]
+        if export_master_tb_before in dist_tbs and export_master_tb not in dist_tbs:
+            DBUtil.engine_exec_sql(engine, f"""
+                set local citus.shard_replication_factor = 1;
+                select create_distributed_table('{export_master_tb}', 'seller_id');
+            """)
+
+    # 当月子表直接建在母表下再清空,保证幂等:母表是分布式时子表自动分布式,sqoop 数据直接写到 worker,主节点不留数据
+    # 原写法建普通表 sqoop 后再 attach,Citus 复制到分片但主节点本地那份不删(实测 nsr 每张月表残留约 1GB),已改
+    # sql = f"""
+    #      drop table if exists {export_tb};
+    #      create table if not exists {export_tb}
+    #             (
+    #             like {export_master_tb}  including comments
+    #      );
+    # """
+    sql = f"""
+         create table if not exists {export_tb} partition of {export_master_tb} for values from ('{date_info}') to ('{next_month}');
+         truncate table {export_tb};
     """
     DBUtil.engine_exec_sql(engine, sql)
 
@@ -93,17 +112,17 @@ if __name__ == '__main__':
 
 
     client.close()
-    # 创建索引并交换分区
-    DBUtil.add_pg_part(
-        engine,
-        source_tb_name=export_tb,
-        part_master_tb=export_master_tb,
-        part_val={
-            "from": [date_info],
-            "to": [next_month]
-        },
-        cp_index_flag=False,
-    )
+    # 子表已直接建在母表下,不再需要挂载分区
+    # DBUtil.add_pg_part(
+    #     engine,
+    #     source_tb_name=export_tb,
+    #     part_master_tb=export_master_tb,
+    #     part_val={
+    #         "from": [date_info],
+    #         "to": [next_month]
+    #     },
+    #     cp_index_flag=False,
+    # )
 
     update_workflow_sql = f"""                
                         update selection.workflow_progress set `status`='导出pg集群完成', status_val=6, over_date=CURRENT_TIME, is_end='是' where page='店铺Feedback'