feature:本地归集重复文件（按照路径判断是否为重复）做更新处理

hefanli · hefanli · commit 72261059eb89 · 2025-10-20T10:41:32.000+08:00
diff --git a/backend/services/data-collection-service/src/main/java/com/dataengine/collection/application/service/CollectionTaskService.java b/backend/services/data-collection-service/src/main/java/com/dataengine/collection/application/service/CollectionTaskService.java
@@ -6,26 +6,35 @@
 import com.dataengine.collection.domain.model.DataxTemplate;
 import com.dataengine.collection.infrastructure.persistence.mapper.CollectionTaskMapper;
 import com.dataengine.collection.infrastructure.persistence.mapper.TaskExecutionMapper;
+import com.dataengine.collection.interfaces.dto.SyncMode;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 import java.util.*;
 
+@Slf4j
 @Service
 @RequiredArgsConstructor
 public class CollectionTaskService {
     private final CollectionTaskMapper taskMapper;
     private final TaskExecutionMapper executionMapper;
+    private final DataxExecutionService dataxExecutionService;
 
     @Transactional
     public CollectionTask create(CollectionTask task) {
-        task.setId(UUID.randomUUID().toString());
         task.setStatus(TaskStatus.READY);
         task.setCreatedAt(LocalDateTime.now());
         task.setUpdatedAt(LocalDateTime.now());
         taskMapper.insert(task);
+        if (Objects.equals(task.getSyncMode(), SyncMode.ONCE.getValue())) {
+            TaskExecution exec = dataxExecutionService.createExecution(task);
+            int timeout = task.getTimeoutSeconds() == null ? 3600 : task.getTimeoutSeconds();
+            dataxExecutionService.runAsync(task, exec.getId(), timeout);
+            log.info("Triggered DataX execution for task {} at {}, execId={}", task.getId(), LocalDateTime.now(), exec.getId());
+        }
         return task;
     }
 
diff --git a/backend/services/data-collection-service/src/main/java/com/dataengine/collection/infrastructure/runtime/datax/DataxJobBuilder.java b/backend/services/data-collection-service/src/main/java/com/dataengine/collection/infrastructure/runtime/datax/DataxJobBuilder.java
@@ -1,19 +1,26 @@
 package com.dataengine.collection.infrastructure.runtime.datax;
 
 import com.dataengine.collection.domain.model.CollectionTask;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Component;
 
-import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 /**
  * 根据任务配置拼装 DataX 作业 JSON 文件
  */
+@Slf4j
 @Component
 @RequiredArgsConstructor
 public class DataxJobBuilder {
@@ -28,11 +35,47 @@ public Path buildJobFile(CollectionTask task) throws IOException {
         try (FileWriter fw = new FileWriter(path.toFile())) {
             String json = task.getConfig() == null || task.getConfig().isEmpty() ?
                     defaultJobJson() : task.getConfig();
+            if (StringUtils.isNotBlank(task.getConfig())) {
+                json = getJobConfig(task);
+            }
+            log.info("Job config: {}", json);
             fw.write(json);
         }
         return path;
     }
 
+    private String getJobConfig(CollectionTask task) {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            Map<String, Object> parameter = objectMapper.readValue(
+                task.getConfig(),
+                new TypeReference<>() {}
+            );
+            Map<String, Object> job = new HashMap<>();
+            Map<String, Object> content = new HashMap<>();
+            Map<String, Object> reader = new HashMap<>();
+            reader.put("name", "nfsreader");
+            reader.put("parameter", parameter);
+            content.put("reader", reader);
+            Map<String, Object> writer = new HashMap<>();
+            writer.put("name", "nfswriter");
+            writer.put("parameter", parameter);
+            content.put("writer", writer);
+            job.put("content", List.of(content));
+            Map<String, Object> setting = new HashMap<>();
+            Map<String, Object> channel = new HashMap<>();
+            channel.put("channel", 2);
+            setting.put("speed", channel);
+            job.put("setting", setting);
+            Map<String, Object> jobConfig = new HashMap<>();
+            jobConfig.put("job", job);
+            return objectMapper.writeValueAsString(jobConfig);
+        } catch (Exception e) {
+            log.error("Failed to parse task config", e);
+            throw new RuntimeException("Failed to parse task config", e);
+        }
+    }
+
     private String defaultJobJson() {
         // 提供一个最小可运行的空 job，实际会被具体任务覆盖
         return "{\n  \"job\": {\n    \"setting\": {\n      \"speed\": {\n        \"channel\": 1\n      }\n    },\n    \"content\": []\n  }\n}";
diff --git a/backend/services/data-collection-service/src/main/java/com/dataengine/collection/interfaces/facade/CollectionTaskController.java b/backend/services/data-collection-service/src/main/java/com/dataengine/collection/interfaces/facade/CollectionTaskController.java
@@ -11,8 +11,7 @@
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.RestController;
 
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 
 @RestController
@@ -73,8 +72,19 @@ private Map<String, Object> parseJsonToMap(String json) {
         }
     }
 
-    private String mapToJsonString(Map<String, Object> map) {
+    private String mapToJsonString(Map<String, Object> map, String taskId) {
         try {
+            if (Objects.nonNull(map) && "LOCAL_COLLECTION".equals(map.get("type"))) {
+                // 本地归集相关校验和处理
+                return objectMapper.writeValueAsString(map);
+            }
+            if (Objects.nonNull(map) && "DATAX".equals(map.get("type"))) {
+                // NFS相关校验和处理
+                map.put("destPath", "/dataset/local/" + taskId);
+                map.put("filePaths", Arrays.asList(map.get("destPath")));
+                return objectMapper.writeValueAsString(map);
+            }
+
             return objectMapper.writeValueAsString(map != null ? map : Map.of());
         } catch (Exception e) {
             return "{}";
@@ -84,9 +94,10 @@ private String mapToJsonString(Map<String, Object> map) {
     @Override
     public ResponseEntity<CollectionTaskResponse> createTask(CreateCollectionTaskRequest body) {
         CollectionTask t = new CollectionTask();
+        t.setId(UUID.randomUUID().toString());
         t.setName(body.getName());
         t.setDescription(body.getDescription());
-        t.setConfig(mapToJsonString(body.getConfig()));
+        t.setConfig(mapToJsonString(body.getConfig(), t.getId()));
         if (body.getSyncMode() != null) { t.setSyncMode(body.getSyncMode().getValue()); }
         t.setScheduleExpression(body.getScheduleExpression());
         t = taskService.create(t);
@@ -102,7 +113,7 @@ public ResponseEntity<CollectionTaskResponse> updateTask(String id, UpdateCollec
         t.setId(id);
         t.setName(body.getName());
         t.setDescription(body.getDescription());
-        t.setConfig(mapToJsonString(body.getConfig()));
+        t.setConfig(mapToJsonString(body.getConfig(), t.getId()));
         if (body.getSyncMode() != null) { t.setSyncMode(body.getSyncMode().getValue()); }
         t.setScheduleExpression(body.getScheduleExpression());
         t = taskService.update(t);
diff --git a/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/application/DatasetApplicationService.java b/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/application/DatasetApplicationService.java
@@ -281,13 +281,13 @@ public void processDataSourceAsync(String datasetId, String dataSourceId) {
             log.info("开始处理数据源文件扫描，数据集ID: {}, 数据源ID: {}", datasetId, dataSourceId);
 
             // 1. 调用数据归集服务获取任务详情
-            CollectionTaskDetailResponse taskDetail = collectionTaskClient.getTaskDetail(dataSourceId);
+            CollectionTaskDetailResponse taskDetail = collectionTaskClient.getTaskDetail(dataSourceId).getData();
             if (taskDetail == null) {
                 log.error("获取归集任务详情失败，任务ID: {}", dataSourceId);
                 return;
             }
 
-            log.info("获取到归集任务详情: {}", taskDetail.getName());
+            log.info("获取到归集任务详情: {}", taskDetail);
 
             // 2. 解析任务配置
             LocalCollectionConfig config = parseTaskConfig(taskDetail.getConfig());
@@ -296,12 +296,6 @@ public void processDataSourceAsync(String datasetId, String dataSourceId) {
                 return;
             }
 
-            // 3. 检查任务类型是否为 LOCAL_COLLECTION
-            if (!"LOCAL_COLLECTION".equalsIgnoreCase(config.getType())) {
-                log.info("任务类型不是 LOCAL_COLLECTION，跳过文件扫描。任务类型: {}", config.getType());
-                return;
-            }
-
             // 4. 获取文件路径列表
             List<String> filePaths = config.getFilePaths();
             if (CollectionUtils.isEmpty(filePaths)) {
diff --git a/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/infrastructure/client/CollectionTaskClient.java b/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/infrastructure/client/CollectionTaskClient.java
@@ -1,5 +1,6 @@
 package com.dataengine.datamanagement.infrastructure.client;
 
+import com.dataengine.common.infrastructure.common.Response;
 import com.dataengine.datamanagement.infrastructure.client.dto.CollectionTaskDetailResponse;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -17,5 +18,5 @@ public interface CollectionTaskClient {
      * @return 任务详情
      */
     @GetMapping("/api/data-collection/tasks/{id}")
-    CollectionTaskDetailResponse getTaskDetail(@PathVariable("id") String taskId);
+    Response<CollectionTaskDetailResponse> getTaskDetail(@PathVariable("id") String taskId);
 }
diff --git a/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/infrastructure/client/dto/LocalCollectionConfig.java b/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/infrastructure/client/dto/LocalCollectionConfig.java
@@ -18,10 +18,4 @@ public class LocalCollectionConfig {
      * 文件路径列表
      */
     private List<String> filePaths;
-
-    /**
-     * 其他配置项
-     */
-    private String sourceType;
-    private String targetType;
 }
diff --git a/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/interfaces/rest/DatasetController.java b/backend/services/data-management-service/src/main/java/com/dataengine/datamanagement/interfaces/rest/DatasetController.java
@@ -7,6 +7,7 @@
 import com.dataengine.datamanagement.domain.model.dataset.Dataset;
 import com.dataengine.datamanagement.interfaces.converter.DatasetConverter;
 import com.dataengine.datamanagement.interfaces.dto.*;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Page;
 import org.springframework.http.HttpStatus;
@@ -18,6 +19,7 @@
 /**
  * 数据集 REST 控制器（UUID 模式）
  */
+@Slf4j
 @RestController
 @RequestMapping("/data-management/datasets")
 public class DatasetController {
@@ -47,6 +49,7 @@ public ResponseEntity<Response<DatasetResponse>> createDataset(@RequestBody Crea
             Dataset dataset = datasetApplicationService.createDataset(createDatasetRequest);
             return ResponseEntity.status(HttpStatus.CREATED).body(Response.ok(DatasetConverter.INSTANCE.convertToResponse(dataset)));
         } catch (IllegalArgumentException e) {
+            log.error("Failed to create dataset", e);
             return ResponseEntity.badRequest().body(Response.error(SystemErrorCode.UNKNOWN_ERROR, null));
         }
     }
diff --git a/backend/services/main-application/src/main/resources/config/application-datacollection.yml b/backend/services/main-application/src/main/resources/config/application-datacollection.yml
@@ -2,7 +2,7 @@ dataengine:
   data-collection:
     # DataX配置
     datax:
-      home-path: ${DATAX_HOME:D:/datax}
+      home-path: ${DATAX_HOME:/opt/datax}
       python-path: ${DATAX_PYTHON_PATH:python3}
       job-config-path: ${DATAX_JOB_PATH:./data/temp/datax/jobs}
       log-path: ${DATAX_LOG_PATH:./logs/datax}
diff --git a/backend/shared/domain-common/src/main/java/com/dataengine/common/infrastructure/common/Response.java b/backend/shared/domain-common/src/main/java/com/dataengine/common/infrastructure/common/Response.java
@@ -3,6 +3,7 @@
 import com.dataengine.common.infrastructure.exception.ErrorCode;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 import lombok.Setter;
 
 import java.io.Serial;
@@ -13,6 +14,7 @@
  */
 @Getter
 @Setter
+@NoArgsConstructor
 @AllArgsConstructor
 public class Response<T> implements Serializable {
     @Serial
diff --git a/deployment/docker/data-platform/docker-compose.yml b/deployment/docker/data-platform/docker-compose.yml
@@ -4,6 +4,7 @@ services:
     container_name: backend
     image: backend
     restart: on-failure
+    privileged: true
     ports:
       - "8080:8080"
     volumes:
