Merge branch 'rm_2022-12-12' into 'master'

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/Location.java

@@ -2,6 +2,8 @@ package com.zhht.irn.entity.dto;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
 
 /**
  * 位置信息
@@ -11,6 +13,7 @@ import lombok.Data;
  **/
 @Data
 @AllArgsConstructor
+@NoArgsConstructor
 public class Location implements Comparable<Location>{
 
    private Double longitude  ;//可选 经度

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelInfo.java

@@ -11,7 +11,7 @@ import java.util.List;
  * @create 2022-11-14 13:33{
  **/
 @Data
-public class TravelInfo {
+public class TravelInfo implements Comparable<TravelInfo>  {
 
     private Integer id  ;//必填 ID
     private String crossId  ;//必填 路口ID
@@ -45,4 +45,9 @@ public class TravelInfo {
     private Double crossingTime  ;//可选 旅行时长，单位 秒 (s)（驶离时刻-到达时刻）
     private List<Location> locations  ;//必填 数组，参考 位置信息字段
     private String remark  ;//可选 备注
+
+    @Override
+    public int compareTo(TravelInfo o) {
+        return (int)(inCrossTime-o.getInCrossTime());
+    }
 }

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

@@ -2,21 +2,29 @@ package com.zhht.irn.job;
 
 import com.zhht.irn.entity.dto.CycleSignalData;
 import com.zhht.irn.entity.dto.TravelEvent;
+import com.zhht.irn.entity.dto.TravelInfo;
+import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
 import com.zhht.irn.functions.TravelEventAndCycleCoProcessFunction;
 import com.zhht.irn.schema.CycleSignalKafkaSchema;
 import com.zhht.irn.schema.TravelEventKafkaSchema;
+import com.zhht.irn.schema.TravelInfoKafkaSchema;
+import com.zhht.irn.sink.TravelLaneInfoSink;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.ConnectedStreams;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
+
+import java.util.List;
 import java.util.Properties;
 
 /**
@@ -26,7 +34,6 @@ import java.util.Properties;
  * @create 2022-11-14 11:43
  **/
 public class TravelSituationAnalysisJob {
-
     public static void main(String[] args)  {
 
         try {
@@ -36,7 +43,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);
@@ -46,65 +53,58 @@ public class TravelSituationAnalysisJob {
             env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
             //设置重启策略
-            env.setRestartStrategy( RestartStrategies.fixedDelayRestart(5, Time.seconds(10)));
+            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
 
-/*
-           关于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", "dn1.zhht:9092,dn2.zhht:9092,dn3.zhht:9092");
-         //  kafkaProperties.setProperty("bootstrap.servers", "172.25.1.251:9092,172.25.1.122:9092,172.25.1.67:9092");
-            kafkaProperties.setProperty("group.id", "travel_event");
+             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");
             kafkaProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
 
 
             Properties kafkaProperties2 = new Properties();
-           kafkaProperties2.setProperty("bootstrap.servers", "dn1.zhht:9092,dn2.zhht:9092,dn3.zhht:9092");
-         //   kafkaProperties2.setProperty("bootstrap.servers", "172.25.1.251:9092,172.25.1.122:9092,172.25.1.67:9092");
-            kafkaProperties2.setProperty("group.id", "cycle_signal");
+              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");
 
-            //旅行事件
-            FlinkKafkaConsumerBase<TravelEvent> travelEvent =
-                    new FlinkKafkaConsumer<TravelEvent>("trips_event_info",
-                            new TravelEventKafkaSchema(),
-                            kafkaProperties) {
-                    };
-            DataStreamSource<TravelEvent> travelEventStream = env.addSource(travelEvent);
 
-           travelEventStream.print();
+            //旅行信息数据
+         //   FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
+            FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
+            DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
             //周期信号
-            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);
             DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
 
-            cycleSignalDataStream.print();
-
 
             //2、使用connect，把2个流联合处理
-        ConnectedStreams<TravelEvent, CycleSignalData> connect = travelEventStream
+            ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
                     .connect(cycleSignalDataStream);
 
-        ConnectedStreams<TravelEvent, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
-                (KeySelector<TravelEvent, String>) travelInfo1 -> travelInfo1.getCrossId(),
+            ConnectedStreams<TravelInfo, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
+                    (KeySelector<TravelInfo, String>) travelInfo1 -> travelInfo1.getCrossId(),
                     (KeySelector<CycleSignalData, String>) cycleSignalData1 -> cycleSignalData1.getCrossCode()
             );
 
-        travelInfoCycleSignalDataConnectedStreams.process(new TravelEventAndCycleCoProcessFunction());
+            SingleOutputStreamOperator<List<TravelLineSinkInfo>> sinkDataStream = travelInfoCycleSignalDataConnectedStreams
+                    .process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
 
-        env.execute();
+            sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
+
+            env.execute("TravelSituationAnalysisJob");
 
         } catch (Exception e){
             e.printStackTrace();
         }
 
+
     }
 }

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

+package com.zhht.irn.sink;
+
+import com.alibaba.druid.pool.DruidPooledConnection;
+import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
+import com.zhht.irn.utils.DruidConnectPoolUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.log4j.Logger;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * TravelLaneInfoSink
+ *
+ * @author Rui meng
+ * @description 行车情况信息sink
+ * @date 2022/12/15 9:39
+ */
+public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo>> {
+
+    private static Logger logger = Logger.getLogger(TravelLaneInfoSink.class);
+
+    Connection connection;
+
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        connection = DruidConnectPoolUtils.getConnection();
+        logger.info("建立数据库连接成功。。。。。。。");
+
+    }
+
+    @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);
+
+        String sql = "INSERT INTO app_cross_line_travel_car_info" +
+                "(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
+                " flow_direction, pass_numbers," +
+                " last_car_inCross_time, average_pass_time, average_pass_speed, average_control_delay, " +
+                "average_stop_delay, average_stop_times, average_car_head_time_gap" +
+                ")" +
+                "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?,?,?,?) ";
+        PreparedStatement preparedStatement = connection.prepareStatement(sql);
+
+        //TODO 这里的时间使用数据携带的时间，使用最后进入路口车辆的时间，这样在补数据时，可以补充到对应的分区下
+        Date date = new Date();
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
+        String statistic_time = sdf.format(date);
+
+
+        for (TravelLineSinkInfo t :value) {
+            // 使用事件数据的时间划分数据分区，这样可以进行补数操作
+            String record_date = sdf2.format(t.getLast_car_inCross_time());
+
+            preparedStatement.setString(1, record_date);
+            preparedStatement.setString(2, t.getCross_id());
+            preparedStatement.setString(3, statistic_time);
+            preparedStatement.setString(4, sdf.format(t.getCycle_begin_time()));
+            preparedStatement.setString(5, sdf.format(t.getCycle_end_time()));
+            preparedStatement.setString(6, t.getLane_id());
+            preparedStatement.setString(7, t.getCycle_id().toString());
+            preparedStatement.setString(8, t.getDirection());
+            preparedStatement.setString(9, t.getFlow_direction());
+            preparedStatement.setInt(10, t.getPass_numbers());
+            preparedStatement.setLong(11, t.getLast_car_inCross_time());
+            preparedStatement.setDouble(12, t.getAverage_pass_time());
+            preparedStatement.setDouble(13, t.getAverage_pass_speed());
+            preparedStatement.setDouble(14, t.getAverage_control_delay());
+            preparedStatement.setDouble(15, t.getAverage_stop_delay());
+            preparedStatement.setDouble(16, t.getAverage_stop_times());
+            preparedStatement.setDouble(17, t.getAverage_car_head_time_gap());
+            preparedStatement.addBatch();
+        }
+        int[] ints = preparedStatement.executeBatch();
+
+        logger.info("写入数据成功。。。。。"+ints.length+"...");
+
+    }
+}