feature:本地归集增加可以指定文件的功能

Author: hefanli

backend/services/data-collection-service/src/main/java/com/dataengine/collection/interfaces/facade/CollectionTaskController.java

@@ -81,7 +81,11 @@ private String mapToJsonString(Map<String, Object> map, String taskId) {
             if (Objects.nonNull(map) && "DATAX".equals(map.get("type"))) {
                 // NFS相关校验和处理
                 map.put("destPath", "/dataset/local/" + taskId);
-                map.put("filePaths", Arrays.asList(map.get("destPath")));
+                if (map.containsKey("filePaths") && map.get("filePaths") != null && map.get("filePaths") instanceof List) {
+                    ((List) map.get("filePaths")).add(map.get("destPath"));
+                } else {
+                    map.put("filePaths", Collections.singletonList(map.get("destPath")));
+                }
                 return objectMapper.writeValueAsString(map);
             }
 

backend/shared/domain-common/src/main/java/com/dataengine/common/domain/model/ChunkUploadPreRequest.java

@@ -28,6 +28,7 @@ public class ChunkUploadPreRequest {
     public void incrementUploadedFileNum() {
         if (this.uploadedFileNum == null) {
             this.uploadedFileNum = 1;
+            return;
         }
         this.uploadedFileNum++;
     }

runtime/datax/nfswriter/src/main/java/com/modelengine/edatamate/plugin/writer/nfswriter/NfsWriter.java

@@ -7,7 +7,6 @@
 import com.alibaba.datax.common.spi.Writer;
 import com.alibaba.datax.common.util.Configuration;
 
-
 import org.apache.commons.lang3.StringUtils;
 
 import java.io.File;
@@ -62,12 +61,14 @@ public static class Task extends Writer.Task {
         private Configuration jobConfig;
         private String mountPoint;
         private String destPath;
+        private List<String> files;
 
         @Override
         public void init() {
             this.jobConfig = super.getPluginJobConf();
             this.destPath = this.jobConfig.getString("destPath");
             this.mountPoint = this.jobConfig.getString("mountPoint");
+            this.files = this.jobConfig.getList("files", Collections.emptyList(), String.class);
         }
 
         @Override
@@ -79,6 +80,9 @@ public void startWrite(RecordReceiver lineReceiver) {
                     if (StringUtils.isBlank(fileName)) {
                         continue;
                     }
+                    if (!files.isEmpty() && !files.contains(fileName)) {
+                        continue;
+                    }
 
                     String filePath = this.mountPoint + "/" + fileName;
                     ShellUtil.runCommand("rsync", Arrays.asList("--no-links", "--chmod=750", "--", filePath,
@@ -93,4 +97,4 @@ public void startWrite(RecordReceiver lineReceiver) {
         public void destroy() {
         }
     }
-}
+}