data output frequency

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

@@ -34,8 +34,11 @@ public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
 
      protected transient Meter outRecordsRate;
 
+    protected  transient Counter outDirtyRecords;
+
      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

@@ -49,6 +49,9 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
 
     private static final Logger LOG = LoggerFactory.getLogger(RetractJDBCOutputFormat.class);
 
+    private static int dirtyDataPrintFrequency = 1000;
+    private static int receiveDataPrintFrequency = 1000;
+
     private String username;
     private String password;
     private String drivername;
@@ -164,8 +167,11 @@ public void writeRecord(Tuple2 tuple2) {
         }
 
         if (retract) {
-            insertWrite(row);
             outRecords.inc();
+            if (outRecords.getCount() % receiveDataPrintFrequency == 0) {
+                LOG.info("Receive data : {}", row);
+            }
+            insertWrite(row);
         } else {
             //do nothing
         }
@@ -197,14 +203,15 @@ private void writeSingleRecord(Row row) {
             upload.execute();
             dbConn.commit();
         } catch (SQLException e) {
-            LOG.error("record insert failed ..", row.toString());
-            LOG.error("", e);
+            if (outDirtyRecords.getCount() % dirtyDataPrintFrequency == 0) {
+                LOG.error("record insert failed ..", row.toString());
+                LOG.error("", e);
+            }
         }
     }
 
     private synchronized void submitExecuteBatch() {
         try {
-            LOG.info("submitExecuteBatch start......");
             this.upload.executeBatch();
             dbConn.commit();
         } catch (SQLException e) {