rdb dirty data

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/metric/MetricConstant.java

@@ -45,6 +45,8 @@ public class MetricConstant {
 
     public static final String DT_NUM_RECORDS_OUT = "dtNumRecordsOut";
 
+    public static final String DT_NUM_DIRTY_RECORDS_OUT = "dtNumDirtyRecordsOut";
+
     public static final String DT_NUM_RECORDS_OUT_RATE = "dtNumRecordsOutRate";
 
     public static final String DT_EVENT_DELAY_GAUGE = "dtEventDelay";

core/src/main/java/com/dtstack/flink/sql/sink/MetricOutputFormat.java

@@ -32,10 +32,13 @@ public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
 
      protected  transient Counter outRecords;
 
+     protected  transient Counter outDirtyRecords;
+
      protected transient Meter outRecordsRate;
 
      public void initMetric() {
         outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        outDirtyRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_DIRTY_RECORDS_OUT);
         outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(outRecords, 20));
      }
 

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

@@ -32,6 +32,7 @@
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -66,9 +67,11 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private String insertQuery;
     public int[] typesArray;
 
+    /** 存储用于批量写入的数据 */
+    protected List<Row> rows = new ArrayList();
+
     private Connection dbConn;
     private PreparedStatement upload;
-    private AtomicInteger batchCount = new AtomicInteger(0);
     private transient ScheduledThreadPoolExecutor timerService;
 
 
@@ -107,7 +110,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
                 throw new SQLException("Table " + tableName + " doesn't exist");
             }
 
-            if (batchWaitInterval > 0) {
+            if (batchWaitInterval > 0 && batchNum > 1) {
                 LOG.info("open batch wait interval scheduled, interval is {} ms", batchWaitInterval);
 
                 timerService = new ScheduledThreadPoolExecutor(1);
@@ -134,6 +137,7 @@ private void establishConnection() throws SQLException, ClassNotFoundException {
         } else {
             dbConn = DriverManager.getConnection(dbURL, username, password);
         }
+        dbConn.setAutoCommit(false);
     }
 
     /**
@@ -150,35 +154,53 @@ private void establishConnection() throws SQLException, ClassNotFoundException {
      * @see PreparedStatement
      */
     @Override
-    public void writeRecord(Tuple2 tuple2) throws IOException {
+    public void writeRecord(Tuple2 tuple2)  {
 
         Tuple2<Boolean, Row> tupleTrans = tuple2;
         Boolean retract = tupleTrans.getField(0);
         Row row = tupleTrans.getField(1);
 
-
         if (typesArray != null && typesArray.length > 0 && typesArray.length != row.getArity()) {
             LOG.warn("Column SQL types array doesn't match arity of passed Row! Check the passed array...");
         }
-        try {
-            if (retract) {
-                insertWrite(row);
-                outRecords.inc();
-            } else {
-                //do nothing
-            }
-        } catch (SQLException | IllegalArgumentException e) {
-            throw new IllegalArgumentException("writeRecord() failed", e);
+
+        if (retract) {
+            insertWrite(row);
+            outRecords.inc();
+        } else {
+            //do nothing
         }
     }
 
 
-    private void insertWrite(Row row) throws SQLException {
-        updatePreparedStmt(row, upload);
-        upload.addBatch();
-        batchCount.incrementAndGet();
-        if (batchCount.get() >= batchNum) {
-            submitExecuteBatch();
+    private void insertWrite(Row row) {
+        checkConnectionOpen(dbConn);
+
+        if (batchNum == 1) {
+            writeSingleRecord(row);
+        } else {
+            try {
+                rows.add(row);
+                updatePreparedStmt(row, upload);
+                upload.addBatch();
+            } catch (SQLException e) {
+                LOG.error("", e);
+            }
+
+            if (rows.size() >= batchNum) {
+                submitExecuteBatch();
+            }
+        }
+    }
+
+    private void writeSingleRecord(Row row) {
+        try {
+            updatePreparedStmt(row, upload);
+            upload.execute();
+        } catch (SQLException e) {
+            outDirtyRecords.inc();
+            LOG.error("record insert failed ..", row.toString());
+            LOG.error("", e);
         }
     }
 
@@ -276,9 +298,30 @@ private synchronized void submitExecuteBatch() {
         try {
             LOG.info("submitExecuteBatch start......");
             this.upload.executeBatch();
-            this.batchCount.set(0);
+            dbConn.commit();
         } catch (SQLException e) {
-            LOG.error("", e);
+            try {
+                dbConn.rollback();
+            } catch (SQLException e1) {
+                LOG.error("rollback  data error !", e);
+            }
+            rows.forEach(this::writeSingleRecord);
+        } finally {
+            rows.clear();
+        }
+    }
+
+    private void checkConnectionOpen(Connection dbConn) {
+        try {
+            if (dbConn.isClosed()) {
+                LOG.info("db connection reconnect..");
+                establishConnection();
+                upload = dbConn.prepareStatement(insertQuery);
+            }
+        } catch (SQLException e) {
+            LOG.error("check connection open failed..", e);
+        } catch (ClassNotFoundException e) {
+            LOG.error("load jdbc class error when reconnect db..", e);
         }
     }
 
@@ -302,7 +345,7 @@ public void close() throws IOException {
             LOG.info("Inputformat couldn't be closed - ", se);
         } finally {
             upload = null;
-            batchCount.set(0);
+            rows.clear();
         }
 
         try {