temp save

Author: FlechazoW

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java

@@ -48,7 +48,6 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -84,7 +83,7 @@ public class ImpalaOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolea
     protected transient PreparedStatement statement;
 
     private transient volatile boolean closed = false;
-    private final AtomicInteger batchCount = new AtomicInteger(0);
+    private int batchCount = 0;
 
     protected String keytabPath;
     protected String krb5confPath;
@@ -216,7 +215,8 @@ private void openJdbc() {
     private void flush() throws SQLException {
         if (Objects.nonNull(statement)) {
             statement.executeBatch();
-            batchCount.set(0);
+            batchCount = 0;
+            statement.clearBatch();
         }
     }
 
@@ -271,9 +271,10 @@ public void writeRecord(Tuple2<Boolean, Row> record) throws IOException {
                 setRowToStatement(statement, fieldTypeList, rowValue, null);
             }
 
+            batchCount++;
             statement.addBatch();
 
-            if (batchCount.incrementAndGet() >= batchSize) {
+            if (batchCount > batchSize) {
                 flush();
             }
         } catch (Exception e) {
@@ -296,7 +297,7 @@ public void close() throws IOException {
             this.scheduler.shutdown();
         }
         // 将还未执行的SQL flush
-        if (batchCount.get() > 0) {
+        if (batchCount > 0) {
             try {
                 flush();
             } catch (Exception e) {

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaSink.java

@@ -60,7 +60,7 @@ public class ImpalaSink implements RetractStreamTableSink<Row>, IStreamSinkGener
     protected String principal;
 
     protected int batchSize = 100;
-    protected long batchWaitInterval = 10000;
+    protected long batchWaitInterval = 60 * 1000L;
     protected String tableName;
     protected String registerTabName;
     protected String storeType;

localTest/src/main/java/com/dtstack/flink/sql/localTest/LocalTest.java

@@ -50,7 +50,7 @@ public static void main(String[] args) throws Exception {
         setLogLevel("INFO");
 
         List<String> propertiesList = new ArrayList<>();
-        String sqlPath = "//Users/wtz/dtstack/job/flinkStreamSQL/sql/impalaDemoThree.sql";
+        String sqlPath = "/Users/wtz/dtstack/job/flinkStreamSQL/sql/impalaDemoThree.sql";
         Map<String, Object> conf = new HashMap<>();
         JSONObject properties = new JSONObject();
 

pom.xml

@@ -37,7 +37,7 @@
         <module>oceanbase</module>
         <module>tidb</module>
         <module>kingbase</module>
-<!--        <module>localTest</module>-->
+        <module>localTest</module>
 
     </modules>