hotfix-#1795][hdfs] fix sink hdfs bug

DTStack · Aug 22, 2023 · 8f347c9 · 8f347c9
1 parent 74c1591
commit 8f347c9
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 24 deletions.
diff --git a/.idea/vcs.xml b/.idea/vcs.xml
diff --git a/...hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/converter/HdfsOrcSyncConverter.java b/...hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/converter/HdfsOrcSyncConverter.java
@@ -60,7 +60,7 @@ public class HdfsOrcSyncConverter extends AbstractRowConverter<RowData, RowData,
 
     private static final long serialVersionUID = 4254984437380862131L;
 
-    private List<String> ColumnNameList;
+    private List<String> columnNameList;
     private transient Map<String, ColumnTypeUtil.DecimalInfo> decimalColInfo;
 
     public HdfsOrcSyncConverter(List<FieldConfig> fieldConfigList, HdfsConfig hdfsConfig) {
@@ -107,7 +107,7 @@ public RowData toInternal(RowData input) throws Exception {
     @Override
     @SuppressWarnings("unchecked")
     public Object[] toExternal(RowData rowData, Object[] data) throws Exception {
-        for (int index = 0; index < fieldTypes.length; index++) {
+        for (int index = 0; index < columnNameList.size(); index++) {
             toExternalConverters.get(index).serialize(rowData, index, data);
         }
         return data;
@@ -195,7 +195,7 @@ protected ISerializationConverter<Object[]> createExternalConverter(String type)
             case "DECIMAL":
                 return (rowData, index, data) -> {
                     ColumnTypeUtil.DecimalInfo decimalInfo =
-                            decimalColInfo.get(ColumnNameList.get(index));
+                            decimalColInfo.get(columnNameList.get(index));
                     HiveDecimal hiveDecimal =
                             HiveDecimal.create(new BigDecimal(rowData.getString(index).toString()));
                     hiveDecimal =
@@ -245,7 +245,7 @@ protected ISerializationConverter<Object[]> createExternalConverter(String type)
     }
 
     public void setColumnNameList(List<String> columnNameList) {
-        this.ColumnNameList = columnNameList;
+        this.columnNameList = columnNameList;
     }
 
     public void setDecimalColInfo(Map<String, ColumnTypeUtil.DecimalInfo> decimalColInfo) {

diff --git a/.../src/main/java/com/dtstack/chunjun/connector/hdfs/converter/HdfsParquetSyncConverter.java b/.../src/main/java/com/dtstack/chunjun/connector/hdfs/converter/HdfsParquetSyncConverter.java
@@ -111,7 +111,7 @@ public RowData toInternal(RowData input) throws Exception {
     @Override
     @SuppressWarnings("unchecked")
     public Group toExternal(RowData rowData, Group group) throws Exception {
-        for (int index = 0; index < fieldTypes.length; index++) {
+        for (int index = 0; index < columnNameList.size(); index++) {
             toExternalConverters.get(index).serialize(rowData, index, group);
         }
         return group;

diff --git a/...dfs/src/main/java/com/dtstack/chunjun/connector/hdfs/converter/HdfsTextSyncConverter.java b/...dfs/src/main/java/com/dtstack/chunjun/connector/hdfs/converter/HdfsTextSyncConverter.java
@@ -51,15 +51,22 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
+import java.util.stream.Collectors;
 
 @Slf4j
 public class HdfsTextSyncConverter
         extends AbstractRowConverter<RowData, RowData, String[], String> {
 
     private static final long serialVersionUID = 1191849197062775272L;
 
+    private List<String> columnNameList;
+
     public HdfsTextSyncConverter(List<FieldConfig> fieldConfigList, HdfsConfig hdfsConfig) {
         super(fieldConfigList.size(), hdfsConfig);
+        columnNameList =
+                hdfsConfig.getColumn().stream()
+                        .map(FieldConfig::getName)
+                        .collect(Collectors.toList());
         for (FieldConfig fieldConfig : fieldConfigList) {
             String type = fieldConfig.getType().getType();
             int left = type.indexOf(ConstantValue.LEFT_PARENTHESIS_SYMBOL);
@@ -103,7 +110,7 @@ public RowData toInternal(RowData input) throws Exception {
     @Override
     @SuppressWarnings("unchecked")
     public String[] toExternal(RowData rowData, String[] data) throws Exception {
-        for (int index = 0; index < fieldTypes.length; index++) {
+        for (int index = 0; index < columnNameList.size(); index++) {
             toExternalConverters.get(index).serialize(rowData, index, data);
         }
         return data;