fix

Pyspark_job/sqoop_export/export_dwt_asin_related_traffic.py

@@ -34,7 +34,8 @@ if __name__ == '__main__':
         report_type = date_type
         report_date = date_info
     elif date_type == 'month_week':
-        export_tb = f"{site_name}_asin_related_30_day"
+        export_tb = f"{site_name}_asin_related_30_day_copy"
+        export_tb_target = f"{site_name}_asin_related_30_day"
         report_type = '30_day'
         report_date = date.today()
     else:
@@ -85,17 +86,27 @@ if __name__ == '__main__':
     DBUtil.engine_exec_sql(engine, sql)
     print("数据类型修改完成，准备创建索引！")
 
-    sql = f"""
-        CREATE INDEX {export_tb}_asin_idx ON {export_tb} USING btree (
-          "asin" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
-        );
+    if date_type == 'month':
+        sql = f"""
+            CREATE INDEX {export_tb}_asin_idx ON {export_tb} USING btree (
+              "asin" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
+            );
 
-        CREATE INDEX {export_tb}_related_asin_idx ON {export_tb} USING gin (
-          "related_asin" COLLATE "pg_catalog"."default" "pg_catalog"."array_ops"
-        );
-    """
-    DBUtil.engine_exec_sql(engine, sql)
-    print("索引创建完成，准备插入流程记录表！")
+            CREATE INDEX {export_tb}_related_asin_idx ON {export_tb} USING gin (
+              "related_asin" COLLATE "pg_catalog"."default" "pg_catalog"."array_ops"
+            );
+        """
+        DBUtil.engine_exec_sql(engine, sql)
+        print("索引创建完成，准备插入流程记录表！")
+    elif date_type == 'month_week':
+        # 构建索引并交换表名
+        DBUtil.exchange_tb(
+            engine=engine,
+            source_tb_name=export_tb,
+            target_tb_name=export_tb_target,
+            cp_index_flag=True
+        )
+        print("索引创建完成，准备插入流程记录表！")
 
     sql = f"""
         REPLACE INTO selection.workflow_everyday