Skip to content

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

旅行信息任务代码提交，未完成，请参考TODO继续编写

parent 556e298e
Showing with 491 additions and 55 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/CycleInfo.java
......@@ -48,44 +48,3 @@ public class CycleInfo {
// 相位列表集合
private List<Stage> stageList;
}
class Stage {
// ID
private Integer id;
// 相位值
private String phaseValue;
// 相位开始时间
private Date beginDateTime;
// 相位结束时间
private Date endDateTime;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
private String endControlModelCode;
// 开始方案ID
private Integer beginPlanId;
// 结束方案ID
private Integer endPlanId;
// 周期时长，单位 秒
private Integer duration;
// 绿灯时长，单位 秒
private Integer green;
// 黄灯时长，单位 秒
private Integer yellow;
// 全红时长，单位 秒
private Integer allRed;
// 周期顺序
private Integer phaseOrder;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/Location.java 0 → 100644
package com.zhht.irn.entity.kafka;
import java.util.Date;
public class Location {
// 位置的经度
private Double longitude;
// 位置的纬度
private Double latitude;
// 此时的速度
private Double speed;
// 此时的时间
private Date dtTranjectory;
// 此时车辆所处的车道ID
private String lineId;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/Stage.java 0 → 100644
package com.zhht.irn.entity.kafka;
import java.util.Date;
public class Stage {
// ID
private Integer id;
// 相位值
private String phaseValue;
// 相位开始时间
private Date beginDateTime;
// 相位结束时间
private Date endDateTime;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
private String endControlModelCode;
// 开始方案ID
private Integer beginPlanId;
// 结束方案ID
private Integer endPlanId;
// 周期时长，单位 秒
private Integer duration;
// 绿灯时长，单位 秒
private Integer green;
// 黄灯时长，单位 秒
private Integer yellow;
// 全红时长，单位 秒
private Integer allRed;
// 周期顺序
private Integer phaseOrder;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/TravelEventInfo.java
package com.zhht.irn.entity.kafka;
import com.zhht.irn.EventType;
import com.zhht.irn.enums.EventType;
import java.util.Date;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/TravelInfo.java
......@@ -93,21 +93,220 @@ public class TravelInfo {
// 备注
private String remark;
}
class Location {
// 位置的经度
private Double longitude;
public Integer getId() {
return id;
}
// 位置的纬度
private Double latitude;
public void setId(Integer id) {
this.id = id;
}
// 此时的速度
private Double speed;
public String getCrossId() {
return crossId;
}
// 此时的时间
private Date dtTranjectory;
public void setCrossId(String crossId) {
this.crossId = crossId;
}
// 此时车辆所处的车道ID
private String lineId;
public String getRecordId() {
return recordId;
}
public void setRecordId(String recordId) {
this.recordId = recordId;
}
public Date getRecordTime() {
return recordTime;
}
public void setRecordTime(Date recordTime) {
this.recordTime = recordTime;
}
public Long getCarId() {
return carId;
}
public void setCarId(Long carId) {
this.carId = carId;
}
public String getCarColor() {
return carColor;
}
public void setCarColor(String carColor) {
this.carColor = carColor;
}
public String getPlate() {
return plate;
}
public void setPlate(String plate) {
this.plate = plate;
}
public String getPlateColor() {
return plateColor;
}
public void setPlateColor(String plateColor) {
this.plateColor = plateColor;
}
public String getCarType() {
return carType;
}
public void setCarType(String carType) {
this.carType = carType;
}
public Long getFirstLineId() {
return firstLineId;
}
public void setFirstLineId(Long firstLineId) {
this.firstLineId = firstLineId;
}
public Double getFirstSpeed() {
return firstSpeed;
}
public void setFirstSpeed(Double firstSpeed) {
this.firstSpeed = firstSpeed;
}
public Date getFirstTime() {
return firstTime;
}
public void setFirstTime(Date firstTime) {
this.firstTime = firstTime;
}
public Long getInCrossLineId() {
return inCrossLineId;
}
public void setInCrossLineId(Long inCrossLineId) {
this.inCrossLineId = inCrossLineId;
}
public Double getInSpeed() {
return inSpeed;
}
public void setInSpeed(Double inSpeed) {
this.inSpeed = inSpeed;
}
public String getInCrossTime() {
return inCrossTime;
}
public void setInCrossTime(String inCrossTime) {
this.inCrossTime = inCrossTime;
}
public long getOutCrossLineId() {
return outCrossLineId;
}
public void setOutCrossLineId(long outCrossLineId) {
this.outCrossLineId = outCrossLineId;
}
public double getOutSpeed() {
return outSpeed;
}
public void setOutSpeed(double outSpeed) {
this.outSpeed = outSpeed;
}
public Date getOutCrossTime() {
return outCrossTime;
}
public void setOutCrossTime(Date outCrossTime) {
this.outCrossTime = outCrossTime;
}
public long getAwayLineId() {
return awayLineId;
}
public void setAwayLineId(long awayLineId) {
this.awayLineId = awayLineId;
}
public double getAwaySpeed() {
return awaySpeed;
}
public void setAwaySpeed(double awaySpeed) {
this.awaySpeed = awaySpeed;
}
public Date getAwayTime() {
return awayTime;
}
public void setAwayTime(Date awayTime) {
this.awayTime = awayTime;
}
public long getLastLineId() {
return lastLineId;
}
public void setLastLineId(long lastLineId) {
this.lastLineId = lastLineId;
}
public double getLastSpeed() {
return lastSpeed;
}
public void setLastSpeed(double lastSpeed) {
this.lastSpeed = lastSpeed;
}
public Date getLastTime() {
return lastTime;
}
public void setLastTime(Date lastTime) {
this.lastTime = lastTime;
}
public Double getCrossingTime() {
return crossingTime;
}
public void setCrossingTime(Double crossingTime) {
this.crossingTime = crossingTime;
}
public List<Location> getLocations() {
return locations;
}
public void setLocations(List<Location> locations) {
this.locations = locations;
}
public String getRemark() {
return remark;
}
public void setRemark(String remark) {
this.remark = remark;
}
}
\ No newline at end of file
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/mysql/TripsMetric.java 0 → 100644
package com.zhht.irn.entity.mysql;
import java.util.Date;
/**
* 旅行信息的准实时指标对应
*
* 纬度：车辆、路口
* 指标：平均速度、最高速度、最低速度、停车次数、停车总时长
*
* @author 吴延飞
* @date 2022-11-02 15:53:00
*/
public class TripsMetric {
// 旅行记录Id
private String recordId;
// 捕获时间
private Date captureTime;
// 丢失时间
private Date lostTime;
// 最高时速
private Double maxSpeed;
// 最低时速
private Double minSpeed;
// 平均时速
private Double avgSpeed;
// 停车总时长
private Integer stopDuration;
// 停车总次数
private Integer stopCount;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelInfoJob.java 0 → 100644
package com.zhht.irn.job;
import com.zhht.irn.entity.kafka.Location;
import com.zhht.irn.entity.kafka.TravelInfo;
import com.zhht.irn.entity.mysql.TripsMetric;
import com.zhht.irn.sink.TripsMetricSink;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
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.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
/**
* 旅行信心实时任务
*
* 计算车辆旅行过程中的指标
* 包括 最高、最低、平均速度，停车总时长，实际停车次数
*/
public class TravelInfoJob {
private static final Logger logger = LoggerFactory.getLogger(TravelInfoJob.class);
public static void main(String[] args) {
// 获取流式运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 获取Kafka配置文件
ParameterTool finkArgs = ParameterTool.fromArgs(args);
String kafkaConfigPath = finkArgs.getRequired("kafka.conf");
ParameterTool kafkaTool;
try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
} catch (IOException e) {
logger.error("kafka 配置文件未设置，请检查!");
throw new RuntimeException(e);
}
// 加载Kafka信息 { topic 相关 }
String groupId = kafkaTool.get("group.id", "test1");
String servers = kafkaTool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(kafkaTool.getRequired("kafka.input.topics").split(","));
String autoCommit = kafkaTool.get("enable.auto.commit", "false");
String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", autoCommit);
properties.setProperty("auto.offset.reset",offsetReset);
int checkpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
String checkpointPath = kafkaTool.get("checkpoint.path", "file:///f/workspace/imooc-flink/state");
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
// 基于Kafka 原始JSON格式的字符串，获取消息流
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
// 开始计算旅行信息相关指标
rawStringStream.map(new MapFunction<String, TravelInfo>() {
@Override
public TravelInfo map(String s) throws Exception {
// json 格式字符串转化为 TravelInfo 实体类
return null;
}
}).filter(new FilterFunction<TravelInfo>() {
@Override
public boolean filter(TravelInfo travelInfo) throws Exception {
// 过滤掉垃圾数据
return true;
}
}).map(new MapFunction<TravelInfo, TripsMetric>() {
@Override
public TripsMetric map(TravelInfo travelInfo) throws Exception {
TripsMetric tripsMetric = new TripsMetric();
// 计算各类指标, 塞入tripsMetric @todo
List<Location> locations = travelInfo.getLocations();
return tripsMetric;
}
}).addSink(new TripsMetricSink()); //@todo 插入数据库的逻辑
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TripsMetricSink.java 0 → 100644
package com.zhht.irn.sink;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.entity.mysql.TripsMetric;
import com.zhht.irn.utils.MySQLUtils;
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.util.List;
/**
* domain traffic
*/
public class TripsMetricSink extends RichSinkFunction<TripsMetric> {
Connection connection;
PreparedStatement insertCyclePstmt;
PreparedStatement insertPhasePstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
// String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
// connection = MySQLUtils.getConnection(path);
// String cycleSql="insert into app_cycle_green_ratio(cycle_id,cross_code,begin_time,end_time,duration,green_ratio) values (?,?,?,?,?,?)";
// insertCyclePstmt = connection.prepareStatement(cycleSql);
// String phaseSql="insert into app_phase_green_ratio(cycle_id,cross_code,phase_value,begin_time,end_time,duration,green_ratio,valid_green) values (?,?,?,?,?,?,?,?)";
// insertPhasePstmt = connection.prepareStatement(phaseSql);
}
@Override
public void close() throws Exception {
super.close();
if(insertCyclePstmt != null) insertCyclePstmt.close();
if(insertPhasePstmt != null) insertPhasePstmt.close();
if(connection != null) connection.close();
}
/**
* 来一条数据就执行一次
*
* 1000w的数据 1000w次
*/
@Override
public void invoke(TripsMetric tripsMetric, Context context) throws Exception {
// System.out.println("=====invoke======" + cycle.getId() + "-->" + cycle.getCrossCode());
// insertCyclePstmt.setLong(1, cycle.getId());
// insertCyclePstmt.setString(2, cycle.getCrossCode());
// insertCyclePstmt.setString(3, cycle.getBeginTime());
// insertCyclePstmt.setString(4, cycle.getEndTime());
// insertCyclePstmt.setInt(5,cycle.getDuration());
// insertCyclePstmt.setDouble(6,cycle.getGreenRatio());
// insertCyclePstmt.execute();
// List<Phase> phaseList = cycle.getPhaseList();
// for(int i=0;i<phaseList.size();i++){
// Phase phase = phaseList.get(i);
// insertPhasePstmt.setLong(1,cycle.getId());
// insertPhasePstmt.setString(2,phase.getCrossCode());
// insertPhasePstmt.setString(3,phase.getPhaseValue());
// insertPhasePstmt.setString(4, cycle.getBeginTime());
// insertPhasePstmt.setString(5, cycle.getEndTime());
// insertPhasePstmt.setInt(6,cycle.getDuration());
// insertPhasePstmt.setDouble(7,phase.getGreenRatio());
// insertPhasePstmt.setInt(8,phase.getValidGreen());
// insertPhasePstmt.addBatch();
// }
// insertPhasePstmt.executeBatch();
// insertPhasePstmt.clearBatch();
}
}
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