add scheduled

Author: todd5167

pom.xml

@@ -14,15 +14,15 @@
         <module>kafka10</module>
         <module>kafka11</module>
         <module>mysql</module>
-        <module>hbase</module>
-        <module>elasticsearch5</module>
-        <module>mongo</module>
-        <module>redis5</module>
+        <!--<module>hbase</module>-->
+        <!--<module>elasticsearch5</module>-->
+        <!--<module>mongo</module>-->
+        <!--<module>redis5</module>-->
         <module>launcher</module>
         <module>rdb</module>
         <module>sqlserver</module>
         <module>oracle</module>
-        <module>cassandra</module>
+        <!--<module>cassandra</module>-->
     </modules>
 
 

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;
 
     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 > 0) {
+            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,32 @@
 
 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.ScheduledFuture;
+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,22 +59,25 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private String tableName;
     private String dbType;
     private RdbSink dbSink;
-    private int batchInterval = 5000;
+    private long batchWaitInterval;
+    private int batchNum;
     private String insertQuery;
     public int[] typesArray;
 
     private Connection dbConn;
     private PreparedStatement upload;
 
-    private int batchCount = 0;
+    private AtomicInteger batchCount = new AtomicInteger(0);
 
     //index field
     private Map<String, List<String>> realIndexes = Maps.newHashMap();
     //full field
     private List<String> fullField = Lists.newArrayList();
 
-    public RetractJDBCOutputFormat() {
+    private transient ScheduledThreadPoolExecutor timerService;
 
+    public RetractJDBCOutputFormat() {
+        this.timerService = new ScheduledThreadPoolExecutor(1);
     }
 
     @Override
@@ -151,13 +160,23 @@ public void writeRecord(Tuple2 tuple2) throws IOException {
 
 
     private void insertWrite(Row row) throws SQLException {
-
         updatePreparedStmt(row, upload);
         upload.addBatch();
-        batchCount++;
-        if (batchCount >= batchInterval) {
+        ScheduledFuture<?> scheduledFuture = null;
+
+        if (batchWaitInterval > 0 && batchCount.get() == 0) {
+            scheduledFuture = registerTimer(batchWaitInterval, this);
+        }
+
+        batchCount.incrementAndGet();
+
+        if (batchCount.get() >= batchNum) {
             upload.executeBatch();
-            batchCount = 0;
+            batchCount.set(0);
+
+            if (scheduledFuture != null) {
+                scheduledFuture.cancel(true);
+            }
         }
     }
 
@@ -250,6 +269,28 @@ private void updatePreparedStmt(Row row, PreparedStatement pstmt) throws SQLExce
         }
     }
 
+    public ScheduledFuture<?> registerTimer(long delay, RetractJDBCOutputFormat outputFormat) {
+        return timerService.schedule(new DelayExecuteBatch(outputFormat), delay, TimeUnit.MILLISECONDS);
+    }
+
+    private final static class DelayExecuteBatch implements Runnable {
+        RetractJDBCOutputFormat outputFormat;
+
+        private DelayExecuteBatch(RetractJDBCOutputFormat outputFormat) {
+            this.outputFormat = outputFormat;
+        }
+
+        @Override
+        public void run() {
+            try {
+                outputFormat.upload.executeBatch();
+                outputFormat.batchCount.set(0);
+            } catch (SQLException e) {
+                LOG.error("delay batch insert error...", e);
+            }
+        }
+    }
+
     /**
      * Executes prepared statement and closes all resources of this instance.
      *
@@ -266,7 +307,7 @@ public void close() throws IOException {
             LOG.info("Inputformat couldn't be closed - " + se.getMessage());
         } finally {
             upload = null;
-            batchCount = 0;
+            batchCount.set(0);
         }
 
         try {
@@ -332,8 +373,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) {
@@ -364,6 +405,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,14 @@ 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");