fix

Pyspark_job/sqoop_export/dwt_amazon_report_pg.py

@@ -23,25 +23,35 @@ if __name__ == '__main__':
     engine = DBUtil.get_db_engine(db_type, site_name)
 
     with engine.connect() as connection:
+        # 不再用 like {master_tb} including all 复制结构：显式建表，新字段直接写进结构里，
+        # 不需要对线上分布式表 master_tb 做 ALTER TABLE ADD COLUMN（容易触发 Citus 分布式死锁）；
+        # 新字段随着导出完成后的换表流程一起生效。建表时不建索引（导出完成后由 exchange_tb
+        # 从 master_tb 复制现有索引重建，整表建索引比逐行插入维护索引快很多）
         sql = f"""
-            drop table if exists {export_tb};
-            create table if not exists {export_tb} 
-            (
-            like {master_tb} including ALL       
+            DROP TABLE IF EXISTS "public"."{export_tb}";
+            CREATE TABLE "public"."{export_tb}" (
+              "asin" varchar(20) COLLATE "pg_catalog"."default" NOT NULL,
+              "monthly_sales" text COLLATE "pg_catalog"."default" NOT NULL,
+              "zr_count" text COLLATE "pg_catalog"."default" NOT NULL,
+              "sp_count" text COLLATE "pg_catalog"."default" NOT NULL,
+              "total_count" text COLLATE "pg_catalog"."default" NOT NULL,
+              "date_info_list" text COLLATE "pg_catalog"."default" NOT NULL,
+              "created_at" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+              "updated_at" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+              "weekly_sales" text COLLATE "pg_catalog"."default",
+              "weekly_views" text COLLATE "pg_catalog"."default",
+              "monthly_views" text COLLATE "pg_catalog"."default",
+              "variation_monthly_sales" text COLLATE "pg_catalog"."default" NOT NULL
             );
+            COMMENT ON COLUMN "public"."{export_tb}"."asin" IS 'asin';
+            COMMENT ON COLUMN "public"."{export_tb}"."monthly_sales" IS 'Amazon月销';
+            COMMENT ON COLUMN "public"."{export_tb}"."zr_count" IS 'zr词数';
+            COMMENT ON COLUMN "public"."{export_tb}"."sp_count" IS 'sp词数';
+            COMMENT ON COLUMN "public"."{export_tb}"."total_count" IS '总词数';
+            COMMENT ON COLUMN "public"."{export_tb}"."date_info_list" IS 'date_info';
+            COMMENT ON COLUMN "public"."{export_tb}"."variation_monthly_sales" IS '变体销量月度历史(逗号分隔,对齐date_info_list)';
 
             SELECT create_distributed_table('{export_tb}', 'asin');
-
-            DO $$
-            DECLARE
-                index_record RECORD;
-            BEGIN
-                FOR index_record IN (SELECT indexname FROM pg_indexes WHERE tablename = '{export_tb}') 
-                LOOP
-                    EXECUTE 'DROP INDEX IF EXISTS ' || index_record.indexname;
-                END LOOP;
-            END;
-            $$;
         """
         connection.execute(sql)