旅行信息指标计算，发布版。已自测通过

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

+package com.zhht.irn.consts;
+
+/**
+ * 常量类
+ *
+ * @author 吴延飞
+ * @date 2022-11-16 13:41:00
+ */
+public class Constant {
+    public static final String CYCLE_TOPIC_NAME = "signal_cycle_data";
+
+    public static final String TRAVEL_TOPIC_NAME = "trips_info";
+
+    public static final String TRAVEL_EVENT_TOPIC_NAME = "trips_event_info";
+
+    // 周期计算的延迟时间 15秒
+    public static final int CYCLE_CALCULATE_LATENESS = 15 * 1000;
+}

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

 package com.zhht.irn.entity;
 
-import java.util.Date;
-
 public class Location {
+    // 旅行记录ID
+    private String recordId;
+
     // 位置的经度
     private Double longitude;
 
@@ -13,7 +14,7 @@ public class Location {
     private Double speed;
 
     // 此时的时间
-    private Date dtTranjectory;
+    private String dtTranjectory;
 
     // 此时车辆所处的车道ID
     private String lineId;
@@ -42,11 +43,11 @@ public class Location {
         this.speed = speed;
     }
 
-    public Date getDtTranjectory() {
+    public String getDtTranjectory() {
         return dtTranjectory;
     }
 
-    public void setDtTranjectory(Date dtTranjectory) {
+    public void setDtTranjectory(String dtTranjectory) {
         this.dtTranjectory = dtTranjectory;
     }
 
@@ -57,4 +58,24 @@ public class Location {
     public void setLineId(String lineId) {
         this.lineId = lineId;
     }
+
+    public String getRecordId() {
+        return recordId;
+    }
+
+    public void setRecordId(String recordId) {
+        this.recordId = recordId;
+    }
+
+    @Override
+    public String toString() {
+        return "Location{" +
+                "recordId='" + recordId + '\'' +
+                ", longitude=" + longitude +
+                ", latitude=" + latitude +
+                ", speed=" + speed +
+                ", dtTranjectory='" + dtTranjectory + '\'' +
+                ", lineId='" + lineId + '\'' +
+                '}';
+    }
 }

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

@@ -26,6 +26,12 @@ public class Travel {
     // 记录时间
     private String recordTime;
 
+    // 旅行开始时间
+    private String travelBeginTime;
+
+    // 旅行结束时间
+    private String travelEndTime;
+
     // 车辆ID
     private Long carId;
 
@@ -151,6 +157,22 @@ public class Travel {
         this.recordTime = recordTime;
     }
 
+    public String getTravelBeginTime() {
+        return travelBeginTime;
+    }
+
+    public void setTravelBeginTime(String travelBeginTime) {
+        this.travelBeginTime = travelBeginTime;
+    }
+
+    public String getTravelEndTime() {
+        return travelEndTime;
+    }
+
+    public void setTravelEndTime(String travelEndTime) {
+        this.travelEndTime = travelEndTime;
+    }
+
     public Long getCarId() {
         return carId;
     }
@@ -399,7 +421,6 @@ public class Travel {
         this.maxSpeed = maxSpeed;
     }
 
-
     @Override
     public String toString() {
         return "Travel{" +
@@ -407,6 +428,8 @@ public class Travel {
                 ", crossId='" + crossId + '\'' +
                 ", recordId='" + recordId + '\'' +
                 ", recordTime='" + recordTime + '\'' +
+                ", travelBeginTime='" + travelBeginTime + '\'' +
+                ", travelEndTime='" + travelEndTime + '\'' +
                 ", carId=" + carId +
                 ", carColor='" + carColor + '\'' +
                 ", plate='" + plate + '\'' +

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

@@ -4,7 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.sink.TravelMetricSink;
-import com.zhht.irn.source.TravelMockSource;
+import com.zhht.irn.utils.FileUtils;
 import com.zhht.irn.utils.FlinkUtils;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -14,6 +14,8 @@ import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
+import static com.zhht.irn.utils.StringUtils.isNotEmpty;
+
 /**
  * 旅行信息实时任务
  *
@@ -30,31 +32,47 @@ public class TravelInfoJob {
         // 通过kafka配置文件，获取String类型的实时数据流
         DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
 
+        // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查
+        rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
+
         // 开始计算旅行信息相关指标
         rawStringStream.map((MapFunction<String, Travel>) s -> {
-            System.out.println(s);
             // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
             Travel travel = JSON.parseObject(s, Travel.class);
+            // -1 代表该值从未被处理过
+            double minSpeed = -1, maxSpeed = -1, totalSpeed = -1;
 
-            List<Location> locations = travel.getLocations();
+            int accTravel = 0;// 拥有完整开始、结束时刻的旅行记录数
+
+            String travelStart = travel.getTravelBeginTime();
+            String travelEnd = travel.getTravelEndTime();
 
-            int stopCount = 0; long stopTime = 0L;
-            double minSpeed = 0.0, maxSpeed = 0.0, totalSpeed = 0.0;
+            // 如果旅行记录没有完整的开始、结束时刻，则无法计算最高、最低、平均速度
+            if(isNotEmpty(travelStart) && isNotEmpty(travelEnd)) {
 
-            boolean stopped = false; // 停车状态，判断截止到处理时的停车状态 false 行驶状态（默认值），true 停车状态
 
-            for(int i = 0; i < locations.size(); i ++) {
+                long longTravelStart = Long.parseLong(travelStart);
+                long longTravelEnd = Long.parseLong(travelEnd);
 
+                List<Location> locations = travel.getLocations();
+                for (int i = 0; i < locations.size(); i++) {
+                    // 该位置必须在旅行的开始和结束时刻之间才能进行速度计算
+                    long dtTimeStamp = Long.parseLong(locations.get(i).getDtTranjectory());
+                    if (dtTimeStamp >= longTravelStart && dtTimeStamp <= longTravelEnd) {
                         double currentSpeed = locations.get(i).getSpeed();
-                if (i == 0) minSpeed = maxSpeed = currentSpeed;
+                        if (minSpeed == -1) minSpeed = currentSpeed;
+                        if (maxSpeed == -1) maxSpeed = currentSpeed;
+                        if (totalSpeed == -1) totalSpeed = 0;
 
-                // 获取最大最小值
                         if (currentSpeed > maxSpeed) maxSpeed = currentSpeed;
                         if (currentSpeed < minSpeed) minSpeed = currentSpeed;
-
-                // 累计速度
                         totalSpeed += currentSpeed;
+                        accTravel ++;
+                    }
+                }
+            }
 
+            // 停车次数的计算不限于旅行的开始时刻和结束时刻
             // 实际停车为整个旅行中车辆速度低于某个下限阈值的(暂定3km/h)次数. 当车辆一直低于该下限阈值(10km/h)时, 视为一次连续的实际停车.
             // 当车辆高于某个上限阈值(暂定10km/h)后, 再次低于下限阈值(暂定3km/h)时, 再记为一次实际停车
 
@@ -63,27 +81,42 @@ public class TravelInfoJob {
             //
             // 停车次数 = 行驶状态 切换至 停车状态 的次数
             // 状态 stopped 从行驶状态切换停车状态一次, 记一次实际停车数 @宇昂提供的计算思路
+
+            // 车辆默认的停车状态为行驶状态
+            boolean stopped = false;
+
+            List<Location> locations = travel.getLocations();
+            long stopTime = 0;
+            int stopCount = 0;
+
+            for(int j = 0; j < locations.size(); j ++) {
+                double speed = locations.get(j).getSpeed();
                 if(stopped) {
                     // 停车状态下，只关注速度上限, 10km/h
                     // 如果当前速度小于 10km/h 算上一次停车的连续停车，只用累计停车时长，不用更改停车状态
-                    if(currentSpeed >= 10) {
+                    if(speed >= 10) {
                         stopped = false;
                     }
-                    stopTime += (locations.get(i).getDtTranjectory().getTime()
-                           - locations.get(i - 1).getDtTranjectory().getTime());
+                    stopTime += (Long.parseLong(locations.get(j).getDtTranjectory())
+                            - Long.parseLong(locations.get(j - 1).getDtTranjectory()));
                 } else {
                     // 行驶状态下，只关注速度下限, 3km/h, 增加停车次数
-                    if (currentSpeed < 3){
+                    if (speed < 3){
                         stopped = true;
                         stopCount ++;
                     }
                 }
             }
 
+
             // 获取5个实时旅行信息指标
             travel.setStopCount(stopCount);
             travel.setStopTime(stopTime);
-            travel.setAvgSpeed(totalSpeed / locations.size());
+            if(totalSpeed == -1) {
+                travel.setAvgSpeed(-1);
+            } else {
+                travel.setAvgSpeed(totalSpeed / accTravel);
+            }
             travel.setMaxSpeed(maxSpeed);
             travel.setMinSpeed(minSpeed);
 

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

@@ -21,12 +21,12 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        // path暂定为固定值
-        String path = "F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
-        connection = MySQLUtils.getConnection(path);
+        connection = MySQLUtils.getConnection();
+
+        //目前值考虑插入，不考虑更新
         String insertSql="" +
                 "insert into app_travel_metric" +
-                "(id, car_id, cross_id, plate, record_id, capture_time, lost_time, arrived_time, incross_time, " +
+                "(incross_line_id, car_id, cross_id, plate, record_id, capture_time, lost_time, arrived_time, incross_time, " +
                 "outcross_time, away_time, stop_count, stop_time, avg_speed, min_speed, max_speed, update_time)" +
                 "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         insertPstmt = connection.prepareStatement(insertSql);
@@ -38,15 +38,10 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
         if(connection != null) connection.close();
     }
 
-    /**
-     * 来一条数据就执行一次
-     *
-     * 1000w的数据  1000w次
-     */
     @Override
     public void invoke(Travel travel, Context context) throws Exception {
-        System.out.println("=====invoke======" + travel.getId() + "-->" + travel.getRecordId());
-        insertPstmt.setLong(1, travel.getId());
+        System.out.println("=====insert travel ======" + travel);
+        insertPstmt.setString(1, travel.getInCrossLineId());
         insertPstmt.setLong(2, travel.getCarId());
         insertPstmt.setString(3, travel.getCrossId());
         insertPstmt.setString(4,travel.getPlate());

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FileUtils.java

+package com.zhht.irn.utils;
+
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.connector.file.sink.FileSink;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ *  文件工具类
+ *
+ * @author 吴延飞
+ * @date 2022-11-18 15:13:00
+ */
+public class FileUtils {
+    public static FileSink<String> getFileSink(String path) {
+        final FileSink<String> sink = FileSink
+                .forRowFormat(new Path(path), new SimpleStringEncoder<String>("UTF-8"))
+                .withRollingPolicy(
+                        DefaultRollingPolicy.builder()
+                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
+                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
+                                .withMaxPartSize(1024 * 1024 * 1024)
+                                .build())
+                .build();
+
+        return sink;
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/StringUtils.java

@@ -25,7 +25,7 @@ public class StringUtils {
      * @return 是否不为空
      */
     public static boolean isNotEmpty(String str) {
-        return str != null && !"".equals(str);
+        return str != null && !"".equals(str) && !str.equals("null");
     }
 
     /**