2022-12-22 切换使用旅行信息进行计算行车情况相关指标

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

@@ -41,6 +41,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     private transient DruidPooledConnection connection;
     private static Map<String, Map<String, String>> dim_cnt_cross_lane_position;
 
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
 
     @Override
     public void open(Configuration parameters) throws Exception {
@@ -59,22 +61,9 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
         ListStateDescriptor cycleSectionStateDescriptor = new ListStateDescriptor("cycleSectionState", TypeInformation.of(Long.class));
 
-        StateTtlConfig stateTtlConfig4TravelInfo = StateTtlConfig
-                // 状态有效时间    12小时
-                .newBuilder(Time.minutes(15))
-                // 设置状态的更新类型
-                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
-                // 已过期还未被清理掉的状态数据不返回给用户
-                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
-                // 过期对象的清理策略 全量清理
-                .cleanupFullSnapshot()
-                .build();
-        travelInfoStateDescriptor.enableTimeToLive(stateTtlConfig4TravelInfo);
-
-
         StateTtlConfig stateTtlConfig4CycleData = StateTtlConfig
                 // 状态有效时间    3小时
-                .newBuilder(Time.minutes(3))
+                .newBuilder(Time.hours(3))
                 // 设置状态的更新类型
                 .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                 // 已过期还未被清理掉的状态数据不返回给用户
@@ -103,15 +92,19 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     @Override
     public void close() throws Exception {
         super.close();
+        if(connection!=null){
             connection.close();
         }
 
+    }
+
     //把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
     // 所有旅行信息存储在状态中，当周期数据过来的时候，去状态中挑选属于这个周期内的旅行信息的数据即可
     @Override
     public void processElement1(TravelInfo value, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.Context ctx,
                                 Collector<List<TravelLineSinkInfo>> out) throws Exception {
 
+        System.out.println("路口id"+value.getCrossId() +"车辆id"+value.getCarId() + "进入路口时间："+sdf.format(value.getInCrossTime()));
         String crossId = value.getCrossId();
         Iterator<Map<String, List<TravelInfo>>> iterator = travelInfoState.values().iterator();
         //如果状态为空，则进行初始化
@@ -143,7 +136,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
         if(crossListMap != null && crossListMap.containsKey(value.getInCrossLineId()))  {
             List<TravelInfo> lineTravelInfos = crossListMap.get(value.getInCrossLineId());
             //如果，当前车道的数据，积压超过2000条，则进行清理操作
-            if(lineTravelInfos.size()>2000){
+            if(lineTravelInfos.size()>10000){
                 System.out.println("移除积压的数据。。。。");
                 List<TravelInfo> travelInfoList = orderByInCrossTime(lineTravelInfos);
                 List<TravelInfo> travelInfoListNew = travelInfoList.subList(500, travelInfoList.size());
@@ -166,6 +159,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                                 Collector<List<TravelLineSinkInfo>> out)  {
 
         try {
+            System.out.println("路口id"+value.getCrossCode() +"周期id"+value.getCycleOrder() + "周期时间："+sdf.format(value.getBeginDateTime())+"~"+sdf.format(value.getEndDateTime()));
             //判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
             if (alreadyHandleCycleSignalDataState.contains(value.getCycleOrder())) {
                 System.out.println("周期数据重复----当前路口id是" + value.getCrossCode() + "当前周期已经计算完成了" + value.getCycleOrder());
@@ -234,12 +228,6 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     public void onTimer(long timestamp, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.OnTimerContext ctx, Collector<List<TravelLineSinkInfo>> out) throws Exception {
         super.onTimer(timestamp, ctx, out);
 
-
-        //更新维表数据 只在每天12点的时候更新
-        SimpleDateFormat sdf = new SimpleDateFormat("HH");
-        if(sdf.format(timestamp).equals("12")){
-            dim_cnt_cross_lane_position = getLaneDimData(connection);
-        }
         dim_cnt_cross_lane_position = getLaneDimData(connection);
     }
 
@@ -388,9 +376,12 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                         //当前周期用到的数据
                         thisData.add(t);
                         //如果，进入路口时距小于周期开始时间，且已经超过30min，则清理这个数据
-                    } else if(t.getInCrossTime()<beginDateTime && beginDateTime - t.getInCrossTime() > 1000*60*30 ) {
+                    }
+
+                    else if(t.getInCrossTime()<beginDateTime && beginDateTime - t.getInCrossTime() > 1000*60*60 ) {
                         System.out.println("移除当前旅行信息-超过30分钟都没有触发计算"+t);
-                    } else {
+                    }
+                    else {
                         //当前周期没有用的的数据
                         leftData.add(t);
                     }

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

@@ -44,7 +44,7 @@ public class TravelSituationAnalysisJob {
             // checkpoint配置
             CheckpointConfig checkpointConfig = env.getCheckpointConfig();
             // checkpoint时间间隔3分钟
-            checkpointConfig.setCheckpointInterval(1 * 60 * 1000);
+            checkpointConfig.setCheckpointInterval(3 * 60 * 1000);
 
             // 两次checkpoint中最短时间间隔1分钟
             checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
@@ -54,17 +54,13 @@ public class TravelSituationAnalysisJob {
             env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
             //设置重启策略
-            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, Time.seconds(10)));
+            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.seconds(10)));
 
-/*
-           关于flink checkpoint存储的设置统一在集群配置文件中设置
-            checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
-            env.setStateBackend(new RocksDBStateBackend("file:///opt/rocksDb/", true));*/
 
 
             Properties kafkaProperties = new Properties();
-            // kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
-            kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
+             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(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
@@ -72,36 +68,23 @@ 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("bootstrap.servers", "dn3.zhht:9092");
+          //  kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
             kafkaProperties2.setProperty("group.id", "cycle_signal2");
             kafkaProperties2.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
             kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
 
-            //旅行事件    
-/*            FlinkKafkaConsumer<TravelEvent> travelEvent =
-                    new FlinkKafkaConsumer<TravelEvent>("trips_event_info",
-                            new TravelEventKafkaSchema(),
-                            kafkaProperties) {
-                    };
-            DataStreamSource<TravelEvent> travelEventStream = env.addSource(travelEvent).setParallelism(3);*/
 
-
-            // FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
-            //   FlinkKafkaConsumer<String> travelInfo = new FlinkKafkaConsumer<>("t_info2", new SimpleStringSchema(), kafkaProperties);
+            //旅行信息数据
+         //   FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
             FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
-          //  travelInfo.setStartFromEarliest();
             DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
-            //  travelInfoStream.print();
-
             //周期信号
-            // FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
+        //    FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
             FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
-           // cycleSignalData.setStartFromEarliest();
             DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
 
-            // cycleSignalDataStream.print();
 
             //2、使用connect，把2个流联合处理
             ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelLaneInfoSink.java

@@ -6,6 +6,7 @@ import com.zhht.irn.utils.DruidConnectPoolUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 
+import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.text.SimpleDateFormat;
 import java.util.Date;
@@ -20,22 +21,26 @@ import java.util.List;
  */
 public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo>> {
 
-    private transient DruidPooledConnection connection;
+    Connection connection;
 
 
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         connection = DruidConnectPoolUtils.getConnection();
+        System.out.println("建立数据库连接成功。。。。。。。");
 
     }
 
     @Override
     public void close() throws Exception {
         super.close();
+        if(connection!=null){
             connection.close();
         }
 
+    }
+
     @Override
     public void invoke(List<TravelLineSinkInfo> value, Context context) throws Exception {
         super.invoke(value, context);
@@ -79,7 +84,9 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
             preparedStatement.setDouble(17, t.getAverage_car_head_time_gap());
             preparedStatement.addBatch();
         }
-        preparedStatement.executeBatch();
+        int[] ints = preparedStatement.executeBatch();
+
+        System.out.println("写入数据成功。。。。。"+ints.length+"...");
 
 
     }