Merge remote-tracking branch 'origin/v1.8.0_dev' into 1.8.0_release

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/side/AsyncReqRow.java
#	core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

Author: todd5167

README.md

@@ -26,6 +26,18 @@
 * Java: JDK8及以上
 * Flink集群: 1.4,1.5,1.8（单机模式不需要安装Flink集群）
 * 操作系统：理论上不限
+* kerberos环境需要在flink-conf.yaml配置security.kerberos.login.keytab以及security.kerberos.login.principal参数，配置案例:
+```
+## hadoop配置文件路径
+fs.hdfs.hadoopconf: /Users/maqi/tmp/hadoopconf/hadoop_250
+security.kerberos.login.use-ticket-cache: true
+security.kerberos.login.keytab: /Users/maqi/tmp/hadoopconf/hadoop_250/maqi.keytab
+security.kerberos.login.principal: [email protected]
+security.kerberos.login.contexts: Client,KafkaClient
+zookeeper.sasl.service-name: zookeeper
+zookeeper.sasl.login-context-name: Client
+
+```
 
 ### 1.3 打包
 
@@ -34,9 +46,16 @@
 ```
 mvn clean package -Dmaven.test.skip
 
-打包结束后，项目根目录下会产生plugins目录，plugins目录下存放编译好的数据同步插件包,在lib目下存放job提交的包
 ```
 
+打包完成后的包结构：
+
+> * dt-center-flinkStreamSQL  
+>  >  * bin: 任务启动脚本
+>  >  * lib: launcher包存储路径，是任务提交的入口
+>  >  * plugins:  插件包存储路径
+>  >  * ........   :  core及插件代码
+
 ### 1.4 启动
 
 #### 1.4.1 启动命令

cassandra/cassandra-side/cassandra-all-side/pom.xml

@@ -36,6 +36,7 @@
                             <goal>shade</goal>
                         </goals>
                         <configuration>
