Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 70ffd196 by 黄准
Options

车道指标相关计算逻辑

parent 6bb3ebdf
Showing with 1101 additions and 406 deletions
realtime/hologram-streaming/pom.xml
......@@ -69,6 +69,16 @@
<artifactId>lombok</artifactId>
<version>1.18.20</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java</artifactId>
<version>1.12.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.11</artifactId>
<version>1.12.0</version>
</dependency>
</dependencies>
</project>
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java
......@@ -16,9 +16,9 @@ public class Cycle {
//信控机编号
private String signalCode;
// 周期开始时间
private Date beginDateTime;
private String beginDateTime;
// 周期结束时间
private Date endDateTime;
private String endDateTime;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
......@@ -30,11 +30,11 @@ public class Cycle {
// 周期时长，单位 秒
private int duration;
// 周期序号
private int cycleOrder;
private Long cycleOrder;
//绿信比
private Double greenRatio;
//损失时间
private int lossTime;
private int lossTime;
// 周期内相位列表（依次执行）
private List<Stage> stageList;
......@@ -54,19 +54,19 @@ public class Cycle {
this.crossCode = crossCode;
}
public Date getBeginDateTime() {
public String getBeginDateTime() {
return beginDateTime;
}
public void setBeginDateTime(Date beginDateTime) {
public void setBeginDateTime(String beginDateTime) {
this.beginDateTime = beginDateTime;
}
public Date getEndDateTime() {
public String getEndDateTime() {
return endDateTime;
}
public void setEndDateTime(Date endDateTime) {
public void setEndDateTime(String endDateTime) {
this.endDateTime = endDateTime;
}
......@@ -78,11 +78,11 @@ public class Cycle {
this.duration = duration;
}
public int getCycleOrder() {
public Long getCycleOrder() {
return cycleOrder;
}
public void setCycleOrder(int cycleOrder) {
public void setCycleOrder(Long cycleOrder) {
this.cycleOrder = cycleOrder;
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/DirectionNorm.java deleted 100644 → 0
package com.zhht.irn.entity;
import lombok.Data;
import java.util.List;
/**
* @Description: 方向指标
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-10-29-20:47
**/
@Data
public class DirectionNorm extends LaneNorm{
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/LaneNorm.java
......@@ -11,22 +11,28 @@ public class LaneNorm extends Stage{
private String corssId;
//车道编号
private String laneId;
//车道名称
private String laneName;
//车道可视长度
private Integer laneLength;
//相位编号
private String phaseId;
//方向
private String direction;
//交通流量
private String trafficCapacity;
private Double trafficCapacity;
//最大排队长度
private String maxQueueLength;
private Integer maxQueueLength;
//有效绿灯时间
private String effectGreenTime;
//通行能力
private String passCapacity;
private Double passCapacity;
//饱和度
private String saturation;
private Double saturation;
//剩余承载力
private String residualCapacity;
private Double residualCapacity;
//识别空间占有率
private String spotSpaceShare;
//层级
private Integer order;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Stage.java
package com.zhht.irn.entity;
import java.util.Date;
import java.util.List;
/**
* 相位实体类
......@@ -14,9 +15,9 @@ public class Stage {
// 相位编号
private String phaseValue;
// 相位开始时间
private Date beginDateTime;
private String beginDateTime;
// 相位结束时间
private Date endDateTime;
private String endDateTime;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
......@@ -39,9 +40,19 @@ public class Stage {
private Double greenRatio;
// 阶段有效绿灯时长
private int validGreen;
//损失时间
private int lossTime;
//相位允许通行的方向
private List<StageForDirection> stageForDirectionList;
public List<StageForDirection> getStageForDirectionList() {
return stageForDirectionList;
}
public void setStageForDirectionList(List<StageForDirection> stageForDirectionList) {
this.stageForDirectionList = stageForDirectionList;
}
public String getPhaseValue() {
return phaseValue;
......@@ -131,19 +142,19 @@ public class Stage {
this.lossTime = lossTime;
}
public Date getBeginDateTime() {
public String getBeginDateTime() {
return beginDateTime;
}
public void setBeginDateTime(Date beginDateTime) {
public void setBeginDateTime(String beginDateTime) {
this.beginDateTime = beginDateTime;
}
public Date getEndDateTime() {
public String getEndDateTime() {
return endDateTime;
}
public void setEndDateTime(Date endDateTime) {
public void setEndDateTime(String endDateTime) {
this.endDateTime = endDateTime;
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/StageForDirection.java 0 → 100644
package com.zhht.irn.entity;
import lombok.Data;
import java.util.List;
import java.util.Map;
/**
* 相位实体类
*/
@Data
public class StageForDirection{
private String direction;
private String traffic_flow_direction;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/TravelEvent.java
......@@ -23,16 +23,16 @@ public class TravelEvent {
private String recordId;
// 记录时间
private Date recordTime;
private String recordTime;
// 事件发生时车道ID
private Long eventLineId;
private String eventLineId;
// 事件发生时速度
private Double eventSpeed;
// 事件时间
private Date eventTime;
private String eventTime;
// 事件类型
private EventType eventType;
......@@ -40,6 +40,26 @@ public class TravelEvent {
// 备注
private String remark;
private String arrayTime;
private String inCorssTime;
public String getArrayTime() {
return arrayTime;
}
public void setArrayTime(String arrayTime) {
this.arrayTime = arrayTime;
}
public String getInCorssTime() {
return inCorssTime;
}
public void setInCorssTime(String inCorssTime) {
this.inCorssTime = inCorssTime;
}
public Integer getId() {
return id;
}
......@@ -64,19 +84,11 @@ public class TravelEvent {
this.recordId = recordId;
}
public Date getRecordTime() {
return recordTime;
}
public void setRecordTime(Date recordTime) {
this.recordTime = recordTime;
}
public Long getEventLineId() {
public String getEventLineId() {
return eventLineId;
}
public void setEventLineId(Long eventLineId) {
public void setEventLineId(String eventLineId) {
this.eventLineId = eventLineId;
}
......@@ -88,11 +100,19 @@ public class TravelEvent {
this.eventSpeed = eventSpeed;
}
public Date getEventTime() {
public String getRecordTime() {
return recordTime;
}
public void setRecordTime(String recordTime) {
this.recordTime = recordTime;
}
public String getEventTime() {
return eventTime;
}
public void setEventTime(Date eventTime) {
public void setEventTime(String eventTime) {
this.eventTime = eventTime;
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/LaneMetricJob.java
......@@ -49,11 +49,11 @@ public class LaneMetricJob {
// 旅行事件数据流的 watermark 生成逻辑
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30))
.withTimestampAssigner((travelEvent, timestamp) -> travelEvent.getEventTime().getTime());
.withTimestampAssigner((travelEvent, timestamp) -> DateUtils.dateToStamp(travelEvent.getEventTime()));
// 周期结束数据流的 cycle 生成逻辑
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((cycle, timeStamp) -> cycle.getBeginDateTime().getTime());
.withTimestampAssigner((cycle, timeStamp) -> DateUtils.dateToStamp(cycle.getBeginDateTime()));
// 获取实时旅行事件数据流
// DataStream<TravelEvent> eventStream = FlinkUtils.createKafkaStream(args, env)
......@@ -75,7 +75,6 @@ public class LaneMetricJob {
.map((record) -> {
System.out.println(record);
Cycle cycle = JSON.parseObject(record, Cycle.class);
System.out.println(cycle);
return cycle;
})
// 给数据流添加水印
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/LaneNormJob.java deleted 100644 → 0
package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Car;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.LXBSink;
import com.zhht.irn.sink.LaneNormSink;
import com.zhht.irn.source.CarSource;
import com.zhht.irn.source.CirlceSource;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import java.sql.Connection;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
/**
* 两阶段提交参考文档：
* https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
*/
public class LaneNormJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 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":""}]
// 造一条数据
/* DataStream<String> circleStream = 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\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
DataStream<String> passCarStream = 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\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
*/
//根据corssid 分组 保证同一路口的数据进行一个进程
DataStream<Cycle> cycleDataStreamSource = env.addSource(new CirlceSource()).
assignTimestampsAndWatermarks(WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> DateUtils.dateToStamp(cycle.getEndDateTime())))
.keyBy(new KeySelector<Cycle,String>() {
@Override
public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode();
}
}).map(new MapFunction<Cycle,Cycle>() {
@Override
public Cycle map(Cycle cycle) throws Exception {
cycle = getCycle(cycle);
return cycle;
}
});
DataStream<Car> carStream = env.addSource(new CarSource()).
assignTimestampsAndWatermarks(WatermarkStrategy.<Car>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car,timestamp)-> DateUtils.dateToStamp(car.getEvenTime()))).
keyBy(new KeySelector<Car,String>() {
@Override
public String getKey(Car car) throws Exception {
return car.toString();
}
}).filter(new FilterFunction<Car>() {
@Override
public boolean filter(Car car) throws Exception {
if("arrived".equals(car.getEventType()) || "inCross".equals(car.getEventType())){
return true;
}else{
return false;
}
}
});
cycleDataStreamSource.join(carStream)
.where(new KeySelector<Cycle, String>() {
@Override
public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode();
}
}).equalTo(new KeySelector<Car, String>() {
@Override
public String getKey(Car car) throws Exception {
return car.getCrossId();
}
}).window(new WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window>() {
@Override
public Collection<Window> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, Car> cycleCarTaggedUnion, long l, WindowAssignerContext windowAssignerContext) {
// 这里以周期流的开始和结束时间作为一个Window
Cycle currentCycle = cycleCarTaggedUnion.getOne();
Car car = cycleCarTaggedUnion.getTwo();
TimeWindow tw = new TimeWindow(DateUtils.stringToDate(currentCycle.getBeginDateTime(), "yyyy-MM-dd HH:mm:ss").getTime(), DateUtils.stringToDate(currentCycle.getEndDateTime(), "yyyy-MM-dd HH:mm:ss").getTime());
return null;
}
@Override
public Trigger<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window> getDefaultTrigger(StreamExecutionEnvironment streamExecutionEnvironment) {
return null;
}
@Override
public TypeSerializer<Window> getWindowSerializer(ExecutionConfig executionConfig) {
return null;
}
@Override
public boolean isEventTime() {
return false;
}
});
DataStream<String> apply = cycleDataStreamSource.coGroup(carStream).where(new KeySelector<Cycle, String>() {
@Override
public String getKey(Cycle value) throws Exception {
return value.getCrossCode();
}
}).equalTo(new KeySelector<Car, String>() {
@Override
public String getKey(Car value) throws Exception {
return value.getCrossId();
}
}).window(TumblingEventTimeWindows.of(Time.seconds(1))).apply(
new CoGroupFunction<Cycle, Car, String>() {
@Override
public void coGroup(Iterable<Cycle> first, Iterable<Car> second, Collector<String> out) throws Exception {
StringBuilder sb = new StringBuilder();
//datastream1的数据流集合
for (Cycle tuple3 : first) {
sb.append(JSON.toJSONString(tuple3)).append("\n");
}
//datastream2的数据流集合
for (Car tuple3 : second) {
sb.append(JSON.toJSONString(tuple3)).append("\n");
}
out.collect(sb.toString());
}
});
apply.print();
env.execute();
}
public static String getMysqlPath(){
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
return path;
}
/*获取车道指标*/
public static List<LaneNorm> getLaneNorm(Cycle cycle){
List lists = new ArrayList<LaneNorm>();
return lists;
}
/*获取信控指标 并计算相关属性*/
public static Cycle getCycle(Cycle cycle) throws Exception {
//启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
//清场损失时间 = 黄灯时长 + 全红时长 - 2
//阶段损失时间 = 启动损失时间 + 清场损失时间
//阶段有效绿灯时间 = 绿灯时长+黄灯时长+全红时长 - 阶段损失时间
//阶段绿信比 = 阶段有效绿灯时间/周期时长
//周期绿信比 = 阶段绿信比之和
int duration = cycle.getDuration();
//周期绿信比
double cycleGreenRatio = 0;
//周期有效绿灯时间
double cycleVaildGreen = 0;
List<Stage> phaseList = cycle.getStageList();
for (Stage phase : phaseList) {
//相位号
String phaseValue = phase.getPhaseValue();
// 绿灯时间
int green = phase.getGreen();
// 黄灯时间
int yellow = phase.getYellow();
// 全红时间
int allRed = phase.getAllRed();
// 启动损失时间，一般定义：2s
int startLossTime = 2;
//清场损失时间
int clearLossTime = yellow + allRed - 2;
//阶段损失时间
int phaseLossTime = startLossTime + clearLossTime;
//阶段有效绿灯时间
int phaseValidGreen = green + yellow + allRed - phaseLossTime;
//周期绿灯有效时间
cycleVaildGreen=cycleVaildGreen+phaseValidGreen;
//阶段绿信比
double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
//组装阶段绿信比
phase.setGreenRatio(phaseGreenRatio);
phase.setValidGreen(phaseValidGreen);
//周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
}
// 组装周期绿信比
cycle.setGreenRatio(cycleGreenRatio);
//组装周期绿灯有效时间
return cycle;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java 0 → 100644
package com.zhht.irn.job;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.functions.CalLaneMetricProcessFunction;
import com.zhht.irn.functions.CalLaneMetricProcessFunction2;
import com.zhht.irn.process.LaneNormProcessFunction;
import com.zhht.irn.sink.GreenRadioSink;
import com.zhht.irn.sink.LXBSink;
import com.zhht.irn.sink.LaneNormSink;
import com.zhht.irn.source.CarSource;
import com.zhht.irn.source.CirlceSource;
import com.zhht.irn.trigger.CycleTimeTrigger;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils;
import com.zhht.irn.window.CycleWindowAssigner;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.util.Collector;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
import java.time.Duration;
import java.util.*;
/**
* 两阶段提交参考文档：
* https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
*/
public class LaneNormJob {
private static List<Map> getDictResult(String sql) throws Exception {
Connection connection = MySQLUtils.getConnection("E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties");
Statement statement = connection.createStatement();
List objects = new ArrayList<Map>();
ResultSet resultset = statement.executeQuery(sql);
List list = resultSetToList(resultset);
return list;
}
public static List resultSetToList(ResultSet rs) throws java.sql.SQLException {
if (rs == null) {
return Collections.EMPTY_LIST;
}
ResultSetMetaData md = rs.getMetaData(); //得到结果集(rs)的结构信息，比如字段数、字段名等
int columnCount = md.getColumnCount(); //返回此 ResultSet 对象中的列数
List list = new ArrayList();
Map rowData;
while (rs.next()) {
rowData = new HashMap(columnCount);
for (int i = 1; i <= columnCount; i++) {
// 说明：md.getColumnLabel 获取的是别名，md.getColumnName 获取的是真实的字段名，此处业务需求为获取别名
rowData.put(md.getColumnLabel(i), rs.getObject(i));
}
list.add(rowData);
}
return list;
}
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//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":""}]
// 造一条数据
/* DataStream<String> circleStream = 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\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
DataStream<String> passCarStream = 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\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
*/
KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
@Override
public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode();
}
};
KeySelector<TravelEvent, String> carKeySelector = new KeySelector<TravelEvent, String>() {
@Override
public String getKey(TravelEvent car) throws Exception {
return car.toString();
}
};
MapFunction<Cycle, Cycle> getCycleFunction = new MapFunction<Cycle, Cycle>() {
@Override
public Cycle map(Cycle cycle) throws Exception {
cycle = getCycle(cycle);
return cycle;
}
};
FilterFunction<TravelEvent> filterFunction = new FilterFunction<TravelEvent>() {
@Override
public boolean filter(TravelEvent car) throws Exception {
if (car.getEventType() == EventType.ARRIVED || car.getEventType() == EventType.INCROSS) {
return true;
} else {
return false;
}
}
};
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car, timestamp) -> DateUtils.dateToStamp(car.getEventTime()));
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> DateUtils.dateToStamp(cycle.getEndDateTime()));
//根据corssid 分组 保证同一路口的数据进行一个进程
DataStream<Cycle> cycleDataStreamSource = env.addSource(new CirlceSource()).
assignTimestampsAndWatermarks(cycleWatermarkStrategy)
.keyBy(cycleKeySelect).map(getCycleFunction);
DataStream<TravelEvent> carStream = env.addSource(new CarSource()).
assignTimestampsAndWatermarks(travelEventWatermarkStrategy).
keyBy(carKeySelector).filter(filterFunction);
ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
SingleOutputStreamOperator<LaneNorm> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction());
process.addSink(new LaneNormSink());
/*
SingleOutputStreamOperator<Cycle> cycleSingleOutputStreamOperator = env.addSource(new CirlceSource())
.assignTimestampsAndWatermarks(WatermarkStrategy.<Cycle>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner(new SerializableTimestampAssigner<Cycle>() {
@Override
public long extractTimestamp(Cycle cycle, long l) {
return DateUtils.dateToStamp(cycle.getEndDateTime());
}
})).keyBy(cycleKeySelect).map(getCycleFunction);;
SingleOutputStreamOperator<TravelEvent> travelEventSingleOutputStreamOperator = env.addSource(new CarSource()).
assignTimestampsAndWatermarks(WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3)).
withTimestampAssigner(new SerializableTimestampAssigner<TravelEvent>() {
@Override
public long extractTimestamp(TravelEvent travelEvent, long l) {
return DateUtils.dateToStamp(travelEvent.getEventTime());
}
})).keyBy(carKeySelector).filter(filterFunction);
*/
/*//自定义窗口事件
WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window> windowAssigner = new WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window>() {
MapState<String, List<Car>> valueState;
@Override
public Collection<Window> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, Car> cycleCarTaggedUnion, long l, WindowAssignerContext windowAssignerContext) {
// 这里以周期流的开始和结束时间作为一个Window
Cycle currentCycle = cycleCarTaggedUnion.getOne();
TimeWindow tw = new TimeWindow(DateUtils.stringToDate(currentCycle.getBeginDateTime(), "yyyy-MM-dd HH:mm:ss").getTime(), DateUtils.stringToDate(currentCycle.getEndDateTime(), "yyyy-MM-dd HH:mm:ss").getTime());
Set<Window> map = new HashSet<>();
map.add(tw);
return map;
}
@Override
public Trigger<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window> getDefaultTrigger(StreamExecutionEnvironment streamExecutionEnvironment) {
streamExecutionEnvironment.getConfig();
return null;
}
@Override
public TypeSerializer<Window> getWindowSerializer(ExecutionConfig executionConfig) {
return null;
}
@Override
public boolean isEventTime() {
return true;
}
};
//周期 过车关联数据
JoinFunction<Cycle, Car, CycleCar> joinFunction = new JoinFunction<Cycle, Car, CycleCar>() {
@Override
public CycleCar join(Cycle first, Car second) throws Exception {
CycleCar cycleCar = (CycleCar) first;
cycleCar.setCrossId(second.getCrossId());
cycleCar.setEvenTime(second.getEvenTime());
cycleCar.setEventLineId(second.getEventLineId());
cycleCar.setEventSpeed(second.getEventSpeed());
cycleCar.setRecordId(second.getRecordId());
return cycleCar;
}
};
//封装过车和周期数据
DataStream<CycleCar> apply = cycleDataStreamSource.join(carStream).where(cycleKeySelect).equalTo(carKeySelector).window(windowAssigner).apply(joinFunction);*/
// 通过join操作，获取一个Tuple<List<TravelEvent>, Cycle> 类型的数据流
/*DataStream<Tuple2<Cycle, TravelEvent>> joinStream = cycleDataStreamSource.coGroup(carStream)
// 根据路口分组
.where(cycle -> cycle.getCrossCode())
.equalTo(travelEvent -> travelEvent.getCrossId())
// 根据周期来进行窗口划分
.window(new CycleWindowAssigner())
.trigger(new CycleTimeTrigger())
//
.apply(new CoGroupFunction<Cycle, TravelEvent, Tuple2<Cycle, TravelEvent>>() {
@Override
public void coGroup(Iterable<Cycle> iterable, Iterable<TravelEvent> iterable1, Collector<Tuple2<Cycle, TravelEvent>> collector) throws Exception {
System.out.println(1);
}
});*/
/*cycleSingleOutputStreamOperator.keyBy(cycle -> cycle.getCrossCode()).intervalJoin(travelEventSingleOutputStreamOperator.keyBy(event -> event.getCrossId()))
.between(Time.seconds(-5), Time.seconds(10))
.process(new ProcessJoinFunction<Cycle, TravelEvent, Object>() {
@Override
public void processElement(Cycle left, TravelEvent right, Context ctx, Collector<Object> out) throws Exception {
System.out.println(1);
out.collect(Tuple2.of(left, right));
}
}).print();*/
// stream.keyBy(data -> data.f0).intervalJoin(stream1.keyBy(data -> data.user)
// ).between(Time.seconds(-5), Time.seconds(10))
// .process(new ProcessJoinFunction<Tuple2<String, Long>, Event, String>() {
// @Override
// public void processElement(Tuple2<String, Long> left, Event right, Context ctx, Collector<String> out) {
// System.out.println(1);
// out.collect(right + " -> " + left);
// }
// }).print();
env.execute();
/*joinStream.keyBy(value -> value.f0.getCrossCode())
.process(new CalLaneMetricProcessFunction())
.print();*/
}
public static List<LaneNorm> getlaneNorm(Table t) {
List<LaneNorm> laneNorms = new ArrayList<>();
return laneNorms;
}
public static String getMysqlPath() {
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
return path;
}
/*获取车道指标*/
public static List<LaneNorm> getLaneNorm(Cycle cycle) {
List lists = new ArrayList<LaneNorm>();
return lists;
}
/*获取信控指标 并计算相关属性*/
public static Cycle getCycle(Cycle cycle) throws Exception {
//启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
//清场损失时间 = 黄灯时长 + 全红时长 - 2
//阶段损失时间 = 启动损失时间 + 清场损失时间
//阶段有效绿灯时间 = 绿灯时长+黄灯时长+全红时长 - 阶段损失时间
//阶段绿信比 = 阶段有效绿灯时间/周期时长
//周期绿信比 = 阶段绿信比之和
int duration = cycle.getDuration();
//周期绿信比
double cycleGreenRatio = 0;
//周期有效绿灯时间
double cycleVaildGreen = 0;
List<Stage> phaseList = cycle.getStageList();
for (Stage phase : phaseList) {
//相位号
String phaseValue = phase.getPhaseValue();
// 绿灯时间
int green = phase.getGreen();
// 黄灯时间
int yellow = phase.getYellow();
// 全红时间
int allRed = phase.getAllRed();
// 启动损失时间，一般定义：2s
int startLossTime = 2;
//清场损失时间
int clearLossTime = yellow + allRed - 2;
//阶段损失时间
int phaseLossTime = startLossTime + clearLossTime;
//阶段有效绿灯时间
int phaseValidGreen = green + yellow + allRed - phaseLossTime;
//周期绿灯有效时间
cycleVaildGreen = cycleVaildGreen + phaseValidGreen;
//阶段绿信比
double phaseGreenRatio = Double.valueOf(phaseValidGreen) / duration;
//组装阶段绿信比
phase.setGreenRatio(phaseGreenRatio);
phase.setValidGreen(phaseValidGreen);
//周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
}
// 组装周期绿信比
cycle.setGreenRatio(cycleGreenRatio);
//组装周期绿灯有效时间
return cycle;
}
}
class Event {
public String user;
public String url;
public long timestamp;
public Event(String user, String url, long timestamp) {
this.user = user;
this.url = url;
this.timestamp = timestamp;
}
@Override
public String toString() {
return "Event{" +
"user='" + user + '\'' +
", url='" + url + '\'' +
", timestamp=" + timestamp +
'}';
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/process/LaneNormProcessFunction.java 0 → 100644
package com.zhht.irn.process;
import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils;
import com.zhht.irn.utils.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* @Description:
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-11-04-9:53
**/
public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm> {
// state 管理当前周期中存储的旅行事件
private MapState<String, List<TravelEvent>> cycleTravel;
//存储mysql维表数据
private MapState<String, List<Map>> dict;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
cycleTravel = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravel", String.class, Object.class));
dict = getRuntimeContext().getMapState(new MapStateDescriptor("dict", String.class, Object.class));
}
@Override
public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
Connection connection = getConnection();
//初始化字典数据
if (dict.get("phaseDirection") == null) {
//查询相位方向信息
dict.put("phaseDirection", getDictResult(connection,"SELECT\n" +
" a.ID id,\n" +
" a.Direction direction,\n" +
" b.Name traffic_flow_direction,\n" +
" c.PID phase_no,\n" +
" c.PName phase_name\n" +
"FROM tb_phase_movement a\n" +
"LEFT JOIN trafficflowdirection b\n" +
" on a.TrafficFlowId = b.ID\n" +
"LEFT JOIN dic_phase c\n" +
" on a.PhaseId = c.ID"));
}
Cycle f1 = cycle;
// 代表周期已到，需要进行相应的指标计算了
//周期内所有过车数据
List<TravelEvent> travelEvents = cycleTravel.get(f1.getCrossCode());
if(travelEvents==null){
travelEvents=new ArrayList<>();
}
List<TravelEvent> car = getCar(travelEvents);
//根据车道分割过车数据
Map<String, List<TravelEvent>> carForLane = getCarForLane(car);
//车道道路流向关系
List<Map> laneList = getDictResult(connection,"SELECT lane_id,name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
" where out_in_name='in' and cross_id='"+f1.getCrossCode()+"'");
//获取所有车道信息
List<Map> dictResult = getDictResult(connection,"select a.*,b.saturation_flow_rate from (\n" +
"SELECT lane_id,position,lane_length,GROUP_CONCAT(position,name) as positionName FROM dim_cnt_cross_lane_position\n" +
" where cross_id='" + f1.getCrossCode() + "' and out_in_type=1 group By lane_id,cross_id,position,lane_length)a left join app_atomic_lane_saturation_flow_rate b\n" +
" on a.lane_id=b.lane_id and b.cross_id='" + f1.getCrossCode() + "'");
//查询路口所有方向
List<Map> positionList = getDictResult(connection, "SELECT position FROM dim_cnt_cross_lane_position\n" +
" where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position");
List<Stage> stageList = f1.getStageList();
for (Stage stage : stageList) {
stage.setStageForDirectionList(getStageForDiretion(stage, dict.get("phaseDirection")));
}
List<LaneNorm> list = new ArrayList<LaneNorm>();
for (Map map : dictResult) {
LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
list.add(laneNorm);
//交通流量为0得不参与计算
if(laneNorm!=null &&laneNorm.getTrafficCapacity()!=0d){
out.collect(laneNorm);
}
}
if(list.size()>0){
for (Map map:positionList){
String position = String.valueOf(map.get("position"));
LaneNorm laneNorm = new LaneNorm();
laneNorm.setLaneName("整体");
laneNorm.setOrder(1);
laneNorm.setCycleOrder(cycle.getCycleOrder());
laneNorm.setDirection(position);
laneNorm.setCorssId(cycle.getCrossCode());
//获取所有这个方向得车道 交通流量为0得不参与计算
List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())&&recode.getTrafficCapacity()!=0d).collect(Collectors.toList());
if(collect.size()==0){
continue;
}
//排队长度取方向车道最大值
laneNorm.setMaxQueueLength(collect.stream().max(Comparator.comparingInt(LaneNorm ::getMaxQueueLength)).get().getMaxQueueLength());
//饱和度取方向车道最大值
laneNorm.setSaturation(collect.stream().max(Comparator.comparingDouble(LaneNorm ::getSaturation)).get().getSaturation());
//通行能力求和
laneNorm.setPassCapacity( collect.stream().mapToDouble(LaneNorm::getPassCapacity).sum());
//交通流量求和
laneNorm.setTrafficCapacity(collect.stream().mapToDouble(LaneNorm::getTrafficCapacity).sum());
//剩余承载力求和
laneNorm.setResidualCapacity(collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum());
//有效绿灯时间 相位下此方向非右转得绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForPosition(cycle.getStageList(),position));
//识别空间空间占有率 车道长度需要交通流量为0得数据
int sum = list.stream().filter(recode -> position.equals(recode.getDirection())).mapToInt(LaneNorm::getLaneLength).sum();
int sum1 = collect.stream().mapToInt(LaneNorm::getMaxQueueLength).sum();
BigDecimal bd = new BigDecimal(100*sum1/sum);
Double d = bd.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d+"%("+sum1+"m)");
//发送方向整体信息
out.collect(laneNorm);
}
}
//过滤过期数据 没有进入事件的数据保留到下次计算
cycleTravel.put(f1.getCrossCode(),car.stream().filter(recode->StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList()));
//关闭mysql连接
if(connection!=null){
connection.close();
}
}
@Override
public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
if (cycleTravel.get(event.getCrossId()) == null) {
cycleTravel.put(event.getCrossId(), new ArrayList<>());
}
cycleTravel.get(event.getCrossId()).add(event);
}
private static String gettEffectGreenTimeForPosition(List<Stage> stageList, String position) {
int AllGreenTime=0;
for (Stage stage : stageList) {
List<StageForDirection> stageForDirectionList = stage.getStageForDirectionList();
for (StageForDirection dir : stageForDirectionList) {
if (position.equals(dir.getDirection()) && !"右转".equals(dir.getTraffic_flow_direction())) {
AllGreenTime += stage.getValidGreen();
//同一个相位累加一次即可
break;
}
}
}
return AllGreenTime+"";
}
/**
* 提前计算所有相位允许的流向
*
* @param stage
* @param list
* @return
*/
private static List<StageForDirection> getStageForDiretion(Stage stage, List<Map> list) {
List<StageForDirection> stageForDirections = new ArrayList<>();
for (Map map : list) {
String phase_no = String.valueOf(map.get("phase_no"));
if (phase_no.equals(stage.getPhaseValue())) {
StageForDirection stageForDirection = new StageForDirection();
stageForDirection.setDirection(String.valueOf(map.get("direction")));
stageForDirection.setTraffic_flow_direction(String.valueOf(map.get("traffic_flow_direction")));
stageForDirections.add(stageForDirection);
}
}
return stageForDirections;
}
/**
* @param map 车道信息
* @param cycle 周期信息
* @param car 过车记录
* @param laneList 车道相关信息
* @return
*/
private static LaneNorm getLaneNorm(Map map, Cycle cycle, Map<String, List<TravelEvent>> car, List<Map> laneList) {
LaneNorm laneNorm = new LaneNorm();
//车道属性 分别是车道编码 方向
String lane_id = String.valueOf(map.get("lane_id"));
String positionName = String.valueOf(map.get("positionName"));
Double trafficCapacity = getTrafficCapacity(cycle, car, lane_id);
Integer lane_length = Integer.parseInt(String.valueOf(map.get("lane_length")));
//车道基本信息
laneNorm.setOrder(0);
laneNorm.setCorssId(cycle.getCrossCode());
laneNorm.setDirection(String.valueOf(map.get("position")));
laneNorm.setLaneId(lane_id);
laneNorm.setCycleOrder(cycle.getCycleOrder());
laneNorm.setLaneName(positionName);
laneNorm.setLaneLength(lane_length);
laneNorm.setTrafficCapacity(trafficCapacity);
//如果交通流量为0，则不统计此车道指标数据
if (trafficCapacity==0d) {
return laneNorm;
}
//车道饱和度
int saturation_flow_rate = getSaturation(map);
//车道有效绿灯时间
String[] effcetGreenTime = gettEffectGreenTime(cycle.getStageList(), lane_id, laneList);
//车道绿信比
double laneEffectGreenTime = Double.valueOf(effcetGreenTime[1]) / cycle.getDuration();
//最大排队长度
Integer maxQueueLength = getMaxQueueLength(car, lane_id);
if(maxQueueLength>lane_length){
maxQueueLength=lane_length;
}
double v = saturation_flow_rate * laneEffectGreenTime;
//车道级别指标计算
laneNorm.setMaxQueueLength(maxQueueLength);
laneNorm.setEffectGreenTime(effcetGreenTime[0]);
laneNorm.setPassCapacity(v);
Double saturation = 100 * (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime));
BigDecimal bd1 = new BigDecimal(saturation);
Double d2 = bd1.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSaturation(d2);
//剩余承载力
Double residualCapacity = (saturation_flow_rate * laneEffectGreenTime) - trafficCapacity;
BigDecimal bd = new BigDecimal(residualCapacity);
Double d = bd.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setResidualCapacity(d<0?0.00:d);
//计算空间占有率
BigDecimal spot = new BigDecimal( 100 * maxQueueLength / lane_length);
Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d1+"%("+maxQueueLength+"m)");
return laneNorm;
}
//计算饱和流率
private static int getSaturation(Map map) {
String name = String.valueOf(map.get("name"));
if (map.containsKey("saturation_flow_rate")) {
return Integer.parseInt(String.valueOf(map.get("saturation_flow_rate")));
}
if ("直行".indexOf(name) != -1) {
return 1550;
}
if ("左转".indexOf(name) != -1) {
return 1450;
}
if ("右转".indexOf(name) != -1) {
return 1650;
} else {
return 1650;
}
}
/**
* 计算通行能力
*
* @param cycle
* @param laneCarMap
* @param laneId
* @return
*/
private static Double getTrafficCapacity(Cycle cycle, Map<String, List<TravelEvent>> laneCarMap, String laneId) {
//排队车辆
int i = 0;
Long aLong = DateUtils.dateDiff(cycle.getBeginDateTime(), cycle.getEndDateTime());
try {
if (laneCarMap.containsKey(laneId)) {
i = laneCarMap.get(laneId).size();
}
} catch (Exception e) {
}
if (i == 0) {
return 0d;
}
BigDecimal spot = new BigDecimal(60 * 60 / aLong);
Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
return d1;
}
/**
* 计算有效绿灯时间
*
* @return
*/
private static String[] gettEffectGreenTime(List<Stage> stageList, String laneId, List<Map> laneList) {
String allGreenTime = "";
int AllGreenTime=0;
//从字典中获取当前车道所有流向信息
List<Map> currentLane = laneList.stream().filter(a -> laneId.equals(String.valueOf(a.get("lane_id")))).collect(Collectors.toList());
//获取周期的相位信息
for (Map map : currentLane) {
int greeTime = 0;
String position = String.valueOf(map.get("position"));
String name = String.valueOf(map.get("name"));
for (Stage stage : stageList) {
List<StageForDirection> stageForDirectionList = stage.getStageForDirectionList();
for (StageForDirection dir : stageForDirectionList) {
if (position.equals(dir.getDirection()) && name.equals(dir.getTraffic_flow_direction())) {
greeTime += stage.getValidGreen();
AllGreenTime += stage.getValidGreen();
}
}
}
if(greeTime!=0){
allGreenTime = allGreenTime + greeTime + ",";
}
}
return allGreenTime.length()>1?new String[]{allGreenTime.substring(0,allGreenTime.length()-1),AllGreenTime+""}:new String[]{"0","0"};
}
/**
* 根据过车查询车道的最大长度
*/
private static Integer getMaxQueueLength(Map<String, List<TravelEvent>> laneCarMap, String laneId) {
//排队车辆
int i = 0;
try {
if (laneCarMap.containsKey(laneId)) {
i = laneCarMap.get(laneId).stream().
filter(a -> StringUtils.isNotEmpty(a.getArrayTime()) &&
StringUtils.isEmpty(a.getInCorssTime())).collect(Collectors.toList()).size();
}
} catch (Exception e) {
}
return (i * 6);
}
private static Connection getConnection()throws Exception{
Connection connection = MySQLUtils.getConnection("E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties");
return connection;
}
private static List<Map> getDictResult(Connection connection,String sql) throws Exception {
Statement statement = connection.createStatement();
ResultSet resultset = statement.executeQuery(sql);
List list = resultSetToList(resultset);
return list;
}
public static List resultSetToList(ResultSet rs) throws java.sql.SQLException {
if (rs == null) {
return Collections.EMPTY_LIST;
}
ResultSetMetaData md = rs.getMetaData(); //得到结果集(rs)的结构信息，比如字段数、字段名等
int columnCount = md.getColumnCount(); //返回此 ResultSet 对象中的列数
List list = new ArrayList();
Map rowData;
while (rs.next()) {
rowData = new HashMap(columnCount);
for (int i = 1; i <= columnCount; i++) {
// 说明：md.getColumnLabel 获取的是别名，md.getColumnName 获取的是真实的字段名，此处业务需求为获取别名
rowData.put(md.getColumnLabel(i), rs.getObject(i));
}
list.add(rowData);
}
return list;
}
//处理过车数据 将其合并
private static List<TravelEvent> getCar(List<TravelEvent> cars) {
List<TravelEvent> lists = new ArrayList<TravelEvent>();
Map<String, TravelEvent> map = new HashMap<String, TravelEvent>();
for (TravelEvent event : cars) {
if(!DateUtils.compareDataIsBefore10Min(event.getEventTime())){
continue;
}
if (!map.containsKey(event.getRecordId())) {
map.put(event.getRecordId(),event);
}
TravelEvent travelEvent = map.get(event.getRecordId());
if (event.getEventType() == EventType.ARRIVED) {
//已到达车道作为计算条件
travelEvent.setEventLineId(event.getEventLineId());
travelEvent.setArrayTime(travelEvent.getEventTime());
map.put(event.getRecordId(),travelEvent);
}
//进入路口时间 数据更新
if (event.getEventType() == EventType.INCROSS) {
travelEvent.setArrayTime(travelEvent.getEventTime());
map.put(event.getRecordId(),travelEvent);
}
//丢失事件也确认为进入路口
if(event.getEventType() == EventType.LAST){
travelEvent.setArrayTime(travelEvent.getEventTime());
map.put(event.getRecordId(),travelEvent);
}
}
for (String key : map.keySet()) {
lists.add(map.get(key));
}
return lists;
}
private static Map<String, List<TravelEvent>> getCarForLane(List<TravelEvent> cars) {
Map<String, List<TravelEvent>> stringListHashMap = new HashMap<>();
for (TravelEvent ev : cars) {
if (!stringListHashMap.containsKey(ev.getEventLineId())) {
List<TravelEvent> list = new ArrayList<>();
list.add(ev);
stringListHashMap.put(ev.getEventLineId(), list);
} else {
stringListHashMap.get(ev.getEventLineId()).add(ev);
}
}
return stringListHashMap;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java
......@@ -52,10 +52,10 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
insertCyclePstmt.setLong(1, cycle.getId());
insertCyclePstmt.setString(2, cycle.getCrossCode());
insertCyclePstmt.setString(3, cycle.getSignalCode());
insertCyclePstmt.setString(4, DateUtils.dateToString(cycle.getBeginDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setString(5, DateUtils.dateToString(cycle.getEndDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertCyclePstmt.setInt(6,cycle.getDuration());
insertCyclePstmt.setInt(7,cycle.getCycleOrder());
insertCyclePstmt.setLong(7,cycle.getCycleOrder());
insertCyclePstmt.setDouble(8,cycle.getGreenRatio());
insertCyclePstmt.setInt(9,cycle.getLossTime());
insertCyclePstmt.setString(10, DateUtils.getTodayTime());
......@@ -66,8 +66,8 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
insertPhasePstmt.setLong(1,cycle.getCycleOrder());
insertPhasePstmt.setString(2,phase.getCrossCode());
insertPhasePstmt.setString(3,phase.getPhaseValue());
insertCyclePstmt.setString(4, DateUtils.dateToString(cycle.getBeginDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setString(5, DateUtils.dateToString(cycle.getEndDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertPhasePstmt.setInt(6,cycle.getDuration());
insertPhasePstmt.setInt(7,phase.getPhaseOrder());
insertPhasePstmt.setDouble(8,phase.getGreenRatio());
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java
package com.zhht.irn.sink;
import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
......@@ -30,15 +31,15 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
// path暂定为固定值
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path);
insertPstmt = connection.prepareStatement("insert into wyf_test(cross_code,lxb) values (?,?)");
updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
insertPstmt = connection.prepareStatement("insert into " +
"app_cross_lane_norm(cross_id,lane_id,direction,cycle_id,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," +
"spot_space_share,lane_name,`order`,update_time) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
}
@Override
public void close() throws Exception {
super.close();
if(insertPstmt != null) insertPstmt.close();
if(updatePstmt != null) updatePstmt.close();
if(connection != null) connection.close();
}
......@@ -49,6 +50,20 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
*/
@Override
public void invoke(LaneNorm value, SinkFunction.Context context) throws Exception {
insertPstmt.setString(1, value.getCorssId());
insertPstmt.setString(2, value.getLaneId());
insertPstmt.setString(3, value.getDirection());
insertPstmt.setLong(4, value.getCycleOrder());
insertPstmt.setDouble(5, value.getMaxQueueLength());
insertPstmt.setString(6,value.getEffectGreenTime());
insertPstmt.setDouble(7,value.getPassCapacity());
insertPstmt.setDouble(8,value.getTrafficCapacity());
insertPstmt.setDouble(9,value.getSaturation());
insertPstmt.setDouble(10,value.getResidualCapacity());
insertPstmt.setString(11,value.getSpotSpaceShare());
insertPstmt.setString(12,value.getLaneName());
insertPstmt.setInt(13,value.getOrder());
insertPstmt.setString(14, DateUtils.getTodayTime());
insertPstmt.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CarSource.java
......@@ -4,6 +4,8 @@ import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Car;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
......@@ -17,54 +19,52 @@ import java.util.*;
* @retrun:
* @Creat :2022-11-01-14:40
**/
public class CarSource implements SourceFunction<Car> {
public class CarSource implements SourceFunction<TravelEvent> {
private static boolean isRunning = true;
public static String[] lanePosition= new String[]{"南左转","南直行","南右转","北左转","北直行","北右转","东左转","东直行","东右转","西左转","西直行","西右转"};
public static String[] inlane= new String[]{"1","2","3","4","5","6","7","8","9","10","11","12"};
public static String[] inlane= new String[]{"01","02","03","04","05","06","07","08","09","10","11","12"};
public static String[] status= new String[]{"arrived","inCross","outCross","away"};
@Override
public void run(SourceContext<Car> sourceContext) throws Exception {
public void run(SourceContext<TravelEvent> sourceContext) throws Exception {
while (isRunning){
Calendar beforeTime = Calendar.getInstance();
Date date = new Date();
List<Car> cars = getCar(beforeTime, date);
for (Car car:cars){
List<TravelEvent> cars = getCar(date);
for (TravelEvent car:cars){
sourceContext.collect(car);
}
//每个3秒发送一条数据
Thread.sleep(10);
Thread.sleep(2000);
}
}
public static List<Car> getCar(Calendar beforeTime,Date date){
public static List<TravelEvent> getCar(Date date){
Calendar aftertime = Calendar.getInstance();
Date date1 = new Date();
//模拟到达事件
List<Car> cars = new ArrayList<>();
List<TravelEvent> cars = new ArrayList<>();
String uuid = UUID.randomUUID()+"";
String carId = UUID.randomUUID()+"";
String inLaneId = inlane[new Random().nextInt(12)];
Car arraycar = new Car();
TravelEvent arraycar = new TravelEvent();
arraycar.setRecordId(uuid);
arraycar.setCrossId("13070201");
arraycar.setEvenTime(DateUtils.formatTime(date1));
arraycar.setCrossId("13070202");
arraycar.setEventTime(DateUtils.formatTime(date1));
arraycar.setEventLineId(inLaneId);
arraycar.setEventSpeed(new Random().nextInt(60)+"");
arraycar.setEventType("arrived");
arraycar.setId(carId);
arraycar.setEventSpeed(55.5);
arraycar.setEventType(EventType.ARRIVED);
arraycar.setId(0);
cars.add(arraycar);
//模拟进入路口事件
String afterTimeSecond = DateUtils.getAfterTimeSecond(aftertime, new Random().nextInt(180));
Car incrosscar = new Car();
incrosscar.setCrossId("13070201");
incrosscar.setEvenTime(afterTimeSecond);
String afterTimeSecond = DateUtils.getAfterTimeSecond(new Random().nextInt(140));
TravelEvent incrosscar = new TravelEvent();
incrosscar.setCrossId("13070202");
incrosscar.setEventTime(afterTimeSecond);
incrosscar.setEventLineId(inLaneId);
incrosscar.setRecordId(uuid);
incrosscar.setId(carId);
incrosscar.setEventSpeed(new Random().nextInt(60)+"");
incrosscar.setEventType("inCross");
incrosscar.setId(0);
incrosscar.setEventSpeed(55.2);
incrosscar.setEventType(EventType.INCROSS);
cars.add(incrosscar);
return cars;
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CirlceSource.java
//package com.zhht.irn.source;
//
//import com.alibaba.fastjson.JSON;
//import com.zhht.irn.entity.Cycle;
//import com.zhht.irn.entity.Stage;
//import com.zhht.irn.utils.DateUtils;
//import org.apache.flink.streaming.api.functions.source.SourceFunction;
//import org.apache.flink.util.TimeUtils;
//
//import java.sql.Timestamp;
//import java.util.*;
//
///**
// * @Description:
// * @Author ：Marinh
// * @Param:
// * @retrun:
// * @Creat :2022-11-01-14:40
// **/
//public class CirlceSource implements SourceFunction<Cycle> {
// private static boolean isRunning = true;
// private static int i = 0;
//
// @Override
// public void run(SourceFunction.SourceContext<Cycle> sourceContext) throws Exception {
// while (isRunning){
// Calendar beforeTime = Calendar.getInstance();
// Date date = new Date();
// sourceContext.collect(getCycle(beforeTime,date));
// //每个3秒发送一条数据
// Thread.sleep(60000);
// }
// }
//
// public static Cycle getCycle(Calendar beforeTime,Date date){
// Cycle cycle = new Cycle();
// cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
// cycle.setCrossCode("13070201");
// cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
// cycle.setEndDateTime(DateUtils.formatTime(date));
// cycle.setDuration(120);
// cycle.setCycleOrder(1);
// ArrayList<Stage> phases = new ArrayList<>();
// for(int i=0;i<3;i++){
// Stage phase = new Stage();
// phase.setPhaseValue("10"+i);
// if(i==0){
// phase.setPhaseValue("1");
// phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,120));
// phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
// phase.setDuration(30);
// phase.setGreen(30);
// phase.setAllRed(87);
// phase.setGreen(3);
// }
// if(i==1){
// phase.setPhaseValue("2");
// phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
// phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
// phase.setDuration(30);
// phase.setGreen(30);
// phase.setAllRed(87);
// phase.setGreen(3);
// }
// if(i==2){
// phase.setPhaseValue("3");
// phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
// phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,0));
// phase.setDuration(60);
// phase.setGreen(60);
// phase.setAllRed(57);
// phase.setGreen(3);
// }
// phases.add(phase);
// }
// cycle.setStageList(phases);
// return cycle;
// }
//
// @Override
// public void cancel() {
// isRunning=false;
// }
//
//}
package com.zhht.irn.source;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.TimeUtils;
import java.sql.Timestamp;
import java.util.*;
/**
* @Description:
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-11-01-14:40
**/
public class CirlceSource implements SourceFunction<Cycle> {
private static boolean isRunning = true;
private static Long cycleOrder = 0L;
@Override
public void run(SourceFunction.SourceContext<Cycle> sourceContext) throws Exception {
while (isRunning){
//每个3秒发送一条数据
Thread.sleep(120000);
Calendar beforeTime = Calendar.getInstance();
Date date = new Date();
sourceContext.collect(getCycle(beforeTime,date));
cycleOrder++;
}
}
public static Cycle getCycle(Calendar beforeTime,Date date){
Cycle cycle = new Cycle();
cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
cycle.setCrossCode("13070202");
cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
cycle.setEndDateTime(DateUtils.formatTime(date));
cycle.setDuration(120);
cycle.setCycleOrder(cycleOrder);
ArrayList<Stage> phases = new ArrayList<>();
for(int i=0;i<4;i++){
Stage phase = new Stage();
phase.setPhaseValue("10"+i);
if(i==0){//南北直行
phase.setPhaseValue("8");
phase.setBeginDateTime(DateUtils.getBeforeTimeSecond(120));
phase.setEndDateTime(DateUtils.getBeforeTimeSecond(90));
phase.setDuration(30);
phase.setGreen(30);
phase.setAllRed(87);
phase.setYellow(3);
phase.setCycleOrder(cycleOrder);
}
if(i==1){//南北左转
phase.setPhaseValue("0");
phase.setBeginDateTime(DateUtils.getBeforeTimeSecond(90));
phase.setEndDateTime(DateUtils.getBeforeTimeSecond(60));
phase.setDuration(30);
phase.setGreen(30);
phase.setAllRed(87);
phase.setYellow(3);
phase.setCycleOrder(cycleOrder);
}
if(i==2){//东西直行
phase.setPhaseValue("9");
phase.setBeginDateTime(DateUtils.getBeforeTimeSecond(60));
phase.setEndDateTime(DateUtils.getBeforeTimeSecond(0));
phase.setDuration(60);
phase.setGreen(60);
phase.setAllRed(57);
phase.setYellow(3);
phase.setCycleOrder(cycleOrder);
}
if(i==3){//东西左转
phase.setPhaseValue("4");
phase.setBeginDateTime(DateUtils.getBeforeTimeSecond(60));
phase.setEndDateTime(DateUtils.getBeforeTimeSecond(0));
phase.setDuration(60);
phase.setGreen(60);
phase.setAllRed(57);
phase.setYellow(3);
phase.setCycleOrder(cycleOrder);
}
phases.add(phase);
}
cycle.setStageList(phases);
return cycle;
}
@Override
public void cancel() {
isRunning=false;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DateUtils.java
......@@ -39,6 +39,25 @@ public class DateUtils {
}
public static Long dateDiff(String startTime, String endTime) {
// 按照传入的格式生成一个simpledateformate对象
SimpleDateFormat sd = new SimpleDateFormat(timeFormat);
long ns = 1000;// 一秒钟的毫秒数
long diff;
long sec = 0;
// 获得两个时间的毫秒时间差异
try {
diff = sd.parse(endTime).getTime() - sd.parse(startTime).getTime();
sec = diff / ns;// 计算差多少秒
} catch (ParseException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return sec;
}
public static Map getLM() {
String currentDateTime = DateUtils.getCurrentDateTime(dateFormat);
......@@ -215,6 +234,26 @@ public class DateUtils {
}
}
/**
* 判断时间是否在当前时间十分钟之类
* @param s
* @return
*/
public static boolean compareDataIsBefore10Min(String s){
try {
String res;
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = new Date();
Date afterDate = new Date(simpleDateFormat.parse(s).getTime() + 600000);
return date.compareTo(afterDate)>0?false:true;
}catch (Exception e){
return false;
}
}
/**
* 获取当前时间
* 字符串格式
......@@ -321,7 +360,8 @@ public class DateUtils {
return time;
}
public static String getBeforeTimeSecond(Calendar beforeTime,int before){
public static String getBeforeTimeSecond(int before){
Calendar beforeTime = Calendar.getInstance();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
beforeTime.add(Calendar.SECOND, -before);// 多少秒之前
Date beforeD = beforeTime.getTime();
......@@ -329,7 +369,8 @@ public class DateUtils {
return time;
}
public static String getAfterTimeSecond(Calendar afterTime,int before){
public static String getAfterTimeSecond(int before){
Calendar afterTime = Calendar.getInstance();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
afterTime.add(Calendar.SECOND, before);// 多少秒之后
Date beforeD = afterTime.getTime();
......@@ -1345,6 +1386,40 @@ public class DateUtils {
return daysStrList;
}
public static boolean belongCalendar(String dBegin, String dEnd, String dDate) {
SimpleDateFormat sdf = new SimpleDateFormat(timeFormat);
Date beginTime = null;
Date endTime = null;
Date nowTime = null;
try {
beginTime = sdf.parse(dBegin);
endTime = sdf.parse(dEnd);
nowTime = sdf.parse(dDate);
} catch (ParseException e) {
e.printStackTrace();
}
if (nowTime.getTime() == beginTime.getTime()
|| nowTime.getTime() == endTime.getTime()) {
return true;
}
Calendar date = Calendar.getInstance();
date.setTime(nowTime);
Calendar begin = Calendar.getInstance();
begin.setTime(beginTime);
Calendar end = Calendar.getInstance();
end.setTime(endTime);
if (date.after(begin) && date.before(end)) {
return true;
} else {
return false;
}
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/window/CycleWindowAssigner.java
......@@ -18,8 +18,8 @@ import java.util.Collections;
public class CycleWindowAssigner extends WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
@Override
public Collection<TimeWindow> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, WindowAssignerContext context) {
long start = element.getOne().getBeginDateTime().getTime();
long end = element.getOne().getEndDateTime().getTime();
long start = DateUtils.dateToStamp(element.getOne().getBeginDateTime());
long end = DateUtils.dateToStamp(element.getOne().getEndDateTime());
return Collections.singletonList(new TimeWindow(start, end));
}
......
realtime/hologram-streaming/src/main/resources/mysql.properties
driver=com.mysql.jdbc.Driver
host=127.0.0.1
port=13305
port=13306
username=root
password=mima
database=test
\ No newline at end of file
database=test
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