Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 3fe1e1b7 by 黄准
Options

Merge remote-tracking branch 'origin/master'

# Conflicts: # realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java # realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Stage.java # realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/FlinkKafka2MysqlApp.java
parents 559146fd 1abc0743
Showing with 993 additions and 94 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java
package com.zhht.irn.entity;
import lombok.Data;
import java.util.Date;
import java.util.List;
/**
......@@ -10,24 +7,111 @@ import java.util.List;
*
*
*/
@Data
public class Cycle {
// 周期编号
private Long id;
// 路口编号
private String crossCode;
//信控机编号
private String signalCode;
// 周期开始时间
private String beginTime;
private String beginDateTime;
// 周期结束时间
private String endTime;
private String endDateTime;
private int duration;
// 周期序号
private int cycleOrder;
//绿信比
private Double greenRatio;
//绿灯有效时间
private Double vaildGreen;
//损失时间
private int lossTime;
// 周期内相位列表（依次执行）
private List<Phase> phaseList;
private List<Stage> stageList;
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getCrossCode() {
return crossCode;
}
public void setCrossCode(String crossCode) {
this.crossCode = crossCode;
}
public String getBeginDateTime() {
return beginDateTime;
}
public void setBeginDateTime(String beginDateTime) {
this.beginDateTime = beginDateTime;
}
public String getEndDateTime() {
return endDateTime;
}
public void setEndDateTime(String endDateTime) {
this.endDateTime = endDateTime;
}
public int getDuration() {
return duration;
}
public void setDuration(int duration) {
this.duration = duration;
}
public int getCycleOrder() {
return cycleOrder;
}
public void setCycleOrder(int cycleOrder) {
this.cycleOrder = cycleOrder;
}
public List<Stage> getStageList() {
return stageList;
}
public void setStageList(List<Stage> stageList) {
this.stageList = stageList;
}
public Double getGreenRatio() {
return greenRatio;
}
public void setGreenRatio(Double greenRatio) {
this.greenRatio = greenRatio;
}
public int getLossTime() {
return lossTime;
}
public void setLossTime(int lossTime) {
this.lossTime = lossTime;
}
public String getSignalCode() {
return signalCode;
}
public void setSignalCode(String signalCode) {
this.signalCode = signalCode;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Phase.java deleted 100644 → 0
package com.zhht.irn.entity;
import lombok.Data;
import java.util.Date;
/**
* 相位实体类
*/
@Data
public class Phase {
//周期ID
private Long cycleId;
// 路口编号
private String crossCode;
// 相位编号
private String phaseValue;
// 相位开始时间
private String beginTime;
// 相位结束时间
private String endTime;
// 相位持续时长
private int duration;
// 相位绿灯时间
private int green;
// 相位黄灯时间
private int yellow;
// 相位全红时间
private int allRed;
// 阶段绿信比
private Double greenRatio;
// 阶段有效绿灯时长
private int validGreen;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Stage.java 0 → 100644
package com.zhht.irn.entity;
import java.util.Date;
/**
* 相位实体类
*/
public class Stage {
//周期ID
private Long cycleOrder;
// 路口编号
private String crossCode;
// 相位编号
private String phaseValue;
// 相位开始时间
private String beginTime;
// 相位结束时间
private String endTime;
// 相位持续时长
private int duration;
// 相位绿灯时间
private int green;
// 相位黄灯时间
private int yellow;
// 相位全红时间
private int allRed;
//相位序号
private int phaseOrder;
// 阶段绿信比
private Double greenRatio;
// 阶段有效绿灯时长
private int validGreen;
//损失时间
private int lossTime;
public String getPhaseValue() {
return phaseValue;
}
public void setPhaseValue(String phaseValue) {
this.phaseValue = phaseValue;
}
public String getBeginTime() {
return beginTime;
}
public void setBeginTime(String beginTime) {
this.beginTime = beginTime;
}
public String getEndTime() {
return endTime;
}
public void setEndTime(String endTime) {
this.endTime = endTime;
}
public int getDuration() {
return duration;
}
public void setDuration(int duration) {
this.duration = duration;
}
public int getGreen() {
return green;
}
public void setGreen(int green) {
this.green = green;
}
public int getYellow() {
return yellow;
}
public void setYellow(int yellow) {
this.yellow = yellow;
}
public int getAllRed() {
return allRed;
}
public void setAllRed(int allRed) {
this.allRed = allRed;
}
public Long getCycleOrder() {
return cycleOrder;
}
public void setCycleOrder(Long cycleOrder) {
this.cycleOrder = cycleOrder;
}
public String getCrossCode() {
return crossCode;
}
public void setCrossCode(String crossCode) {
this.crossCode = crossCode;
}
public Double getGreenRatio() {
return greenRatio;
}
public void setGreenRatio(Double greenRatio) {
this.greenRatio = greenRatio;
}
public int getValidGreen() {
return validGreen;
}
public void setValidGreen(int validGreen) {
this.validGreen = validGreen;
}
public int getPhaseOrder() {
return phaseOrder;
}
public void setPhaseOrder(int phaseOrder) {
this.phaseOrder = phaseOrder;
}
public int getLossTime() {
return lossTime;
}
public void setLossTime(int lossTime) {
this.lossTime = lossTime;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/CycleInfo.java 0 → 100644
package com.zhht.irn.entity.kafka;
import java.util.Date;
import java.util.List;
/**
* 周期数据（信控数据）
* 每个路口的信号控制等每完成一个周期，就发送该周期的信息到实时数据流中。
* 数据结构参考最新的实时数据文档 --> 周期数据（信控）
* 全息路口实时数据文档v1.1 (2022.11.1）》
* https://www.yuque.com/docs/share/9cd59d71-6165-46da-ae43-bc5388149672?#
*
*/
public class CycleInfo {
// ID
private Integer id;
// 路口编号
private String crossCode;
// 信控机编号
private String signalCode;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
private String endControlModelCode;
// 开始方案ID
private Integer beginPlanId;
// 结束方案ID
private Integer endPlanId;
// 周期开始时间
private Date beginDateTime;
// 周期结束时间
private Date endDateTime;
// 周期时长，单位 秒
private Double duration;
// 周期顺序
private Integer cycleOrder;
// 相位列表集合
private List<Stage> stageList;
}
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 0 → 100644
package com.zhht.irn.entity.kafka;
import com.zhht.irn.enums.EventType;
import java.util.Date;
/**
* 旅行时间信息
* 车辆在路口旅行过程中，发生 捕获、到达、进入、通过、驶离、丢失这6中类型的事件时发送当前的事件信息
* 数据结构参考最新的实时数据文档 --> 旅行事件信息
* 全息路口实时数据文档v1.1 (2022.11.1）》
* https://www.yuque.com/docs/share/9cd59d71-6165-46da-ae43-bc5388149672?#
*
*/
public class TravelEventInfo {
// ID
private Integer id;
// 路口ID
private String crossId;
// 记录ID
private String recordId;
// 记录时间
private Date recordTime;
// 事件发生时车道ID
private Long eventLineId;
// 事件发生时速度
private Double eventSpeed;
// 事件时间
private Date eventTime;
// 事件类型
private EventType eventType;
// 备注
private String remark;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/TravelInfo.java 0 → 100644
package com.zhht.irn.entity.kafka;
import java.util.Date;
import java.util.List;
/**
* 旅行信息：车辆在路口完成从出现到丢失时的完整过程的记录
* 数据结构参考最新的实时数据文档
* 《全息路口实时数据文档v1.1 (2022.11.1）》 --> 旅行信息
* https://www.yuque.com/docs/share/9cd59d71-6165-46da-ae43-bc5388149672?#
*
* @author 吴延飞
* @date 2022-11-02 09:34:12
*/
public class TravelInfo {
// ID
private Integer id;
// 路口ID
private String crossId;
// 记录ID
private String recordId;
// 记录时间
private Date recordTime;
// 车辆ID
private Long carId;
// 车辆颜色
private String carColor;
// 车牌号码
private String plate;
// 车牌颜色
private String plateColor;
// 车辆类型
private String carType;
// 捕获车道ID
private Long firstLineId;
// 捕获时速度
private Double firstSpeed;
// 捕获时间
private Date firstTime;
// 进入车道ID
private Long inCrossLineId;
// 进入时速度
private Double inSpeed;
// 进入时间
private String inCrossTime;
// 通过车道ID
private long outCrossLineId;
// 通过时速度
private double outSpeed;
// 通过时间
private Date outCrossTime;
// 驶离车道ID
private long awayLineId;
// 驶离时速度
private double awaySpeed;
// 驶离时间
private Date awayTime;
// 丢失车道ID
private long lastLineId;
// 丢失时速度
private double lastSpeed;
// 丢失时间
private Date lastTime;
// 旅行时长，驶离时刻 - 到达时刻， 单位秒
private Double crossingTime;
// 旅行时点位集合
private List<Location> locations;
// 备注
private String remark;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getCrossId() {
return crossId;
}
public void setCrossId(String crossId) {
this.crossId = crossId;
}
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/enums/EventType.java 0 → 100644
package com.zhht.irn.enums;
/**
* 事件类型
* 车辆在路口所能触发的事件类型
* 分为 捕获、到达、进入、通过、驶离、丢失
*/
public enum EventType {
// 1、捕获，2、到达，3、进入，4、通过、5、驶离，6、丢失
FIRST, ARRIVED, INCROSS, OUTCROSS, AWAY, LAST
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/FlinkKafka2MysqlApp.java
package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.LXBSink;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import javax.xml.crypto.Data;
import java.util.List;
/**
......@@ -43,24 +39,26 @@ public class FlinkKafka2MysqlApp {
//绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
Cycle cycle = JSON.parseObject(s, Cycle.class);
Long allValidGreen = 0L;
List<Phase> phaseList = cycle.getPhaseList();
for (Phase phase : phaseList)
{
List<Stage> phaseList = cycle.getStageList();
for (Stage phase : phaseList) {
// 绿灯时间
long green = phase.getGreen();
// 黄灯时间
long yellow = phase.getYellow();
// 全红时间
long allRed = phase.getAllRed();
// 启动损失时间取一般定义：2s
long startLossTime = 2L;
//清场损失时间
long clearLossTime = yellow + allRed - 2;
allValidGreen = allValidGreen + green - startLossTime - clearLossTime;
}
return Tuple2.of(cycle.getCrossCode(), 0.00);
// 绿信比
Double lxb = allValidGreen.doubleValue() / cycle.getDuration();
return Tuple2.of(cycle.getCrossCode(), lxb);
}
});
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/GreenRadioJob.java
......@@ -2,7 +2,7 @@ package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.GreenRadioSink;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
......@@ -20,14 +20,14 @@ public class GreenRadioJob {
// DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
// 目前假设从Kafka接入的数据为json格式的字符串
// Cycle Topic 的消息格式如下：
// {"Id":"","CrossCode":"","BeginTime":"","EndTime":"","Duration":"","CycleOrder":"",
// "detail":[{"PhaseValue":"101","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"102","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"103","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"104","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}]
// {"id":"1","crossCode":"0001","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"300","cycleOrder":"1",
// "stageList":[
// {"phaseValue":"101","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:01:00","duration":"60","green":"30","yellow":"3","allRed":"0","phaseOrder":"1"},
// {"phaseValue":"102","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"240","green":"120","yellow":"3","allRed":"0","phaseOrder":"2"}
// ]
//}
// 造一条数据
DataStream<String> stream = env.fromElements("{\"id\":\"1\",\"crossCode\":\"0001\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"phaseList\":[{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"102\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
DataStream<String> stream = env.fromElements("{\"id\":\"2\",\"crossCode\":\"0001\",\"signalCode\":\"1000001\",\"beginDateTime\":\"2020-08-02 08:10:03\",\"endDateTime\":\"2020-08-02 08:12:11\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"stageList\":[{\"phaseValue\":\"101\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endDateTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\",\"phaseOrder\":\"1\"},{\"phaseValue\":\"102\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endDateTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\",\"phaseOrder\":\"2\"}]}");
DataStream<Cycle> lxbStream = stream.map(new MapFunction<String, Cycle>() {
@Override
public Cycle map(String s) throws Exception {
......@@ -42,16 +42,17 @@ public class GreenRadioJob {
int duration = cycle.getDuration();
String crossCode = cycle.getCrossCode();
double cycleGreenRatio = 0;
List<Phase> phaseList = cycle.getPhaseList();
for (Phase phase : phaseList) {
int cycleLossTime = 0;
List<Stage> stageList = cycle.getStageList();
for (Stage stage : stageList) {
//相位号
String phaseValue = phase.getPhaseValue();
String phaseValue = stage.getPhaseValue();
// 绿灯时间
int green = phase.getGreen();
int green = stage.getGreen();
// 黄灯时间
int yellow = phase.getYellow();
int yellow = stage.getYellow();
// 全红时间
int allRed = phase.getAllRed();
int allRed = stage.getAllRed();
// 启动损失时间，一般定义：2s
int startLossTime = 2;
//清场损失时间
......@@ -66,15 +67,19 @@ public class GreenRadioJob {
//阶段绿信比
double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
System.out.println("******phaseGreenRatio:"+phaseGreenRatio);
//组装阶段绿信比
phase.setGreenRatio(phaseGreenRatio);
phase.setValidGreen(phaseValidGreen);
//组装阶段绿信比、有效绿灯时间、损失时间
stage.setGreenRatio(phaseGreenRatio);
stage.setValidGreen(phaseValidGreen);
stage.setLossTime(phaseLossTime);
//周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
//周期损失时间= 各个阶段损失时间之和
cycleLossTime = cycleLossTime + phaseLossTime;
System.out.println("******cycleGreenRatio:"+cycleGreenRatio);
}
// 组装周期绿信比
// 组装周期绿信比、损失时间
cycle.setGreenRatio(cycleGreenRatio);
cycle.setLossTime(cycleLossTime);
return cycle;
}
});
......
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/GreenRadioSink.java
package com.zhht.irn.sink;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.DateUtils;
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.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
......@@ -17,15 +20,17 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
Connection connection;
PreparedStatement insertCyclePstmt;
PreparedStatement insertPhasePstmt;
Date date = new Date();
SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd :hh:mm:ss");
@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 (?,?,?,?,?,?)";
String cycleSql="insert into app_cycle_green_ratio(id,cross_code,signal_code,begin_time,end_time,duration,cycle_order,green_ratio,loss_time,update_time) 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 (?,?,?,?,?,?,?,?)";
String phaseSql="insert into app_phase_green_ratio(cycle_order,cross_code,phase_value,begin_time,end_time,duration,phase_order,green_ratio,valid_green,loss_time,update_time) values (?,?,?,?,?,?,?,?,?,?,?)";
insertPhasePstmt = connection.prepareStatement(phaseSql);
}
@Override
......@@ -46,22 +51,29 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
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.setString(3, cycle.getSignalCode());
insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertCyclePstmt.setInt(6,cycle.getDuration());
insertCyclePstmt.setInt(7,cycle.getCycleOrder());
insertCyclePstmt.setDouble(8,cycle.getGreenRatio());
insertCyclePstmt.setInt(9,cycle.getLossTime());
insertCyclePstmt.setString(10, DateUtils.getTodayTime());
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());
List<Stage> stageList = cycle.getStageList();
for(int i=0;i<stageList.size();i++){
Stage phase = stageList.get(i);
insertPhasePstmt.setLong(1,cycle.getCycleOrder());
insertPhasePstmt.setString(2,phase.getCrossCode());
insertPhasePstmt.setString(3,phase.getPhaseValue());
insertPhasePstmt.setString(4, cycle.getBeginTime());
insertPhasePstmt.setString(5, cycle.getEndTime());
insertPhasePstmt.setString(4, cycle.getBeginDateTime());
insertPhasePstmt.setString(5, cycle.getEndDateTime());
insertPhasePstmt.setInt(6,cycle.getDuration());
insertPhasePstmt.setDouble(7,phase.getGreenRatio());
insertPhasePstmt.setInt(8,phase.getValidGreen());
insertPhasePstmt.setInt(7,phase.getPhaseOrder());
insertPhasePstmt.setDouble(8,phase.getGreenRatio());
insertPhasePstmt.setInt(9,phase.getValidGreen());
insertPhasePstmt.setInt(10,phase.getLossTime());
insertPhasePstmt.setString(11, DateUtils.getTodayTime());
insertPhasePstmt.addBatch();
}
insertPhasePstmt.executeBatch();
......
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