去掉日志打印

realtime/hologram-streaming/src/main/java/com/zhht/irn/consts/Constant.java

@@ -16,5 +16,5 @@ public class Constant {
     // 周期计算的延迟时间 15秒
     public static final int CYCLE_CALCULATE_LATENESS = 15 * 1000;
 
-    public static final int INSERT_BATCH_SIZE = 100;
+    public static final int INSERT_BATCH_SIZE = 1000;
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalDirectionSecondMetricFunction.java

@@ -138,7 +138,6 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
         listState.clear();
         for(Map.Entry<String, Boolean> en : globalRecordStopStatusMap.entrySet()) {
             if(en.getValue() == true) {
-                System.out.println("stay stop status car record id:" + en.getKey());
                 listState.add(en.getKey());
             }
         }
@@ -146,7 +145,7 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
         if(cachedTravelEvent != null) {
             List<TravelEvent> newOne = new ArrayList<>();
             for(TravelEvent t : cachedTravelEvent) {
-                // 删除掉这个周期结束时间之间的数据
+                // 删除掉这个周期结束时间之前的数据
                 if(Long.parseLong(t.getEventTime()) >= Long.parseLong(cycle.getEndDateTime())) {
                     newOne.add(t);
                 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/DirectionEvalJob.java

@@ -4,20 +4,14 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
-import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
 import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
 import com.zhht.irn.sink.DirectionEvalSecondSink;
-import com.zhht.irn.source.CycleMockSource;
-import com.zhht.irn.source.TravelEventMockSource;
 import com.zhht.irn.utils.FlinkUtils;
 import com.zhht.irn.utils.StringUtils;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.KeyedStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
-import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,7 +37,6 @@ public class DirectionEvalJob {
         DataStream<Cycle> cycleStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
-                    System.out.println(record);
                     return JSON.parseObject(record, Cycle.class);
                 });
 
@@ -51,7 +44,6 @@ public class DirectionEvalJob {
         DataStream<TravelEvent> travelEventStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
-                    System.out.println(record);
                     return JSON.parseObject(record, TravelEvent.class);
                 }).filter(event -> {
                     return StringUtils.isNotEmpty(event.getEventLineId());

realtime/hologram-streaming/src/main/resources/kafka.properties

@@ -4,4 +4,4 @@ auto.offset.reset=latest
 group.id=TravelInfoJob
 checkpoint.path=file:///data/flink/state
 checkpoint.interval=60000
-enable.auto.commit=false
\ No newline at end of file
+enable.auto.commit=true
\ No newline at end of file