[feat] impala支持动态分区及对kudu表不做分区处理

Author: FlechazoW

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaDialect.java

@@ -18,18 +18,23 @@
 
 package com.dtstack.flink.sql.sink.impala;
 
+import com.dtstack.flink.sql.sink.impala.table.ImpalaTableInfo;
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
 import com.google.common.collect.Lists;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 
-import java.util.*;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 /**
  * Date: 2020/1/17
  * Company: www.dtstack.com
+ *
  * @author maqi
  */
 public class ImpalaDialect implements JDBCDialect {
@@ -41,9 +46,14 @@ public class ImpalaDialect implements JDBCDialect {
 
     private List<String> primaryKeys;
 
-    public ImpalaDialect(TypeInformation[] fieldTypes, List<String> primaryKeys){
+    private String storeType;
+
+    public ImpalaDialect(TypeInformation[] fieldTypes,
+                         List<String> primaryKeys,
+                         String storeType) {
         this.fieldTypes = fieldTypes;
         this.primaryKeys = primaryKeys;
+        this.storeType = storeType;
     }
 
     @Override
@@ -65,7 +75,7 @@ public String quoteIdentifier(String identifier) {
     public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
         //跳过primary key字段
         String setClause = Arrays.stream(fieldNames)
-                .filter(f -> CollectionUtils.isNotEmpty(primaryKeys) ? !primaryKeys.contains(f) : true)
+                .filter(f -> !CollectionUtils.isNotEmpty(primaryKeys) || !primaryKeys.contains(f))
                 .map(f -> quoteIdentifier(f) + "=?")
                 .collect(Collectors.joining(", "));
         String conditionClause = Arrays.stream(conditionFields)
@@ -83,14 +93,18 @@ public String getInsertIntoStatement(String schema, String tableName, String[] f
 
         List<String> partitionFieldsList = Objects.isNull(partitionFields) ? Lists.newArrayList() : Arrays.asList(partitionFields);
 
+        if (storeType.equalsIgnoreCase(ImpalaTableInfo.KUDU_TYPE)) {
+            return buildKuduInsertSql(schemaInfo, tableName, fieldNames, fieldTypes);
+        }
+
         String columns = Arrays.stream(fieldNames)
                 .filter(f -> !partitionFieldsList.contains(f))
                 .map(this::quoteIdentifier)
                 .collect(Collectors.joining(", "));
 
         String placeholders = Arrays.stream(fieldTypes)
                 .map(f -> {
-                    if(String.class.getName().equals(f.getTypeClass().getName())){
+                    if (String.class.getName().equals(f.getTypeClass().getName())) {
                         return "cast( ? as string)";
                     }
                     return "?";
@@ -106,4 +120,20 @@ public String getInsertIntoStatement(String schema, String tableName, String[] f
         return "INSERT INTO " + schemaInfo + quoteIdentifier(tableName) +
                 "(" + columns + ")" + partitionStatement + " VALUES (" + placeholders + ")";
     }
+
+    private String buildKuduInsertSql(String schemaInfo, String tableName, String[] fieldNames, TypeInformation[] fieldTypes) {  // kudu表的Insert语句
+        String columns = Arrays.stream(fieldNames)
+                .map(this::quoteIdentifier)
+                .collect(Collectors.joining(", "));
+        String placeholders = Arrays.stream(fieldTypes)
+                .map(f -> {
+                    if (String.class.getName().equals(f.getTypeClass().getName())) {
+                        return "cast( ? as string)";
+                    }
+                    return "?";
+                })
+                .collect(Collectors.joining(", "));
+        return "INSERT INTO " + schemaInfo + quoteIdentifier(tableName) +
+                "(" + columns + ")" + " VALUES (" + placeholders + ")";
+    }
 }

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java

@@ -6,7 +6,6 @@
 import org.apache.hadoop.security.UserGroupInformation;
 
 import java.io.IOException;
-import java.security.PrivilegedAction;
 import java.security.PrivilegedExceptionAction;
 import java.util.List;
 
@@ -24,28 +23,28 @@ public class ImpalaOutputFormat extends JDBCUpsertOutputFormat {
     private Integer authMech;
 
     public ImpalaOutputFormat(
-        JDBCOptions options,
-        String[] fieldNames,
-        String[] keyFields,
-        String[] partitionFields,
-        int[] fieldTypes,
-        int flushMaxSize,
-        long flushIntervalMills,
-        boolean allReplace,
-        String updateMode,
-        Integer authMech,
-        String keytabPath,
-        String krb5confPath,
-        String principal) {
+            JDBCOptions options,
+            String[] fieldNames,
+            String[] keyFields,
+            String[] partitionFields,
+            int[] fieldTypes,
+            int flushMaxSize,
+            long flushIntervalMills,
+            boolean allReplace,
+            String updateMode,
+            Integer authMech,
+            String keytabPath,
+            String krb5confPath,
+            String principal) {
         super(options,
-            fieldNames,
-            keyFields,
-            partitionFields,
-            fieldTypes,
-            flushMaxSize,
-            flushIntervalMills,
-            allReplace,
-            updateMode);
+                fieldNames,
+                keyFields,
+                partitionFields,
+                fieldTypes,
+                flushMaxSize,
+                flushIntervalMills,
+                allReplace,
+                updateMode);
         this.authMech = authMech;
         this.keytabPath = keytabPath;
         this.krb5confPath = krb5confPath;
@@ -71,11 +70,11 @@ public Void run() throws IOException {
             super.open(taskNumber, numTasks);
         }
     }
-    
+
     public static Builder impalaBuilder() {
         return new Builder();
     }
-    
+
     public static class Builder {
         private Integer authMech;
         private String keytabPath;
@@ -178,24 +177,24 @@ public Builder setPrincipal(String principal) {
             this.principal = principal;
             return this;
         }
-        
+
         public ImpalaOutputFormat build() {
             checkNotNull(options, "No options supplied.");
             checkNotNull(fieldNames, "No fieldNames supplied.");
             return new ImpalaOutputFormat(
-                options,
-                fieldNames,
-                keyFields,
-                partitionFields,
-                fieldTypes,
-                flushMaxSize,
-                flushIntervalMills,
-                allReplace,
-                updateMode,
-                authMech,
-                keytabPath,
-                krb5confPath,
-                principal);
+                    options,
+                    fieldNames,
+                    keyFields,
+                    partitionFields,
+                    fieldTypes,
+                    flushMaxSize,
+                    flushIntervalMills,
+                    allReplace,
+                    updateMode,
+                    authMech,
+                    keytabPath,
+                    krb5confPath,
+                    principal);
         }
     }
 }

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaSink.java

@@ -41,9 +41,15 @@ public ImpalaSink() {
 
     @Override
     public ImpalaOutputFormat getOutputFormat() {
+        ImpalaDialect impalaDialect = new ImpalaDialect(
+                getFieldTypes(),
+                primaryKeys,
+                impalaTableInfo.getStoreType()
+        );
+
         JDBCOptions jdbcOptions = JDBCOptions.builder()
                 .setDbUrl(getImpalaJdbcUrl())
-                .setDialect(new ImpalaDialect(getFieldTypes(), primaryKeys))
+                .setDialect(impalaDialect)
                 .setUsername(userName)
                 .setPassword(password)
                 .setTableName(tableName)

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/table/ImpalaSinkParser.java

@@ -59,7 +59,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         Integer authMech = MathUtil.getIntegerVal(props.get(ImpalaTableInfo.AUTHMECH_KEY.toLowerCase()));
         authMech = authMech == null ? 0 : authMech;
         impalaTableInfo.setAuthMech(authMech);
-        List authMechs = Arrays.asList(new Integer[]{0, 1, 2, 3});
+        List<Integer> authMechs = Arrays.asList(0, 1, 2, 3);
 
         if (!authMechs.contains(authMech)) {
             throw new IllegalArgumentException("The value of authMech is illegal, Please select 0, 1, 2, 3");
@@ -79,20 +79,26 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
             impalaTableInfo.setPassword(MathUtil.getString(props.get(ImpalaTableInfo.PASSWORD_KEY.toLowerCase())));
         }
 
+        String storeType = MathUtil.getString(props.get(ImpalaTableInfo.STORE_TYPE_KEY.toLowerCase()));
+        impalaTableInfo.setStoreType(storeType);
+
         String enablePartitionStr = (String) props.get(ImpalaTableInfo.ENABLEPARITION_KEY.toLowerCase());
         boolean enablePartition = MathUtil.getBoolean(enablePartitionStr == null ? "false" : enablePartitionStr);
         impalaTableInfo.setEnablePartition(enablePartition);
-        if (enablePartition) {
+
+        if (!storeType.equalsIgnoreCase(ImpalaTableInfo.KUDU_TYPE) && enablePartition) {
             String partitionFields = MathUtil.getString(props.get(ImpalaTableInfo.PARTITIONFIELDS_KEY.toLowerCase()));
             impalaTableInfo.setPartitionFields(partitionFields);
         }
 
+        impalaTableInfo.setType(CURR_TYPE);
+
         impalaTableInfo.check();
         return impalaTableInfo;
     }
 
     @Override
-    public  Class dbTypeConvertToJavaType(String fieldType) {
+    public Class dbTypeConvertToJavaType(String fieldType) {
         switch (fieldType.toLowerCase()) {
             case "boolean":
                 return Boolean.class;

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/table/ImpalaTableInfo.java

@@ -50,6 +50,10 @@ public class ImpalaTableInfo extends RdbTableInfo {
 
     public static final String PARTITIONFIELDS_KEY = "partitionFields";
 
+    public static final String STORE_TYPE_KEY = "storeType";
+
+    public static final String KUDU_TYPE = "kudu";
+
     private static final String CURR_TYPE = "impala";
 
     private static final String PARTITION_FIELD_SPLIT_REGEX = ",";
@@ -72,6 +76,11 @@ public class ImpalaTableInfo extends RdbTableInfo {
 
     private String[] partitionFields;
 
+    /**
+     * If the type of storage is kudu, the logic of partition is ignore
+     */
+    private String storeType;
+
     public ImpalaTableInfo() {
         setType(CURR_TYPE);
     }
@@ -148,11 +157,20 @@ public void setPartitionFields(String partitionFields) {
         this.partitionFields = StringUtils.split(partitionFields, PARTITION_FIELD_SPLIT_REGEX);
     }
 
+    public void setStoreType(String storeType) {
+        this.storeType = storeType;
+    }
+
+    public String getStoreType() {
+        return storeType;
+    }
+
     @Override
     public boolean check() {
         Preconditions.checkNotNull(this.getUrl(), "impala field of url is required");
         Preconditions.checkNotNull(this.getTableName(), "impala field of tableName is required");
         Preconditions.checkNotNull(this.getAuthMech(), "impala field of authMech is required");
+        Preconditions.checkNotNull(this.getStoreType(), "impala field of storeType is required");
         Integer authMech = getAuthMech();
 
         if (authMech == 1) {
@@ -163,12 +181,12 @@ public boolean check() {
             Preconditions.checkNotNull(this.getKrbServiceName(), "impala field of krbServiceName is required");
         } else if (authMech == 2) {
             Preconditions.checkNotNull(this.getUserName(), "impala field of userName is required");
-        }else if (authMech == 3) {
+        } else if (authMech == 3) {
             Preconditions.checkNotNull(this.getUserName(), "impala field of userName is required");
             Preconditions.checkNotNull(this.getPassword(), "impala field of password is required");
         }
 
-        if (isEnablePartition()) {
+        if (!this.getStoreType().equalsIgnoreCase(KUDU_TYPE) && isEnablePartition()) {
             Preconditions.checkArgument(this.getPartitionFields().length > 0, "impala field of partitionFields is required");
         }