Merge remote-tracking branch 'origin/master'

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

@@ -32,8 +32,8 @@ public class Travel {
     // 旅行结束时间
     private String travelEndTime;
 
-    // 旅行时长= 旅行结束 - 旅行开始 (毫秒)
-    private long travelDuration;
+    // 流向
+    private String flowDirection;
 
     // 车辆ID
     private Long carId;
@@ -105,7 +105,7 @@ public class Travel {
     private String lastTime;
 
     // 旅行时长，驶离时刻 -  到达时刻， 单位秒
-    private int crossingTime;
+    private double crossingTime;
 
     // 旅行时点位集合
     private List<Location> locations;
@@ -360,11 +360,11 @@ public class Travel {
         this.lastTime = lastTime;
     }
 
-    public int getCrossingTime() {
+    public double getCrossingTime() {
         return crossingTime;
     }
 
-    public void setCrossingTime(int crossingTime) {
+    public void setCrossingTime(double crossingTime) {
         this.crossingTime = crossingTime;
     }
 
@@ -424,12 +424,12 @@ public class Travel {
         this.maxSpeed = maxSpeed;
     }
 
-    public long getTravelDuration() {
-        return travelDuration;
+    public String getFlowDirection() {
+        return flowDirection;
     }
 
-    public void setTravelDuration(long travelDuration) {
-        this.travelDuration = travelDuration;
+    public void setFlowDirection(String flowDirection) {
+        this.flowDirection = flowDirection;
     }
 
     @Override
@@ -441,7 +441,7 @@ public class Travel {
                 ", recordTime='" + recordTime + '\'' +
                 ", travelBeginTime='" + travelBeginTime + '\'' +
                 ", travelEndTime='" + travelEndTime + '\'' +
-                ", travelDuration='" + travelDuration + '\'' +
+                ", flowDirection='" + flowDirection + '\'' +
                 ", carId=" + carId +
                 ", carColor='" + carColor + '\'' +
                 ", plate='" + plate + '\'' +
@@ -466,7 +466,6 @@ public class Travel {
                 ", lastSpeed=" + lastSpeed +
                 ", lastTime='" + lastTime + '\'' +
                 ", crossingTime=" + crossingTime +
-                ", locations=" + locations +
                 ", remark='" + remark + '\'' +
                 ", stopCount=" + stopCount +
                 ", stopTime=" + stopTime +

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

@@ -31,7 +31,7 @@ public class TravelInfoJob {
 
         // 通过kafka配置文件，获取String类型的实时数据流
         DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
-
+//        DataStream<String> rawStringStream = env.readTextFile("./data/2022-11-23--11/.part-820a21c1-f4fa-485e-a3ed-b65e15fefdfa-0.inprogress.59c1cbee-8597-4dd7-a2fc-12c8df8db83d");
         // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查
         rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
 
@@ -46,15 +46,11 @@ public class TravelInfoJob {
 
             String travelStart = travel.getTravelBeginTime();
             String travelEnd = travel.getTravelEndTime();
-            long duration = -1L;
 
             // 如果旅行记录没有完整的开始、结束时刻，则无法计算最高、最低、平均速度
             if(isNotEmpty(travelStart) && isNotEmpty(travelEnd)) {
-
-
                 long longTravelStart = Long.parseLong(travelStart);
                 long longTravelEnd = Long.parseLong(travelEnd);
-                duration = longTravelEnd - longTravelStart;
 
                 List<Location> locations = travel.getLocations();
                 for (int i = 0; i < locations.size(); i++) {
@@ -121,7 +117,6 @@ public class TravelInfoJob {
             }
             travel.setMaxSpeed(maxSpeed);
             travel.setMinSpeed(minSpeed);
-            travel.setTravelDuration(duration);
 
             logger.info("完成旅行信息的指标计算，当前旅行车辆旅行ID{}的实时指标为， 最高速度：{}， 最低速度：{}， 平均速度：{}， 停车次数： {}，" +
                     "停车总时长：{}", travel.getRecordId(), maxSpeed, minSpeed, travel.getAvgSpeed(), stopCount, stopTime);

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

 package com.zhht.irn.sink;
 
 import com.zhht.irn.entity.Travel;
-import com.zhht.irn.utils.DateUtils;
-import com.zhht.irn.utils.MySQLUtils;
+import com.zhht.irn.utils.DorisUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+
 import java.sql.Connection;
+import java.sql.Date;
 import java.sql.PreparedStatement;
 
 import static com.zhht.irn.utils.DateUtils.longToString;
@@ -22,15 +23,16 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = MySQLUtils.getConnection();
+        connection = DorisUtils.getConnection();
 
         //目前值考虑插入，不考虑更新
         String insertSql="" +
                 "insert into app_travel_metric" +
-                "(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, travel_begin_time," +
-                " travel_end_time, arrived_line_id, outcross_line_id, away_line_id, duration)" +
-                "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+                "(record_id, cross_id, record_time, travel_begin_time, travel_end_time, flow_direction, car_id, car_color, plate, plate_color, car_type, " +
+                "first_lane_id, first_time, first_speed, arrived_lane_id, arrived_time, arrived_speed, incross_lane_id, incross_time, incross_speed, " +
+                "outcross_lane_id, outcross_time, outcross_speed, away_lane_id, away_time, away_speed, last_lane_id, last_time, last_speed, duration, min_speed, " +
+                "max_speed, avg_speed, stop_count, stop_time, statistic_time)" +
+                "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         insertPstmt = connection.prepareStatement(insertSql);
     }
     @Override
@@ -43,29 +45,49 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     @Override
     public void invoke(Travel travel, Context context) throws Exception {
         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());
-        insertPstmt.setString(5, travel.getRecordId());
-        insertPstmt.setString(6, longToString(travel.getFirstTime(), timeFormat));
-        insertPstmt.setString(7, longToString(travel.getLastTime(), timeFormat));
-        insertPstmt.setString(8, longToString(travel.getArrivedTime(), timeFormat));
-        insertPstmt.setString(9, longToString(travel.getInCrossTime(), timeFormat));
-        insertPstmt.setString(10, longToString(travel.getOutCrossTime(), timeFormat));
-        insertPstmt.setString(11, longToString(travel.getAwayTime(), timeFormat));
-        insertPstmt.setInt(12, travel.getStopCount());
-        insertPstmt.setDouble(13, travel.getStopTime());
-        insertPstmt.setDouble(14, travel.getAvgSpeed());
-        insertPstmt.setDouble(15, travel.getMinSpeed());
-        insertPstmt.setDouble(16, travel.getMaxSpeed());
-        insertPstmt.setString(17, DateUtils.getTodayTime());
-        insertPstmt.setString(18, longToString(travel.getTravelBeginTime(), timeFormat));
-        insertPstmt.setString(19, longToString(travel.getTravelEndTime(), timeFormat));
-        insertPstmt.setString(20, travel.getArrivedLineId());
+        insertPstmt.setString(1, travel.getRecordId());
+        insertPstmt.setString(2, travel.getCrossId());
+        insertPstmt.setDate(3, new Date(Long.parseLong(travel.getRecordTime())));
+        insertPstmt.setString(4, longToString(travel.getTravelBeginTime(), timeFormat));
+        insertPstmt.setString(5, longToString(travel.getTravelEndTime(), timeFormat));
+        insertPstmt.setString(6, travel.getFlowDirection());
+        insertPstmt.setLong(7, travel.getCarId());
+        insertPstmt.setString(8, travel.getCarColor());
+        insertPstmt.setString(9, travel.getPlate());
+        insertPstmt.setString(10, travel.getPlateColor());
+        insertPstmt.setString(11, travel.getCarType());
+
+        insertPstmt.setString(12, travel.getFirstLineId());
+        insertPstmt.setString(13, longToString(travel.getFirstTime(), timeFormat));
+        insertPstmt.setDouble(14, travel.getFirstSpeed());
+
+        insertPstmt.setString(15, travel.getArrivedLineId());
+        insertPstmt.setString(16, longToString(travel.getArrivedTime(), timeFormat));
+        insertPstmt.setDouble(17, travel.getArrivedSpeed());
+
+        insertPstmt.setString(18, travel.getInCrossLineId());
+        insertPstmt.setString(19, longToString(travel.getInCrossTime(), timeFormat));
+        insertPstmt.setDouble(20, travel.getInSpeed());
+
         insertPstmt.setString(21, travel.getOutCrossLineId());
-        insertPstmt.setString(22, travel.getAwayLineId());
-        insertPstmt.setLong(23, travel.getTravelDuration());
+        insertPstmt.setString(22, longToString(travel.getOutCrossTime(), timeFormat));
+        insertPstmt.setDouble(23, travel.getOutSpeed());
+
+        insertPstmt.setString(24, travel.getAwayLineId());
+        insertPstmt.setString(25, longToString(travel.getAwayTime(), timeFormat));
+        insertPstmt.setDouble(26, travel.getAwaySpeed());
+
+        insertPstmt.setString(27, travel.getLastLineId());
+        insertPstmt.setString(28, longToString(travel.getLastTime(), timeFormat));
+        insertPstmt.setDouble(29, travel.getLastSpeed());
+
+        insertPstmt.setDouble(30, travel.getCrossingTime());
+        insertPstmt.setDouble(31, travel.getMinSpeed());
+        insertPstmt.setDouble(32, travel.getMaxSpeed());
+        insertPstmt.setDouble(33, travel.getAvgSpeed());
+        insertPstmt.setInt(34, travel.getStopCount());
+        insertPstmt.setDouble(35, travel.getStopTime());
+        insertPstmt.setString(36, longToString(System.currentTimeMillis(), timeFormat));
 
         insertPstmt.execute();
     }

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

+package com.zhht.irn.utils;
+
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+public class DorisUtils {
+
+    public static Connection getConnection() throws Exception {
+        String propertyPath = ".\\realtime\\hologram-streaming\\src\\main\\resources\\doris.properties";
+        return getConnection(propertyPath);
+    }
+
+    public static Connection getConnection(String propertyPath) throws Exception {
+        ParameterTool tool = ParameterTool.fromPropertiesFile(propertyPath);
+
+        // 根据路径的配置文件获取MySQL链接参数
+        String driver = tool.getRequired("driver");
+        String username = tool.getRequired("username");
+        String password = tool.getRequired("password");
+        String host = tool.getRequired("host");
+        String port = tool.getRequired("port");
+        String database = tool.get("database", "default");
+
+        String jdbcUrl = String.format("jdbc:mysql://%s:%s/%s", host, port, database);
+        Class.forName(driver);
+        return DriverManager.getConnection(jdbcUrl, username, password);
+    }
+
+    public static void close(Connection connection, PreparedStatement pstmt) {
+        if(null != pstmt) {
+            try {
+                pstmt.close();
+            } catch (SQLException e) {
+                e.printStackTrace();
+            }
+        }
+
+        if(null != connection) {
+            try {
+                connection.close();
+            } catch (SQLException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}