[hotfix-#1745][connector][hdfs][hive] Data loss occurs when multiple …

…tasks write to the same hdfs (or hive) data source
DTStack · Jul 19, 2023 · c9c8c8a · c9c8c8a
1 parent a05900e
commit c9c8c8a
Show file tree

Hide file tree

Showing 5 changed files with 17 additions and 1 deletion.
diff --git a/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/table/HdfsDynamicTableFactory.java b/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/table/HdfsDynamicTableFactory.java
@@ -135,6 +135,7 @@ private HdfsConfig getHdfsConfig(ReadableConfig config) {
         hdfsConfig.setEncoding(config.get(BaseFileOptions.ENCODING));
         hdfsConfig.setMaxFileSize(config.get(BaseFileOptions.MAX_FILE_SIZE));
         hdfsConfig.setNextCheckRows(config.get(BaseFileOptions.NEXT_CHECK_ROWS));
+        hdfsConfig.setJobTimeStamp(config.get(BaseFileOptions.JOB_TIMESTAMP));
 
         hdfsConfig.setDefaultFS(config.get(HdfsOptions.DEFAULT_FS));
         hdfsConfig.setFileType(config.get(HdfsOptions.FILE_TYPE));

diff --git a/...-hive/src/main/java/com/dtstack/chunjun/connector/hive/table/HiveDynamicTableFactory.java b/...-hive/src/main/java/com/dtstack/chunjun/connector/hive/table/HiveDynamicTableFactory.java
@@ -228,6 +228,7 @@ private HiveConfig getHiveConf(ReadableConfig config) {
         hiveConf.setEncoding(config.get(BaseFileOptions.ENCODING));
         hiveConf.setMaxFileSize(config.get(BaseFileOptions.MAX_FILE_SIZE));
         hiveConf.setNextCheckRows(config.get(BaseFileOptions.NEXT_CHECK_ROWS));
+        hiveConf.setJobTimeStamp(config.get(BaseFileOptions.JOB_TIMESTAMP));
 
         hiveConf.setDefaultFS(config.get(HdfsOptions.DEFAULT_FS));
         hiveConf.setFileType(config.get(HdfsOptions.FILE_TYPE));

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/config/BaseFileConfig.java b/chunjun-core/src/main/java/com/dtstack/chunjun/config/BaseFileConfig.java
@@ -50,4 +50,6 @@ public class BaseFileConfig extends CommonConfig {
     private long nextCheckRows = 5000;
 
     private String suffix;
+
+    private String jobTimeStamp = "";
 }
diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java b/chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java
@@ -107,7 +107,12 @@ protected void initVariableFields() {
         } else {
             outputFilePath = baseFileConfig.getPath();
         }
-        tmpPath = outputFilePath + File.separatorChar + TMP_DIR_NAME;
+        tmpPath =
+                outputFilePath
+                        + File.separatorChar
+                        + TMP_DIR_NAME
+                        + baseFileConfig.getJobTimeStamp();
+        log.info("initVariableFields get tmpPath: {}", tmpPath);
         nextNumForCheckDataSize = baseFileConfig.getNextCheckRows();
         openSource();
     }

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/table/options/BaseFileOptions.java b/chunjun-core/src/main/java/com/dtstack/chunjun/table/options/BaseFileOptions.java
@@ -62,4 +62,11 @@ public class BaseFileOptions {
                     .longType()
                     .defaultValue(5000L)
                     .withDescription("The number of data written in the next file size check");
+
+    public static final ConfigOption<String> JOB_TIMESTAMP =
+            ConfigOptions.key("properties.job-timestamp")
+                    .stringType()
+                    .defaultValue(String.valueOf(System.currentTimeMillis()))
+                    .withDescription(
+                            "To solve the problem of file loss caused by multiple tasks writing to the same output source at the same time");
 }