Merge branch 'v1.5.0_dev_sink_batch_num' into 'v1.5.0_dev'

rdb sink batch and interval

rdb写入时支持：batchSize 和 batchWaitInterval 参数设置。

See merge request !42

Author: zoudaokoulife

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java

@@ -61,7 +61,9 @@ public abstract class RdbSink implements RetractStreamTableSink<Row>, Serializab
 
     protected String dbType;
 
-    protected int batchInterval = 1;
+    protected int batchNum = 1;
+
+    protected long batchWaitInterval = 10000;
 
     protected int[] sqlTypes;
 
@@ -89,7 +91,8 @@ public RichSinkFunction createJdbcSinkFunc() {
         outputFormat.setUsername(userName);
         outputFormat.setPassword(password);
         outputFormat.setInsertQuery(sql);
-        outputFormat.setBatchInterval(batchInterval);
+        outputFormat.setBatchNum(batchNum);
+        outputFormat.setBatchWaitInterval(batchWaitInterval);
         outputFormat.setTypesArray(sqlTypes);
         outputFormat.setTableName(tableName);
         outputFormat.setDbType(dbType);
@@ -112,7 +115,12 @@ public RdbSink genStreamSink(TargetTableInfo targetTableInfo) {
 
         Integer tmpSqlBatchSize = rdbTableInfo.getBatchSize();
         if (tmpSqlBatchSize != null) {
-            setBatchInterval(tmpSqlBatchSize);
+            setBatchNum(tmpSqlBatchSize);
+        }
+
+        Long batchWaitInterval = rdbTableInfo.getBatchWaitInterval();
+        if (batchWaitInterval != null) {
+            setBatchWaitInterval(batchWaitInterval);
         }
 
         Integer tmpSinkParallelism = rdbTableInfo.getParallelism();
@@ -198,13 +206,13 @@ public TableSink<Tuple2<Boolean, Row>> configure(String[] fieldNames, TypeInform
         return this;
     }
 
-    /**
-     * Set the default frequency submit updated every submission
-     *
-     * @param batchInterval
-     */
-    public void setBatchInterval(int batchInterval) {
-        this.batchInterval = batchInterval;
+
+    public void setBatchNum(int batchNum) {
+        this.batchNum = batchNum;
+    }
+
+    public void setBatchWaitInterval(long batchWaitInterval) {
+        this.batchWaitInterval = batchWaitInterval;
     }
 
     @Override

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

@@ -20,26 +20,31 @@
 
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import java.sql.*;
-import java.util.*;
 import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import com.dtstack.flink.sql.sink.MetricOutputFormat;
 
 /**
  * OutputFormat to write tuples into a database.
  * The OutputFormat has to be configured using the supplied OutputFormatBuilder.
  *
- * @see Tuple
- * @see DriverManager
  */
 public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private static final long serialVersionUID = 1L;
@@ -53,23 +58,25 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private String tableName;
     private String dbType;
     private RdbSink dbSink;
-    private int batchInterval = 5000;
+    // trigger preparedStatement execute batch interval
+    private long batchWaitInterval = 10000l;
+    // PreparedStatement execute batch num
+    private int batchNum = 1;
     private String insertQuery;
     public int[] typesArray;
 
     private Connection dbConn;
     private PreparedStatement upload;
+    private AtomicInteger batchCount = new AtomicInteger(0);
+    private transient ScheduledThreadPoolExecutor timerService;
 
-    private int batchCount = 0;
-    private static int rowLenth = 1000;
 
     //index field
     private Map<String, List<String>> realIndexes = Maps.newHashMap();
     //full field
     private List<String> fullField = Lists.newArrayList();
 
     public RetractJDBCOutputFormat() {
-
     }
 
     @Override
@@ -86,8 +93,20 @@ public void configure(Configuration parameters) {
     @Override
     public void open(int taskNumber, int numTasks) throws IOException {
         try {
+            LOG.info("PreparedStatement execute batch num is {}", batchNum);
             establishConnection();
             initMetric();
+
+            if (batchWaitInterval > 0) {
+                LOG.info("open batch wait interval scheduled, interval is {} ms", batchWaitInterval);
+
+                timerService = new ScheduledThreadPoolExecutor(1);
+                timerService.scheduleAtFixedRate(() -> {
+                    submitExecuteBatch();
+                }, 0, batchWaitInterval, TimeUnit.MILLISECONDS);
+
+            }
+
             if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
                 if (isReplaceInsertQuery()) {
                     insertQuery = dbSink.buildUpdateSql(tableName, Arrays.asList(dbSink.getFieldNames()), realIndexes, fullField);
@@ -141,9 +160,6 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
         try {
             if (retract) {
                 insertWrite(row);
-                if (outRecords.getCount()%rowLenth == 0){
-                    LOG.info(row.toString());
-                }
                 outRecords.inc();
             } else {
                 //do nothing
@@ -155,13 +171,11 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
 
 
     private void insertWrite(Row row) throws SQLException {
-
         updatePreparedStmt(row, upload);
         upload.addBatch();
-        batchCount++;
-        if (batchCount >= batchInterval) {
-            upload.executeBatch();
-            batchCount = 0;
+        batchCount.incrementAndGet();
+        if (batchCount.get() >= batchNum) {
+            submitExecuteBatch();
         }
     }
 
@@ -254,6 +268,16 @@ private void updatePreparedStmt(Row row, PreparedStatement pstmt) throws SQLExce
         }
     }
 
+
+    private synchronized void submitExecuteBatch() {
+        try {
+            this.upload.executeBatch();
+            this.batchCount.set(0);
+        } catch (SQLException e) {
+            LOG.error("", e);
+        }
+    }
+
     /**
      * Executes prepared statement and closes all resources of this instance.
      *
@@ -266,19 +290,23 @@ public void close() throws IOException {
                 upload.executeBatch();
                 upload.close();
             }
+            if (null != timerService) {
+                timerService.shutdown();
+                LOG.info("batch wait interval scheduled service  closed ");
+            }
         } catch (SQLException se) {
-            LOG.info("Inputformat couldn't be closed - " + se.getMessage());
+            LOG.info("Inputformat couldn't be closed - ", se);
         } finally {
             upload = null;
-            batchCount = 0;
+            batchCount.set(0);
         }
 
         try {
             if (dbConn != null) {
                 dbConn.close();
             }
         } catch (SQLException se) {
-            LOG.info("Inputformat couldn't be closed - " + se.getMessage());
+            LOG.info("Inputformat couldn't be closed - ", se);
         } finally {
             dbConn = null;
         }
@@ -336,8 +364,8 @@ public void setDbSink(RdbSink dbSink) {
         this.dbSink = dbSink;
     }
 
-    public void setBatchInterval(int batchInterval) {
-        this.batchInterval = batchInterval;
+    public void setBatchNum(int batchNum) {
+        this.batchNum = batchNum;
     }
 
     public void setInsertQuery(String insertQuery) {
@@ -368,6 +396,11 @@ public Map<String, List<String>> getRealIndexes() {
         return realIndexes;
     }
 
+
+    public void setBatchWaitInterval(long batchWaitInterval) {
+        this.batchWaitInterval = batchWaitInterval;
+    }
+
     public List<String> getFullField() {
         return fullField;
     }

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbSinkParser.java

@@ -43,6 +43,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         rdbTableInfo.setUserName(MathUtil.getString(props.get(RdbTableInfo.USER_NAME_KEY.toLowerCase())));
         rdbTableInfo.setPassword(MathUtil.getString(props.get(RdbTableInfo.PASSWORD_KEY.toLowerCase())));
         rdbTableInfo.setBatchSize(MathUtil.getIntegerVal(props.get(RdbTableInfo.BATCH_SIZE_KEY.toLowerCase())));
+        rdbTableInfo.setBatchWaitInterval(MathUtil.getLongVal(props.get(RdbTableInfo.BATCH_WAIT_INTERVAL_KEY.toLowerCase())));
         rdbTableInfo.setBufferSize(MathUtil.getString(props.get(RdbTableInfo.BUFFER_SIZE_KEY.toLowerCase())));
         rdbTableInfo.setFlushIntervalMs(MathUtil.getString(props.get(RdbTableInfo.FLUSH_INTERVALMS_KEY.toLowerCase())));
 

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbTableInfo.java

@@ -39,6 +39,8 @@ public class RdbTableInfo extends TargetTableInfo {
 
     public static final String BATCH_SIZE_KEY = "batchSize";
 
+    public static final String BATCH_WAIT_INTERVAL_KEY = "batchWaitInterval";
+
     public static final String BUFFER_SIZE_KEY = "bufferSize";
 
     public static final String FLUSH_INTERVALMS_KEY = "flushIntervalMs";
@@ -53,6 +55,8 @@ public class RdbTableInfo extends TargetTableInfo {
 
     private Integer batchSize;
 
+    private Long batchWaitInterval;
+
     private String bufferSize;
 
     private String flushIntervalMs;
@@ -113,6 +117,15 @@ public void setFlushIntervalMs(String flushIntervalMs) {
         this.flushIntervalMs = flushIntervalMs;
     }
 
+
+    public Long getBatchWaitInterval() {
+        return batchWaitInterval;
+    }
+
+    public void setBatchWaitInterval(Long batchWaitInterval) {
+        this.batchWaitInterval = batchWaitInterval;
+    }
+
     @Override
     public boolean check() {
         Preconditions.checkNotNull(url, "rdb field of URL is required");