解决中文乱吗问题

Author: dapeng

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

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
 import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
 
 import java.util.Arrays;
 import java.util.List;
@@ -38,6 +39,12 @@ public class ImpalaDialect implements JDBCDialect {
 
     private static final String IMPALA_PARTITION_KEYWORD = "partition";
 
+    private TypeInformation[] fieldTypes;
+
+    public ImpalaDialect(TypeInformation[] fieldTypes){
+        this.fieldTypes = fieldTypes;
+    }
+
     @Override
     public boolean canHandle(String url) {
         return url.startsWith("jdbc:impala:");
@@ -70,8 +77,13 @@ public String getInsertIntoStatement(String schema, String tableName, String[] f
                 .map(this::quoteIdentifier)
                 .collect(Collectors.joining(", "));
 
-        String placeholders = Arrays.stream(fieldNames)
-                .map(f -> "?")
+        String placeholders = Arrays.stream(fieldTypes)
+                .map(f -> {
+                    if(String.class.getName().equals(f.getTypeClass().getName())){
+                        return "cast( ? as string)";
+                    }
+                    return "?";
+                })
                 .collect(Collectors.joining(", "));
 
         String partitionFieldStr = partitionFieldsList.stream()

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

@@ -41,14 +41,14 @@ public class ImpalaSink extends AbstractRdbSink implements IStreamSinkGener<Abst
     private ImpalaTableInfo impalaTableInfo;
 
     public ImpalaSink() {
-        super(new ImpalaDialect());
+        super(null);
     }
 
     @Override
     public JDBCUpsertOutputFormat getOutputFormat() {
         JDBCOptions jdbcOptions = JDBCOptions.builder()
                 .setDbUrl(getImpalaJdbcUrl())
-                .setDialect(jdbcDialect)
+                .setDialect(new ImpalaDialect(getFieldTypes()))
                 .setUsername(userName)
                 .setPassword(password)
                 .setTableName(tableName)

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

@@ -54,6 +54,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         impalaTableInfo.setBufferSize(MathUtil.getString(props.get(ImpalaTableInfo.BUFFER_SIZE_KEY.toLowerCase())));
         impalaTableInfo.setFlushIntervalMs(MathUtil.getString(props.get(ImpalaTableInfo.FLUSH_INTERVALMS_KEY.toLowerCase())));
         impalaTableInfo.setSchema(MathUtil.getString(props.get(ImpalaTableInfo.SCHEMA_KEY.toLowerCase())));
+        impalaTableInfo.setUpdateMode(MathUtil.getString(props.get(ImpalaTableInfo.UPDATE_KEY.toLowerCase())));
 
         Integer authMech = MathUtil.getIntegerVal(props.get(ImpalaTableInfo.AUTHMECH_KEY.toLowerCase()));
         authMech = authMech == null ? 0 : authMech;