Skip to content

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

按方向按秒计算的四个指标，逻辑已完成，未测试

parent 1f8737a4
Showing with 397 additions and 3 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/metric/DirectionEvalSecondMetric.java 0 → 100644
package com.zhht.irn.entity.metric;
/**
* 按方向指标计算结果实体类
*
* @author 吴延飞
* @date 2022-11-16 14:49:00
*/
public class DirectionEvalSecondMetric {
private String crossId;
private String lineId;
private long cycleOrder;
private long currentSecond;
private int accArrivedFlow;
private int accQueueLength;
private int stopFlagCount;
private int accStopCount;
public String getCrossId() {
return crossId;
}
public void setCrossId(String crossId) {
this.crossId = crossId;
}
public String getLineId() {
return lineId;
}
public void setLineId(String lineId) {
this.lineId = lineId;
}
public long getCycleOrder() {
return cycleOrder;
}
public void setCycleOrder(long cycleOrder) {
this.cycleOrder = cycleOrder;
}
public long getCurrentSecond() {
return currentSecond;
}
public void setCurrentSecond(long currentSecond) {
this.currentSecond = currentSecond;
}
public int getAccArrivedFlow() {
return accArrivedFlow;
}
public void setAccArrivedFlow(int accArrivedFlow) {
this.accArrivedFlow = accArrivedFlow;
}
public int getAccQueueLength() {
return accQueueLength;
}
public void setAccQueueLength(int accQueueLength) {
this.accQueueLength = accQueueLength;
}
public int getStopFlagCount() {
return stopFlagCount;
}
public void setStopFlagCount(int stopFlagCount) {
this.stopFlagCount = stopFlagCount;
}
public int getAccStopCount() {
return accStopCount;
}
public void setAccStopCount(int accStopCount) {
this.accStopCount = accStopCount;
}
@Override
public String toString() {
return "DirectionEvalSecondMetric{" +
"crossId='" + crossId + '\'' +
", lineId='" + lineId + '\'' +
", cycleOrder=" + cycleOrder +
", currentSecond=" + currentSecond +
", accArrivedFlow=" + accArrivedFlow +
", accQueueLength=" + accQueueLength +
", stopFlagCount=" + stopFlagCount +
", accStopCount=" + accStopCount +
'}';
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/enums/EventType.java
...@@ -3,9 +3,9 @@ package com.zhht.irn.enums; ...@@ -3,9 +3,9 @@ package com.zhht.irn.enums;
/** /**
* 事件类型 * 事件类型
* 车辆在路口所能触发的事件类型 * 车辆在路口所能触发的事件类型
* 分为 捕获、到达、进入、通过、驶离、丢失 * 分为 捕获、到达、进入、通过、驶离、丢失，停车，启动
*/ */
public enum EventType { public enum EventType {
// 1、捕获，2、到达，3、进入，4、通过、5、驶离，6、丢失 // 1、捕获，2、到达，3、进入，4、通过、5、驶离，6、丢失, 7、停车，8、启动
FIRST, ARRIVED, INCROSS, OUTCROSS, AWAY, LAST FIRST, ARRIVED, INCROSS, OUTCROSS, AWAY, LAST, STOP, STARTUP
} }
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalDirectionSecondMetricFunction.java 0 → 100644
package com.zhht.irn.functions;
import com.zhht.irn.consts.Constant;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
import com.zhht.irn.enums.EventType;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.util.Collector;
import java.util.*;
import java.util.stream.Collectors;
/**
* 按方向 按秒计算 方向指标
*
* 交通流量、排队长度指标，停车状态车辆数，停车次数
*
* @author 吴延飞
* @date 2022-11-15 11:23:00
*/
public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle, TravelEvent, DirectionEvalSecondMetric> {
private MapState<String, List<TravelEvent>> mapState;
private ListState<String> listState;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
mapState = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravelEvent", String.class, List.class));
listState = getRuntimeContext().getListState(new ListStateDescriptor<String>("list", String.class));
}
@Override
public void processElement(Cycle cycle, TravelEvent travelEvent, ProcessJoinFunction<Cycle, TravelEvent, DirectionEvalSecondMetric>.Context ctx, Collector<DirectionEvalSecondMetric> out) throws Exception {
// 进入该方法表明，周期已经到了，而且周期已经向过车记录interval join了[-9,1]分钟的旅行事件数据
// 先累计旅行事件数据到 mapState 中, 只累计到达和进入事件，其他类型计算用不上
List<TravelEvent> cachedTravelEvent = mapState.get(cycle.getCrossCode());
if(cachedTravelEvent == null) {
cachedTravelEvent = new ArrayList<>();
mapState.put(cycle.getCrossCode(), cachedTravelEvent);
}
if(travelEvent.getEventType() == EventType.ARRIVED || travelEvent.getEventType() == EventType.INCROSS
|| travelEvent.getEventType() == EventType.STOP || travelEvent.getEventType() == EventType.STARTUP)
cachedTravelEvent.add(travelEvent);
// 确定触发条件
if(Long.parseLong(travelEvent.getEventTime()) - Constant.CYCLE_CALCULATE_LATENESS > Long.parseLong(cycle.getEndDateTime())) {
long cycleStart = Long.parseLong(cycle.getBeginDateTime());
long cycleEnd = Long.parseLong(cycle.getEndDateTime());
// 计算每个车道的到达流量, 按秒计算
for(long start = cycleStart; start <= cycleEnd; start += 1000) {
// 首先过滤出到当前秒的事件数据
long finalStart = start;
if (cachedTravelEvent != null) {
List<TravelEvent> currentEventList = cachedTravelEvent.stream().filter(event -> {
return Long.parseLong(event.getEventTime()) >= cycleStart && Long.parseLong(event.getEventTime()) <= finalStart;
}).collect(Collectors.toList());
// 根据车道分组
Map<String, List<TravelEvent>> lineMap = new HashMap<>(); // key： 车道ID， value：对应到车道的截止到当前时刻的旅行事件记录集合
currentEventList.stream().forEach(record -> {
List<TravelEvent> lineList = lineMap.get(record.getEventLineId());
if(lineList == null) {
lineList = new ArrayList<>();
lineMap.put(record.getEventLineId(), lineList);
}
lineList.add(record);
});
// 计算两个指标，一个 “发生到达，未进入的车辆数”， 一个“发生到达数”，粒度计算到车道级
for(Map.Entry<String, List<TravelEvent>> entry : lineMap.entrySet()) {
List<TravelEvent> travelEventList = entry.getValue();
// 发生到达的旅行记录集合Set
Set<String> recordIdSet = new HashSet<>();
// 每个旅行对应的形成状态，key: 记录ID， value: 停车状态，进入该map表示停过车。
Map<String, Boolean> recordStopStatusMap = new HashMap<>();
// 计算停车状态和停车次数（同一个周期，只算一次）
// 先获取上一个周期所对应的处于停车状态的记录
for(String recordId : listState.get()) {
recordStopStatusMap.put(recordId, true);
}
// 再处理本周期内发生的事件
for(int i = 0; i < travelEventList.size(); i ++) {
// 严格一点，这里要对事件进行时间的排序，暂且不做
TravelEvent e = travelEventList.get(i);
if(e.getEventType() == EventType.STOP) {
recordStopStatusMap.put(e.getRecordId(), true);
} else if(e.getEventType() == EventType.STARTUP) {
recordStopStatusMap.put(e.getRecordId(), false);
}
}
List<TravelEvent> lineArrivedList = travelEventList.stream().filter(record -> {
if (record.getEventType() == EventType.ARRIVED) {
recordIdSet.add(record.getRecordId());
return true;
} else {
return false;
}
}).collect(Collectors.toList());
List<TravelEvent> lineArrivedAndInCrossList = travelEventList.stream().filter(record -> {
return record.getEventType() == EventType.INCROSS && recordIdSet.contains(record.getRecordId());
}).collect(Collectors.toList());
Set<String> arrivedRecordIdSet = new HashSet<>();
Set<String> arrivedAndInCrossRecordIdSet = new HashSet<>();
for(TravelEvent travel: lineArrivedList) {
arrivedRecordIdSet.add(travel.getRecordId());
}
for(TravelEvent travel: lineArrivedAndInCrossList) {
arrivedAndInCrossRecordIdSet.add(travel.getRecordId());
}
int lineArrivedCount = arrivedRecordIdSet.size();
int lineArrivedAndInCrossCount = arrivedAndInCrossRecordIdSet.size();
int stopFlagCount = 0;
int stopCount = 0;
for(Map.Entry<String, Boolean> en : recordStopStatusMap.entrySet()) {
if(en.getValue() == true) stopFlagCount ++;
stopCount ++;
}
DirectionEvalSecondMetric metric = new DirectionEvalSecondMetric();
metric.setCrossId(cycle.getCrossCode());
metric.setLineId(entry.getKey());
metric.setCurrentSecond(start);
metric.setCycleOrder(cycle.getCycleOrder());
metric.setAccArrivedFlow(lineArrivedCount);
metric.setAccQueueLength(lineArrivedCount - lineArrivedAndInCrossCount);
metric.setStopFlagCount(stopFlagCount);
metric.setAccStopCount(stopCount);
out.collect(metric);
}
}
}
// 计算后要清空
cachedTravelEvent.clear();
}
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/DirectionEvalJob.java 0 → 100644
package com.zhht.irn.job;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.consts.Constant;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
import com.zhht.irn.sink.DirectionEvalSecondSink;
import com.zhht.irn.utils.FileUtils;
import com.zhht.irn.utils.FlinkUtils;
import com.zhht.irn.utils.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
/**
* 按方向评价实时任务，按秒计算
*
* 计算路口的方向指标
* 此任务包括 交通流量，排队长度；延误时间、停车次数 四个指标
*
* @author 吴延飞
* @date 2022-11-16 15:03:00
*/
public class DirectionEvalJob {
private static final Logger logger = LoggerFactory.getLogger(DirectionEvalJob.class);
public static void main(String[] args) throws Exception {
// 获取流式运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 获取信控数据-周期结束实时流 Kafka
DataStream<Cycle> cycleStream =
// env.addSource(new CycleMockSource())
FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "directionEvalJob")
.map(record -> {
System.out.println("信控周期->" + record);
return JSON.parseObject(record, Cycle.class);
});
// 获取旅行事件数据实时流 kafka
DataStream<TravelEvent> travelEventStream =
// env.addSource(new TravelEventMockSource())
FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME)
.map(record -> {
System.out.println("旅行事件->" + record);
return JSON.parseObject(record, TravelEvent.class);
}).filter(event -> {
return StringUtils.isNotEmpty(event.getEventLineId());
});
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy
.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3)))
.withTimestampAssigner((cycle, timestamp) -> Long.parseLong(cycle.getEndDateTime()))
.withIdleness(Duration.ofSeconds(600));
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy
.<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((xx, timestamp) -> Long.parseLong(xx.getEventTime()));
// 周期数据 interval join 事件流数据，计算 交通流量和排队长度两个指标
// 旅行事件数据流
KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream.assignTimestampsAndWatermarks(travelEventWatermarkStrategy)
.keyBy(record -> record.getCrossId());
// 周期数据join 旅行事件
cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
.keyBy(record -> record.getCrossCode())
.intervalJoin(travelEventKeyedStream)
.between(Time.minutes(-9), Time.minutes(1))
.process(new CalDirectionSecondMetricFunction())
.addSink(new DirectionEvalSecondSink());
// 旅行数据流
env.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/DirectionEvalSecondSink.java 0 → 100644
package com.zhht.irn.sink;
import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
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;
/**
* 按方向评价实时指标
*
* 结果表：app_direction_eval_second_metric
* 计算结果指标包含 交通流量，排队长度，延误时间，停车次数
*/
public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecondMetric> {
Connection connection;
PreparedStatement insertPstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
connection = MySQLUtils.getConnection();
String insertSql="insert into app_direction_eval_second_metric" +
"(cross_id, line_id, cycle_order, current_second, acc_arrived_flow, acc_queue_length, acc_stop_count, stop_flag_count) values (?, ?, ?, ?, ?, ?, ?, ?)";
insertPstmt = connection.prepareStatement(insertSql);
}
@Override
public void close() throws Exception {
super.close();
if (insertPstmt != null) insertPstmt.close();
if (connection != null) connection.close();
}
@Override
public void invoke(DirectionEvalSecondMetric metric, Context context) throws Exception {
System.out.println("===== insert metric {" + metric + "}");
insertPstmt.setString(1, metric.getCrossId());
insertPstmt.setString(2, metric.getLineId());
insertPstmt.setLong(3, metric.getCycleOrder());
insertPstmt.setLong(4, metric.getCurrentSecond());
insertPstmt.setInt(5, metric.getAccArrivedFlow());
insertPstmt.setInt(6, metric.getAccQueueLength());
insertPstmt.setInt(7, metric.getAccStopCount());
insertPstmt.setInt(8, metric.getStopFlagCount());
insertPstmt.execute();
}
}
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