添加日志

src/main/java/com/dsk/flink/dsc/sync/SyncCustomerDataSource.java

@@ -46,7 +46,7 @@ public class SyncCustomerDataSource {
     public static void main(String[] args) throws Exception {
 
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        env.setParallelism(1);
+        //env.setParallelism(1);
         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(100, 60000));
         env.enableCheckpointing(120000);
         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
@@ -91,7 +91,6 @@ public class SyncCustomerDataSource {
                 .name("dsc-source")
                 .uid("dsc-source");
 
-        tsGroupStream.print("source==>");
         OutputTag<Tuple6<String,String,String,String,String,Long>> logSlideTag = new OutputTag<Tuple6<String,String,String,String,String,Long>>("log_slide") {};
 
         SingleOutputStreamOperator<Tuple3<String, String, Long>> slide = tsGroupStream
@@ -99,8 +98,6 @@ public class SyncCustomerDataSource {
                 .name("dsc-sql")
                 .uid("dsc-sql");
 
-        slide.print("dsc-sql ==>");
-
         SingleOutputStreamOperator<String> groupWindowSqlResultStream = slide
                 .keyBy(value -> value.f1)
                 .window(TumblingProcessingTimeWindows.of(Time.milliseconds(100)))
@@ -123,13 +120,12 @@ public class SyncCustomerDataSource {
                 .name("dsc-max")
                 .uid("dsc-max");
 
-        groupWindowSqlResultStream.print("dsc-max==>");
         groupWindowSqlResultStream.addSink(new MysqlDataTransferSinkBatch(envProps))
                 .name("dsc-sink")
                 .uid("dsc-sink");
 
         DataStream<Tuple6<String,String,String,String,String,Long>> sideOutput = slide.getSideOutput(logSlideTag);
-        sideOutput.print("log==>");
+
         sideOutput.addSink(JdbcSink.sink(
                 "INSERT INTO dsc_cdc_log (`table`,op_type,pk_columns,pk_values,data_json,cdc_ts) values (?,?,?,?,?,?)",
                 (ps,t) -> {
@@ -151,8 +147,8 @@ public class SyncCustomerDataSource {
                         .withUsername(envProps.getDb_username())
                         .withPassword(envProps.getDb_password())
                         .build()
-        )).uid("deleteProject")
-                .name("deleteProject");
+        )).uid("dsc-log")
+          .name("dsc-log");
 
         env.execute("dsc-client");
     }