Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit a6c61816 by 吴延飞
Options

修改旅行信息实时计算任务的写入逻辑，改为写入到Doris数据库

parent 153adf3d
Showing with 149 additions and 83 deletions
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;
......@@ -51,7 +51,7 @@ public class Travel {
private String carType;
// 捕获车道ID
private String firstLineId;
private String firstLaneId;
// 捕获时速度
private Double firstSpeed;
......@@ -60,7 +60,7 @@ public class Travel {
private String firstTime;
// 达到车道ID
private String arrivedLineId;
private String arrivedLaneId;
// 到达时速度
private Double arrivedSpeed;
......@@ -69,7 +69,7 @@ public class Travel {
private String arrivedTime;
// 进入车道ID
private String inCrossLineId;
private String inCrossLaneId;
// 进入时速度
private Double inSpeed;
......@@ -78,7 +78,7 @@ public class Travel {
private String inCrossTime;
// 通过车道ID
private String outCrossLineId;
private String outCrossLaneId;
// 通过时速度
private Double outSpeed;
......@@ -87,7 +87,7 @@ public class Travel {
private String outCrossTime;
// 驶离车道ID
private String awayLineId;
private String awayLaneId;
// 驶离时速度
private Double awaySpeed;
......@@ -96,7 +96,7 @@ public class Travel {
private String awayTime;
// 丢失车道ID
private String lastLineId;
private String lastLaneId;
// 丢失时速度
private Double lastSpeed;
......@@ -105,7 +105,7 @@ public class Travel {
private String lastTime;
// 旅行时长，驶离时刻 - 到达时刻， 单位秒
private int crossingTime;
private double crossingTime;
// 旅行时点位集合
private List<Location> locations;
......@@ -216,12 +216,12 @@ public class Travel {
this.carType = carType;
}
public String getFirstLineId() {
return firstLineId;
public String getFirstLaneId() {
return firstLaneId;
}
public void setFirstLineId(String firstLineId) {
this.firstLineId = firstLineId;
public void setFirstLaneId(String firstLaneId) {
this.firstLaneId = firstLaneId;
}
public Double getFirstSpeed() {
......@@ -240,12 +240,12 @@ public class Travel {
this.firstTime = firstTime;
}
public String getArrivedLineId() {
return arrivedLineId;
public String getArrivedLaneId() {
return arrivedLaneId;
}
public void setArrivedLineId(String arrivedLineId) {
this.arrivedLineId = arrivedLineId;
public void setArrivedLaneId(String arrivedLaneId) {
this.arrivedLaneId = arrivedLaneId;
}
public Double getArrivedSpeed() {
......@@ -264,12 +264,12 @@ public class Travel {
this.arrivedTime = arrivedTime;
}
public String getInCrossLineId() {
return inCrossLineId;
public String getInCrossLaneId() {
return inCrossLaneId;
}
public void setInCrossLineId(String inCrossLineId) {
this.inCrossLineId = inCrossLineId;
public void setInCrossLaneId(String inCrossLaneId) {
this.inCrossLaneId = inCrossLaneId;
}
public Double getInSpeed() {
......@@ -288,12 +288,12 @@ public class Travel {
this.inCrossTime = inCrossTime;
}
public String getOutCrossLineId() {
return outCrossLineId;
public String getOutCrossLaneId() {
return outCrossLaneId;
}
public void setOutCrossLineId(String outCrossLineId) {
this.outCrossLineId = outCrossLineId;
public void setOutCrossLaneId(String outCrossLaneId) {
this.outCrossLaneId = outCrossLaneId;
}
public Double getOutSpeed() {
......@@ -312,12 +312,12 @@ public class Travel {
this.outCrossTime = outCrossTime;
}
public String getAwayLineId() {
return awayLineId;
public String getAwayLaneId() {
return awayLaneId;
}
public void setAwayLineId(String awayLineId) {
this.awayLineId = awayLineId;
public void setAwayLaneId(String awayLaneId) {
this.awayLaneId = awayLaneId;
}
public Double getAwaySpeed() {
......@@ -336,12 +336,12 @@ public class Travel {
this.awayTime = awayTime;
}
public String getLastLineId() {
return lastLineId;
public String getLastLaneId() {
return lastLaneId;
}
public void setLastLineId(String lastLineId) {
this.lastLineId = lastLineId;
public void setLastLaneId(String lastLaneId) {
this.lastLaneId = lastLaneId;
}
public Double getLastSpeed() {
......@@ -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,32 +441,31 @@ public class Travel {
", recordTime='" + recordTime + '\'' +
", travelBeginTime='" + travelBeginTime + '\'' +
", travelEndTime='" + travelEndTime + '\'' +
", travelDuration='" + travelDuration + '\'' +
", flowDirection='" + flowDirection + '\'' +
", carId=" + carId +
", carColor='" + carColor + '\'' +
", plate='" + plate + '\'' +
", plateColor='" + plateColor + '\'' +
", carType='" + carType + '\'' +
", firstLineId='" + firstLineId + '\'' +
", firstLaneId='" + firstLaneId + '\'' +
", firstSpeed=" + firstSpeed +
", firstTime='" + firstTime + '\'' +
", arrivedLineId='" + arrivedLineId + '\'' +
", arrivedLaneId='" + arrivedLaneId + '\'' +
", arrivedSpeed=" + arrivedSpeed +
", arrivedTime='" + arrivedTime + '\'' +
", inCrossLineId='" + inCrossLineId + '\'' +
", inCrossLaneId='" + inCrossLaneId + '\'' +
", inSpeed=" + inSpeed +
", inCrossTime='" + inCrossTime + '\'' +
", outCrossLineId='" + outCrossLineId + '\'' +
", outCrossLaneId='" + outCrossLaneId + '\'' +
", outSpeed=" + outSpeed +
", outCrossTime='" + outCrossTime + '\'' +
", awayLineId='" + awayLineId + '\'' +
", awayLaneId='" + awayLaneId + '\'' +
", awaySpeed=" + awaySpeed +
", awayTime='" + awayTime + '\'' +
", lastLineId='" + lastLineId + '\'' +
", lastLaneId='" + lastLaneId + '\'' +
", 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(21, travel.getOutCrossLineId());
insertPstmt.setString(22, travel.getAwayLineId());
insertPstmt.setLong(23, travel.getTravelDuration());
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.getFirstLaneId());
insertPstmt.setString(13, longToString(travel.getFirstTime(), timeFormat));
insertPstmt.setDouble(14, travel.getFirstSpeed());
insertPstmt.setString(15, travel.getArrivedLaneId());
insertPstmt.setString(16, longToString(travel.getArrivedTime(), timeFormat));
insertPstmt.setDouble(17, travel.getArrivedSpeed());
insertPstmt.setString(18, travel.getInCrossLaneId());
insertPstmt.setString(19, longToString(travel.getInCrossTime(), timeFormat));
insertPstmt.setDouble(20, travel.getInSpeed());
insertPstmt.setString(21, travel.getOutCrossLaneId());
insertPstmt.setString(22, longToString(travel.getOutCrossTime(), timeFormat));
insertPstmt.setDouble(23, travel.getOutSpeed());
insertPstmt.setString(24, travel.getAwayLaneId());
insertPstmt.setString(25, longToString(travel.getAwayTime(), timeFormat));
insertPstmt.setDouble(26, travel.getAwaySpeed());
insertPstmt.setString(27, travel.getLastLaneId());
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 0 → 100644
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();
}
}
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment