add unstructed field type 'INT'

qlangtech · Oct 18, 2022 · 23e12a2 · 23e12a2
1 parent 9785260
commit 23e12a2
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 21 deletions.
diff --git a/ftpwriter/src/main/java/com/alibaba/datax/plugin/writer/ftpwriter/FtpWriter.java b/ftpwriter/src/main/java/com/alibaba/datax/plugin/writer/ftpwriter/FtpWriter.java
@@ -4,7 +4,6 @@
 import com.alibaba.datax.common.plugin.RecordReceiver;
 import com.alibaba.datax.common.spi.Writer;
 import com.alibaba.datax.common.util.Configuration;
-import com.alibaba.datax.common.util.RetryUtil;
 import com.alibaba.datax.plugin.ftp.common.FtpHelper;
 import com.alibaba.datax.plugin.unstructuredstorage.writer.UnstructuredStorageWriterUtil;
 import com.alibaba.datax.plugin.writer.ftpwriter.util.Constant;
@@ -17,7 +16,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.Callable;
 
 public class FtpWriter extends Writer {
     public static class Job extends Writer.Job {

diff --git a/...e-util/src/main/java/com/alibaba/datax/plugin/unstructuredstorage/reader/ColumnEntry.java b/...e-util/src/main/java/com/alibaba/datax/plugin/unstructuredstorage/reader/ColumnEntry.java
@@ -1,18 +1,18 @@
 package com.alibaba.datax.plugin.unstructuredstorage.reader;
 
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-
+import com.alibaba.fastjson.JSON;
 import org.apache.commons.lang3.StringUtils;
 
-import com.alibaba.fastjson.JSON;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
 
 public class ColumnEntry {
     private Integer index;
     private String type;
     private String value;
     private String format;
     private DateFormat dateParse;
+    private UnstructuredStorageReaderUtil.Type ctype;
 
     public Integer getIndex() {
         return index;
@@ -22,12 +22,13 @@ public void setIndex(Integer index) {
         this.index = index;
     }
 
-    public String getType() {
-        return type;
+    public UnstructuredStorageReaderUtil.Type getCType() {
+        return ctype;
     }
 
     public void setType(String type) {
         this.type = type;
+        this.ctype = UnstructuredStorageReaderUtil.Type.valueOf(type.toUpperCase());
     }
 
     public String getValue() {

diff --git a/...va/com/alibaba/datax/plugin/unstructuredstorage/reader/UnstructuredStorageReaderUtil.java b/...va/com/alibaba/datax/plugin/unstructuredstorage/reader/UnstructuredStorageReaderUtil.java
@@ -9,7 +9,6 @@
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.TypeReference;
-import com.csvreader.CsvReader;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -230,7 +229,7 @@ public static void doReadFromStream(BufferedReader reader, String context,
                                         TaskPluginCollector taskPluginCollector) {
         String encoding = readerSliceConfig.getString(Key.ENCODING,
                 Constant.DEFAULT_ENCODING);
-      //  Character fieldDelimiter = null;
+        //  Character fieldDelimiter = null;
 //        String delimiterInStr = readerSliceConfig
 //                .getString(Key.FIELD_DELIMITER);
 //        if (null != delimiterInStr && 1 != delimiterInStr.length()) {
@@ -259,9 +258,7 @@ public static void doReadFromStream(BufferedReader reader, String context,
                 .getListColumnEntry(readerSliceConfig, Key.COLUMN);
 
 
-
-
-      // CsvReader csvReader = null;
+        // CsvReader csvReader = null;
 
         // every line logic
         try {
@@ -306,7 +303,7 @@ public static void doReadFromStream(BufferedReader reader, String context,
                     UnstructuredStorageReaderErrorCode.RUNTIME_EXCEPTION,
                     String.format("运行时异常 : %s", e.getMessage()), e);
         } finally {
-          //  csvReader.close();
+            //  csvReader.close();
             IOUtils.closeQuietly(reader);
         }
     }
@@ -359,10 +356,13 @@ public static Record transportOneRecord(RecordSender recordSender,
             recordSender.sendToWriter(record);
         } else {
             try {
+                Type type = null;
+                Integer columnIndex;
+                String columnConst;
                 for (ColumnEntry columnConfig : columnConfigs) {
-                    String columnType = columnConfig.getType();
-                    Integer columnIndex = columnConfig.getIndex();
-                    String columnConst = columnConfig.getValue();
+                    type = columnConfig.getCType();
+                    columnIndex = columnConfig.getIndex();
+                    columnConst = columnConfig.getValue();
 
                     String columnValue = null;
 
@@ -394,7 +394,7 @@ public static Record transportOneRecord(RecordSender recordSender,
                     } else {
                         columnValue = columnConst;
                     }
-                    Type type = Type.valueOf(columnType.toUpperCase());
+                    // Type type = Type.valueOf(columnType.toUpperCase());
                     // it's all ok if nullFormat is null
                     if (columnValue.equals(nullFormat)) {
                         columnValue = null;
@@ -403,6 +403,7 @@ public static Record transportOneRecord(RecordSender recordSender,
                         case STRING:
                             columnGenerated = new StringColumn(columnValue);
                             break;
+                        case INT:
                         case LONG:
                             try {
                                 columnGenerated = new LongColumn(columnValue);
@@ -460,7 +461,7 @@ public static Record transportOneRecord(RecordSender recordSender,
                             break;
                         default:
                             String errorMessage = String.format(
-                                    "您配置的列类型暂不支持 : [%s]", columnType);
+                                    "您配置的列类型暂不支持 : [%s]", type);
                             LOG.error(errorMessage);
                             throw DataXException
                                     .asDataXException(
@@ -504,8 +505,8 @@ public static List<ColumnEntry> getListColumnEntry(
         return result;
     }
 
-    private enum Type {
-        STRING, LONG, BOOLEAN, DOUBLE, DATE,
+    public enum Type {
+        STRING, LONG, INT, BOOLEAN, DOUBLE, DATE,
         ;
     }