2022-12-26 更改日志输出方式

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

@@ -157,7 +157,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
             //判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
             if (alreadyHandleCycleSignalDataState.contains(value.getCycleOrder())) {
-                System.out.println("周期数据重复----当前路口id是" + value.getCrossCode() + "当前周期已经计算完成了" + value.getCycleOrder());
+                logger.info("周期数据重复----当前路口id是" + value.getCrossCode() + "当前周期已经计算完成了" + value.getCycleOrder());
                 return;
             }
             cycleSectionState.add(value.getBeginDateTime());
@@ -369,7 +369,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                     }
 
                     else if(t.getInCrossTime()<beginDateTime && beginDateTime - t.getInCrossTime() > 1000*60*60 ) {
-                        System.out.println("移除当前旅行信息-超过30分钟都没有触发计算"+t);
+                        logger.info("移除当前旅行信息-超过30分钟都没有触发计算"+t);
                     }
                     else {
                         //当前周期没有用的的数据

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

@@ -77,6 +77,8 @@ public class TravelSituationAnalysisJob {
             //旅行信息数据
          //   FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
             FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
+            travelInfo.setStartFromTimestamp(1671717600000L);
+
             DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
             //周期信号