流量选品-月销趋势图导出优化

Pyspark_job/sqoop_export/dwt_amazon_report_pg.py

@@ -24,24 +24,30 @@ if __name__ == '__main__':
 
     with engine.connect() as connection:
         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)