+                            <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <excludes>
                                     <exclude>org.slf4j</exclude>

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -38,6 +38,7 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Collector;
@@ -129,14 +130,14 @@ protected void reloadCache() {
 
 
     @Override
-    public void flatMap(Row value, Collector<Row> out) throws Exception {
+    public void flatMap(CRow input, Collector<CRow> out) throws Exception {
         List<Object> inputParams = Lists.newArrayList();
         for (Integer conValIndex : sideInfo.getEqualValIndex()) {
-            Object equalObj = value.getField(conValIndex);
+            Object equalObj = input.row().getField(conValIndex);
             if (equalObj == null) {
                 if(sideInfo.getJoinType() == JoinType.LEFT){
-                    Row data = fillData(value, null);
-                    out.collect(data);
+                    Row data = fillData(input.row(), null);
+                    out.collect(new CRow(data, input.change()));
                 }
                 return;
             }
@@ -148,8 +149,8 @@ public void flatMap(Row value, Collector<Row> out) throws Exception {
         List<Map<String, Object>> cacheList = cacheRef.get().get(key);
         if (CollectionUtils.isEmpty(cacheList)) {
             if (sideInfo.getJoinType() == JoinType.LEFT) {
-                Row row = fillData(value, null);
-                out.collect(row);
+                Row row = fillData(input.row(), null);
+                out.collect(new CRow(row, input.change()));
             } else {
                 return;
             }
@@ -158,7 +159,7 @@ public void flatMap(Row value, Collector<Row> out) throws Exception {
         }
 
         for (Map<String, Object> one : cacheList) {
-            out.collect(fillData(value, one));
+            out.collect(new CRow(fillData(input.row(), one), input.change()));
         }
 
     }
@@ -268,7 +269,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
                         LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
                         Thread.sleep(5 * 1000);
                     } catch (InterruptedException e1) {
-                        e1.printStackTrace();
+                        LOG.error("", e1);
                     }
                 }
 

cassandra/cassandra-side/cassandra-async-side/pom.xml

@@ -52,6 +52,7 @@
                             <goal>shade</goal>
                         </goals>
                         <configuration>
+                            <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <excludes>
                                     <exclude>org.slf4j</exclude>

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java

@@ -47,6 +47,7 @@
 import org.apache.flink.configuration.Configuration;
 import com.google.common.collect.Lists;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
@@ -160,17 +161,17 @@ private void connCassandraDB(CassandraSideTableInfo tableInfo) {
     }
 
     @Override
-    public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
-
+    public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
+        CRow inputCopy = new CRow(input.row(), input.change());
         JsonArray inputParams = new JsonArray();
         StringBuffer stringBuffer = new StringBuffer();
         String sqlWhere = " where ";
 
         for (int i = 0; i < sideInfo.getEqualFieldList().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
-            Object equalObj = input.getField(conValIndex);
+            Object equalObj = inputCopy.row().getField(conValIndex);
             if (equalObj == null) {
-                dealMissKey(input, resultFuture);
+                dealMissKey(inputCopy, resultFuture);
                 return;
             }
             inputParams.add(equalObj);
@@ -194,13 +195,13 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             if (val != null) {
 
                 if (ECacheContentType.MissVal == val.getType()) {
-                    dealMissKey(input, resultFuture);
+                    dealMissKey(inputCopy, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
-                    List<Row> rowList = Lists.newArrayList();
+                    List<CRow> rowList = Lists.newArrayList();
                     for (Object jsonArray : (List) val.getContent()) {
-                        Row row = fillData(input, jsonArray);
-                        rowList.add(row);
+                        Row row = fillData(inputCopy.row(), jsonArray);
+                        rowList.add(new CRow(row, inputCopy.change()));
                     }
                     resultFuture.complete(rowList);
                 } else {
@@ -238,20 +239,20 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
                 cluster.closeAsync();
                 if (rows.size() > 0) {
                     List<com.datastax.driver.core.Row> cacheContent = Lists.newArrayList();
-                    List<Row> rowList = Lists.newArrayList();
+                    List<CRow> rowList = Lists.newArrayList();
                     for (com.datastax.driver.core.Row line : rows) {
-                        Row row = fillData(input, line);
+                        Row row = fillData(inputCopy.row(), line);
                         if (openCache()) {
                             cacheContent.add(line);
                         }
-                        rowList.add(row);
+                        rowList.add(new CRow(row,inputCopy.change()));
                     }
                     resultFuture.complete(rowList);
                     if (openCache()) {
                         putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
                     }
                 } else {
-                    dealMissKey(input, resultFuture);
+                    dealMissKey(inputCopy, resultFuture);
                     if (openCache()) {
                         putCache(key, CacheMissVal.getMissKeyObj());
                     }

cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java

@@ -68,9 +68,8 @@ public class CassandraSideParser extends AbsSideTableParser {
 
     public static final String POOL_TIMEOUT_MILLIS_KEY = "poolTimeoutMillis";
 
-    static {
-        keyPatternMap.put(SIDE_SIGN_KEY, SIDE_TABLE_SIGN);
-        keyHandlerMap.put(SIDE_SIGN_KEY, CassandraSideParser::dealSideSign);
+    public CassandraSideParser() {
+        addParserHandler(SIDE_SIGN_KEY, SIDE_TABLE_SIGN, this::dealSideSign);
     }
 
     @Override
@@ -97,7 +96,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         return cassandraSideTableInfo;
     }
 
-    private static void dealSideSign(Matcher matcher, TableInfo tableInfo) {
+    private void dealSideSign(Matcher matcher, TableInfo tableInfo) {
     }
 
     public Class dbTypeConvertToJavaType(String fieldType) {

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

@@ -48,7 +48,7 @@
 import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
-import com.dtstack.flink.sql.sink.MetricOutputFormat;
+import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends MetricOutputFormat {
+public class CassandraOutputFormat extends DtRichOutputFormat<Tuple2> {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);
@@ -193,7 +193,6 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
         try {
             if (retract) {
                 insertWrite(row);
-                outRecords.inc();
             } else {
                 //do nothing
             }
@@ -204,14 +203,24 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
 
     private void insertWrite(Row row) {
         try {
+
+            if(outRecords.getCount() % ROW_PRINT_FREQUENCY == 0){
+                LOG.info("Receive data : {}", row);
+            }
+
             String cql = buildSql(row);
             if (cql != null) {
                 ResultSet resultSet = session.execute(cql);
                 resultSet.wasApplied();
+                outRecords.inc();
             }
         } catch (Exception e) {
+            if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
+                LOG.error("record insert failed ..", row.toString().substring(0, 100));
+                LOG.error("", e);
+            }
+
             outDirtyRecords.inc();
-            LOG.error("[upsert] is error:" + e.getMessage());
         }
     }
 

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseDialect.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.clickhouse;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+
+import java.util.Optional;
+
+/**
+ * Date: 2020/1/15
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class ClickhouseDialect implements JDBCDialect {
+
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:clickhouse:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("ru.yandex.clickhouse.ClickHouseDriver");
+    }
+
+    @Override
+    public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
+        throw new RuntimeException("Clickhouse does not support update sql, please remove primary key or use append mode");
+    }
+}

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseSink.java

@@ -21,58 +21,39 @@
 
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
+import com.dtstack.flink.sql.sink.rdb.JDBCOptions;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
-import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
+import com.dtstack.flink.sql.sink.rdb.format.JDBCUpsertOutputFormat;
 
 import java.util.List;
 import java.util.Map;
 
 
 public class ClickhouseSink extends RdbSink implements IStreamSinkGener<RdbSink> {
-
-    private static final String CLICKHOUSE_DRIVER = "ru.yandex.clickhouse.ClickHouseDriver";
-
     public ClickhouseSink() {
+        super(new ClickhouseDialect());
     }
 
     @Override
-    public RetractJDBCOutputFormat getOutputFormat() {
-        return new RetractJDBCOutputFormat();
-    }
-
-    @Override
-    public void buildSql(String scheam, String tableName, List<String> fields) {
-        buildInsertSql(tableName, fields);
-    }
-
-    @Override
-    public String buildUpdateSql(String schema, String tableName, List<String> fieldNames, Map<String, List<String>> realIndexes, List<String> fullField) {
-        return null;
-    }
-
-    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;
-        System.out.println("---insert sql----");
-        System.out.println(sql);
-    }
-
-
-    @Override
-    public String getDriverName() {
-        return CLICKHOUSE_DRIVER;
+    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();
     }