2022-12-19

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

@@ -219,8 +219,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             long currentProcessingTime = ctx.timerService().currentProcessingTime();
             // 只在某个时间段注册定时器
             SimpleDateFormat sdf = new SimpleDateFormat("HH");
-            if (sdf.format(currentProcessingTime).equals("11")) {
-                ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 30 * 60 * 1000);
+            if (sdf.format(currentProcessingTime).equals("16")) {
+                ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 60 * 1000);
             }
 
         } catch (Exception e) {
@@ -236,10 +236,11 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
 
         //更新维表数据 只在每天12点的时候更新
-        SimpleDateFormat sdf = new SimpleDateFormat("HH");
-        if(sdf.format(timestamp).equals("12")){
+       /* SimpleDateFormat sdf = new SimpleDateFormat("HH");
+        if(sdf.format(timestamp).equals("16")){
             dim_cnt_cross_lane_position = getLaneDimData(connection);
-        }
+        }*/
+        System.out.println("触发定时器执行了。。。");
         dim_cnt_cross_lane_position = getLaneDimData(connection);
     }
 

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

@@ -66,7 +66,7 @@ public class TravelSituationAnalysisJob {
             // kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
             kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
             //  kafkaProperties.setProperty("group.id", "travel_event");
-            kafkaProperties.setProperty("group.id", "trips_info2");
+            kafkaProperties.setProperty("group.id", "trips_info12");
             kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
             kafkaProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
 
@@ -74,7 +74,7 @@ public class TravelSituationAnalysisJob {
             Properties kafkaProperties2 = new Properties();
             //   kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
             kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
-            kafkaProperties2.setProperty("group.id", "cycle_signal2");
+            kafkaProperties2.setProperty("group.id", "cycle_signal12");
             kafkaProperties2.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
             kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
 
@@ -93,7 +93,7 @@ public class TravelSituationAnalysisJob {
           //  travelInfo.setStartFromEarliest();
             DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
-            //  travelInfoStream.print();
+              travelInfoStream.print();
 
             //周期信号
             // FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
@@ -101,7 +101,7 @@ public class TravelSituationAnalysisJob {
            // cycleSignalData.setStartFromEarliest();
             DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
 
-            // cycleSignalDataStream.print();
+             cycleSignalDataStream.print();
 
             //2、使用connect，把2个流联合处理
             ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
@@ -115,7 +115,7 @@ public class TravelSituationAnalysisJob {
             SingleOutputStreamOperator<List<TravelLineSinkInfo>> sinkDataStream = travelInfoCycleSignalDataConnectedStreams
                     .process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
 
-            sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
+          //  sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
 
             env.execute("TravelSituationAnalysisJob");