Merge branch 'develop_930' of github.com:ModelEngine-Group/data-platform into develop_930

Author: chenghh-9609

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTaskService.java

@@ -96,7 +96,7 @@ private void executeTask(CleaningTask task, CreateCleaningTaskRequest request, S
     private void prepareTask(CleaningTask task, List<OperatorInstance> instances) {
         TaskProcess process = new TaskProcess();
         process.setInstanceId(task.getId());
-        process.setDatasetPath(DATASET_PATH + "/" + task.getSrcDatasetId());
+        process.setDatasetPath(FLOW_PATH + "/" + task.getId() + "/dataset.jsonl");
         process.setExportPath(DATASET_PATH + "/" + task.getDestDatasetId());
         process.setExecutorType(ExecutorType.DATA_PLATFORM.getValue());
         process.setProcess(instances.stream()

runtime/python-executor/data_platform/core/constant.py

@@ -2,7 +2,7 @@
 
 class Fields(object):
     result = 'execute_result'
-    flow_id = 'flow_id'
+    instance_id = 'instance_id'
     export_path = 'export_path'
 
 

runtime/python-executor/data_platform/core/dataset.py

@@ -70,7 +70,7 @@ def preprocess_dataset(dataset: rd.Dataset, cfg) -> rd.Dataset:
 
     def process_batch_arrow(table: pa.Table, names_list=None) -> pa.Table:
         name2value_table = {
-            Fields.flow_id: cfg.flow_id,
+            Fields.instance_id: cfg.instance_id,
             Fields.export_path: cfg.export_path
         }
 

runtime/python-executor/data_platform/sqlite_manager/persistence_atction.py

@@ -11,12 +11,12 @@
 from data_platform.sqlite_manager.sqlite_manager import SQLiteManager
 
 
-def delete_task_info(flow_id: str):
-    db_path = f"/flow/{flow_id}.db"
+def delete_task_info(instance_id: str):
+    db_path = f"/flow/{instance_id}.db"
     try:
         os.remove(db_path)
     except Exception as e:
-        logger.warning(f"delete database for flow：{flow_id} error", e)
+        logger.warning(f"delete database for flow：{instance_id} error", e)
 
 
 class TaskInfoPersistence:
@@ -31,7 +31,7 @@ def load_sql_dict():
             return json.load(f)
 
     def persistence_task_info(self, sample: Dict[str, Any]):
-        flow_id = str(sample.get("flow_id"))
+        instance_id = str(sample.get("instance_id"))
         meta_file_name = str(sample.get("sourceFileName"))
         meta_file_type = str(sample.get("sourceFileType"))
         meta_file_id = int(sample.get("sourceFileId"))
@@ -49,17 +49,17 @@ def persistence_task_info(self, sample: Dict[str, Any]):
         incremental = str(sample.get("incremental") if sample.get("incremental") else '')
         child_id = sample.get("childId")
         slice_num = sample.get('slice_num', 0)
-        insert_data = [flow_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size,
+        insert_data = [instance_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size,
                        file_type, file_name, file_path, source_file_modify_time, status, operator_id, error_code,
                        incremental, child_id, slice_num]
-        self.insert_clean_result(insert_data, flow_id)
+        self.insert_clean_result(insert_data, instance_id)
 
-    def insert_clean_result(self, insert_data, flow_id):
+    def insert_clean_result(self, insert_data, instance_id):
         retries = 0
         max_retries = 20
         retry_delay = 1
         while retries <= max_retries:
-            db_path = f"/flow/{flow_id}.db"
+            db_path = f"/flow/{instance_id}.db"
             try:
                 insert_sql = str(self.sql_dict.get("insert_sql"))
                 create_tables_sql = str(self.sql_dict.get("create_tables_sql"))
@@ -78,21 +78,21 @@ def insert_clean_result(self, insert_data, flow_id):
                     raise RuntimeError(82000, str(e)) from None
         raise Exception("Max retries exceeded")
 
-    def query_task_info(self, flow_ids: list[str]):
+    def query_task_info(self, instance_ids: list[str]):
         result = {}
         create_tables_sql = self.sql_dict.get("create_tables_sql")
         query_sql = self.sql_dict.get("query_sql")
         current_result = None
-        for flow_id in flow_ids:
-            db_path = f"/flow/{flow_id}.db"
+        for instance_id in instance_ids:
+            db_path = f"/flow/{instance_id}.db"
             try:
                 with SQLiteManager.create_connect(db_path) as conn:
                     conn.execute("PRAGMA journal_mode=WAL")
                     conn.execute(create_tables_sql)
-                    cursor = conn.execute(query_sql, [flow_id])
+                    cursor = conn.execute(query_sql, [instance_id])
                     current_result = cursor.fetchall()
             except Exception as e:
-                logger.warning("flow_id: {}, query job result error: {}", flow_id, str(e))
+                logger.warning("instance_id: {}, query job result error: {}", instance_id, str(e))
             if current_result:
-                result[flow_id] = current_result
+                result[instance_id] = current_result
         return result

runtime/python-executor/data_platform/sqlite_manager/sql/sql_config.json

@@ -1,7 +1,7 @@
 {
-  "query_sql": "SELECT * FROM task_instance_info WHERE flow_id = ?",
+  "query_sql": "SELECT * FROM task_instance_info WHERE instance_id = ?",
   "db_path": "/flow/sqlite.db",
-  "insert_sql": "INSERT INTO task_instance_info (flow_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size, file_type, file_name, file_path, source_file_modify_time, status, operator_id, error_code, incremental, childId, slice_num) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-  "create_tables_sql": "CREATE TABLE IF NOT EXISTS task_instance_info (flow_id TEXT, meta_file_name TEXT, meta_file_type TEXT, meta_file_id INTEGER, meta_file_size TEXT, file_id INTEGER, file_size TEXT, file_type TEXT, file_name TEXT, file_path TEXT, source_file_modify_time INTEGER, status INTEGER, operator_id TEXT, error_code TEXT, incremental TEXT, childId INTEGER, slice_num INTEGER DEFAULT 0)",
-  "delete_task_instance_sql": "DELETE FROM task_instance_info WHERE flow_id = ?"
+  "insert_sql": "INSERT INTO task_instance_info (instance_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size, file_type, file_name, file_path, source_file_modify_time, status, operator_id, error_code, incremental, childId, slice_num) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+  "create_tables_sql": "CREATE TABLE IF NOT EXISTS task_instance_info (instance_id TEXT, meta_file_name TEXT, meta_file_type TEXT, meta_file_id INTEGER, meta_file_size TEXT, file_id INTEGER, file_size TEXT, file_type TEXT, file_name TEXT, file_path TEXT, source_file_modify_time INTEGER, status INTEGER, operator_id TEXT, error_code TEXT, incremental TEXT, childId INTEGER, slice_num INTEGER DEFAULT 0)",
+  "delete_task_instance_sql": "DELETE FROM task_instance_info WHERE instance_id = ?"
 }