impala sink

Author: todd5167

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends MetricOutputFormat {
+public class CassandraOutputFormat extends MetricOutputFormat<Tuple2> {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java

@@ -37,7 +37,7 @@
  *
  * @author xuqianjin
  */
-public class ConsoleOutputFormat extends MetricOutputFormat {
+public class ConsoleOutputFormat extends MetricOutputFormat<Tuple2> {
 
     private static final Logger LOG = LoggerFactory.getLogger(ConsoleOutputFormat.class);
 

db2/db2-sink/src/main/java/com/dtstack/flink/sql/sink/db/DbDialect.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.sink.db;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+
+import java.util.Optional;
+
+/**
+ * Date: 2020/1/19
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class DbDialect implements JDBCDialect {
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:db2:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("com.ibm.db2.jcc.DB2Driver");
+    }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return identifier;
+    }
+
+}

db2/db2-sink/src/main/java/com/dtstack/flink/sql/sink/db/DbSink.java

@@ -1,6 +1,8 @@
 package com.dtstack.flink.sql.sink.db;
 
+import com.dtstack.flink.sql.sink.rdb.JDBCOptions;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
+import com.dtstack.flink.sql.sink.rdb.format.JDBCUpsertOutputFormat;
 
 import java.util.List;
 import java.util.Map;
@@ -10,42 +12,23 @@ public class DbSink extends RdbSink {
     private static final String DB2_DRIVER = "com.ibm.db2.jcc.DB2Driver";
 
     public DbSink() {
+        super(new DbDialect());
     }
-
-    @Override
-    public void buildSql(String schema, String tableName, List<String> fields) {
-        buildInsertSql(tableName, fields);
-    }
-
-    private void buildInsertSql(String tableName, List<String> fields) {
-        String sqlTmp = "insert into " +  tableName + " (${fields}) values (${placeholder})";
-        String fieldsStr = "";
-        String placeholder = "";
-
-        for (String fieldName : fields) {
-            fieldsStr += "," + fieldName;
-            placeholder += ",?";
-        }
-
-        fieldsStr = fieldsStr.replaceFirst(",", "");
-        placeholder = placeholder.replaceFirst(",", "");
-
-        sqlTmp = sqlTmp.replace("${fields}", fieldsStr).replace("${placeholder}", placeholder);
-        this.sql = sqlTmp;
-    }
-
-    @Override
-    public String buildUpdateSql(String schema, String tableName, List<String> fieldNames, Map<String, List<String>> realIndexes, List<String> fullField) {
-        return null;
-    }
-
-    @Override
-    public String getDriverName() {
-        return DB2_DRIVER;
-    }
-
     @Override
-    public RetractJDBCOutputFormat getOutputFormat() {
-        return new RetractJDBCOutputFormat();
+    public JDBCUpsertOutputFormat getOutputFormat() {
+        JDBCOptions jdbcOptions = JDBCOptions.builder()
+                .setDBUrl(dbURL).setDialect(jdbcDialect)
+                .setUsername(userName).setPassword(password)
+                .setTableName(tableName).build();
+
+        return JDBCUpsertOutputFormat.builder()
+                .setOptions(jdbcOptions)
+                .setFieldNames(fieldNames)
+                .setFlushMaxSize(batchNum)
+                .setFlushIntervalMills(batchWaitInterval)
+                .setFieldTypes(sqlTypes)
+                .setKeyFields(primaryKeys)
+                .setAllReplace(allReplace)
+                .setUpdateMode(updateMode).build();
     }
 }

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

@@ -45,7 +45,7 @@
  * author: [email protected]
  * date: 2017-6-29
  */
-public class HbaseOutputFormat extends MetricOutputFormat {
+public class HbaseOutputFormat extends MetricOutputFormat<Tuple2> {
 
     private static final Logger LOG = LoggerFactory.getLogger(HbaseOutputFormat.class);
 

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

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.sink.impala;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Date: 2020/1/17
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class ImpalaDialect implements JDBCDialect {
+    private static final long serialVersionUID = 1L;
+
+    private static final String IMPALA_PARTITION_KEYWORD = " partition";
+
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:impala:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("com.cloudera.impala.jdbc41.Driver");
+    }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return identifier;
+    }
+
+    @Override
+    public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
+        throw new RuntimeException("impala does not support update sql, please remove primary key or use append mode");
+    }
+
+    @Override
+    public String getInsertIntoStatement(String schema, String tableName, String[] fieldNames, String[] partitionFields) {
+
+        String schemaInfo = StringUtils.isEmpty(schema) ? "" : quoteIdentifier(schema) + ".";
+
+        List<String> partitionFieldsList = Arrays.asList(partitionFields);
+
+        String columns = Arrays.stream(fieldNames)
+                .filter(f -> !partitionFieldsList.contains(f))
+                .map(this::quoteIdentifier)
+                .collect(Collectors.joining(", "));
+
+        String placeholders = Arrays.stream(fieldNames)
+                .map(f -> "?")
+                .collect(Collectors.joining(", "));
+
+        String partitionFieldStr = Arrays.stream(partitionFields)
+                .map(field -> field.replaceAll("\"", "'"))
+                .collect(Collectors.joining(", "));
+
+        String partitionStatement = StringUtils.isEmpty(partitionFieldStr) ? "" : IMPALA_PARTITION_KEYWORD + "(" + partitionFieldStr + ")";
+
+        return "INSERT INTO " + schemaInfo + quoteIdentifier(tableName) +
+                "(" + columns + ")" + partitionStatement + " VALUES (" + placeholders + ")";
+    }
+}