Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 43dd7c70 by 余根龙
Options

Merge remote-tracking branch 'origin/master'

parents 2b17cfdc dd04942a
Showing with 161 additions and 294 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/LaneNorm.java
...@@ -32,7 +32,12 @@ public class LaneNorm extends Stage{ ...@@ -32,7 +32,12 @@ public class LaneNorm extends Stage{
//剩余承载力 //剩余承载力
private Double residualCapacity; private Double residualCapacity;
//识别空间占有率 //识别空间占有率
private String spotSpaceShare; private Double spotSpaceShare;
//识别空间占有长度
private Integer spotSpaceLength;
//层级 //层级
private Integer order; private Integer order;
//类型
private String type;
} }
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java
...@@ -3,44 +3,21 @@ package com.zhht.irn.job; ...@@ -3,44 +3,21 @@ package com.zhht.irn.job;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.*; import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType; 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.process.LaneNormProcessFunction;
import com.zhht.irn.sink.GreenRadioSink;
import com.zhht.irn.sink.LXBSink;
import com.zhht.irn.sink.LaneNormSink; import com.zhht.irn.sink.LaneNormSink;
import com.zhht.irn.source.CarSource; import com.zhht.irn.source.CarSource;
import com.zhht.irn.source.CirlceSource; import com.zhht.irn.source.CirlceSource;
import com.zhht.irn.trigger.CycleTimeTrigger;
import com.zhht.irn.utils.DateUtils; import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils; import com.zhht.irn.utils.FlinkUtils;
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.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.FilterFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction; 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.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.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 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.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.time.Duration;
import java.util.*; import java.util.*;
...@@ -49,230 +26,60 @@ import java.util.*; ...@@ -49,230 +26,60 @@ import java.util.*;
* https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka * https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
*/ */
public class LaneNormJob { 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 { public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class); DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
// 目前假设从Kafka接入的数据为json格式的字符串 DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
// 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>() { KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
@Override @Override
public String getKey(Cycle cycle) throws Exception { public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode(); return cycle.getCrossCode();
} }
}; };
KeySelector<TravelEvent, String> carKeySelector = new KeySelector<TravelEvent, String>() { KeySelector<TravelEvent, String> carKeySelector = new KeySelector<TravelEvent, String>() {
@Override @Override
public String getKey(TravelEvent car) throws Exception { public String getKey(TravelEvent car) throws Exception {
return car.toString(); return car.toString();
} }
}; };
MapFunction<String, Cycle> getCycleFunction = new MapFunction<String, Cycle>() {
MapFunction<Cycle, Cycle> getCycleFunction = new MapFunction<Cycle, Cycle>() {
@Override @Override
public Cycle map(Cycle cycle) throws Exception { public Cycle map(String cycle) throws Exception {
cycle = getCycle(cycle); Cycle cycle1 = JSON.parseObject(cycle, Cycle.class);
return cycle; cycle1 = getCycle(cycle1);
System.out.println("周期数据->"+cycle);
return cycle1;
}
};
MapFunction<String, TravelEvent> getTravelEventFunction = new MapFunction<String, TravelEvent>() {
@Override
public TravelEvent map(String travelEvent) throws Exception {
System.out.println("旅行事件数据->"+travelEvent);
return JSON.parseObject(travelEvent, TravelEvent.class);
} }
}; };
FilterFunction<TravelEvent> filterFunction = new FilterFunction<TravelEvent>() { FilterFunction<TravelEvent> filterFunction = new FilterFunction<TravelEvent>() {
@Override @Override
public boolean filter(TravelEvent car) throws Exception { public boolean filter(TravelEvent car) throws Exception {
if (car.getEventType() == EventType.ARRIVED || car.getEventType() == EventType.INCROSS) { //只需要到达进入丢失事件
if (car.getEventType() == EventType.ARRIVED || car.getEventType() == EventType.INCROSS || car.getEventType() == EventType.LAST) {
return true; return true;
} else { } else {
return false; return false;
} }
} }
}; };
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car, timestamp) -> Long.parseLong(car.getEventTime()));
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) -> Long.parseLong(cycle.getEndDateTime()));
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> DateUtils.dateToStamp(cycle.getEndDateTime()));
//根据corssid 分组 保证同一路口的数据进行一个进程 //根据corssid 分组 保证同一路口的数据进行一个进程
DataStream<Cycle> cycleDataStreamSource = env.addSource(new CirlceSource()). DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy).keyBy(cycleKeySelect);
assignTimestampsAndWatermarks(cycleWatermarkStrategy) DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).keyBy(carKeySelector).filter(filterFunction);
.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()); ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
SingleOutputStreamOperator<LaneNorm> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()); SingleOutputStreamOperator<LaneNorm> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction());
process.addSink(new LaneNormSink()); 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(); 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;
} }
/*获取信控指标 并计算相关属性*/ /*获取信控指标 并计算相关属性*/
...@@ -322,25 +129,3 @@ public class LaneNormJob { ...@@ -322,25 +129,3 @@ public class LaneNormJob {
return cycle; 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
...@@ -80,13 +80,18 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -80,13 +80,18 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
"SELECT lane_id,position,lane_length,GROUP_CONCAT(position,name) as positionName FROM dim_cnt_cross_lane_position\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" + " 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() + "'"); " 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" + 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"); " where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position");
//查询路口所有的流向
List<Map> positionNameList = getDictResult(connection, " SELECT position,name FROM dim_cnt_cross_lane_position\n" +
" where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position,name");
List<Stage> stageList = f1.getStageList(); List<Stage> stageList = f1.getStageList();
//获取相位关系
for (Stage stage : stageList) { for (Stage stage : stageList) {
stage.setStageForDirectionList(getStageForDiretion(stage, dict.get("phaseDirection"))); stage.setStageForDirectionList(getStageForDiretion(stage, dict.get("phaseDirection")));
} }
//获取所有车道的指标信息
List<LaneNorm> list = new ArrayList<LaneNorm>(); List<LaneNorm> list = new ArrayList<LaneNorm>();
for (Map map : dictResult) { for (Map map : dictResult) {
LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList); LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
...@@ -96,42 +101,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -96,42 +101,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
out.collect(laneNorm); out.collect(laneNorm);
} }
} }
if(list.size()>0){ //获取所有方向的指标
for (Map map:positionList){ getPositionNorm(list,positionList,cycle,out);
String position = String.valueOf(map.get("position")); //获取所有流向的指标
LaneNorm laneNorm = new LaneNorm(); getPositionNameNorm(list,positionNameList,cycle,out);
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())); cycleTravel.put(f1.getCrossCode(),car.stream().filter(recode->StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList()));
//关闭mysql连接 //关闭mysql连接
...@@ -184,6 +157,105 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -184,6 +157,105 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
return stageForDirections; return stageForDirections;
} }
/**
* 生成流向指标
* @param list
* @param positionNameList
* @param cycle
* @param out
*/
private static void getPositionNameNorm(List<LaneNorm> list,List<Map> positionNameList,Cycle cycle , Collector<LaneNorm> out){
if(list.size()>0){
for (Map map:positionNameList){
String position = String.valueOf(map.get("position"));
String name = String.valueOf(map.get("name"));
LaneNorm laneNorm = new LaneNorm();
laneNorm.setLaneName(position+name);
laneNorm.setOrder(0);
laneNorm.setCycleOrder(cycle.getCycleOrder());
laneNorm.setDirection(position);
laneNorm.setCorssId(cycle.getCrossCode());
laneNorm.setType("1");
//获取所有这个方向得车道 交通流量为0得不参与计算
List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())&& recode.getLaneName().indexOf(position+name)!=-1 &&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( new BigDecimal( collect.stream().mapToDouble(LaneNorm::getPassCapacity).sum()).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//交通流量求和
laneNorm.setTrafficCapacity(collect.stream().mapToDouble(LaneNorm::getTrafficCapacity).sum());
//剩余承载力求和
double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
laneNorm.setResidualCapacity(new BigDecimal(sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//有效绿灯时间 相位下此方向非右转得绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForPosition(cycle.getStageList(),position));
//识别空间空间占有率 车道长度需要交通流量为0得数据
int sum = list.stream().filter(recode -> position.equals(recode.getDirection()) && recode.getLaneName().indexOf(position+name)!=-1).mapToInt(LaneNorm::getLaneLength).sum();
int sum1 = collect.stream().mapToInt(LaneNorm::getMaxQueueLength).sum();
BigDecimal bd = new BigDecimal((double)sum1/sum);
Double d = bd.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d>1d?1d:d);
laneNorm.setSpotSpaceLength(sum1);
//发送方向整体信息
out.collect(laneNorm);
}
}
}
/**
* 生成方向指标
* @param list
* @param positionList
* @param cycle
* @param out
*/
private static void getPositionNorm(List<LaneNorm> list,List<Map> positionList,Cycle cycle , Collector<LaneNorm> out){
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());
laneNorm.setType("0");
//获取所有这个方向得车道 交通流量为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( new BigDecimal( collect.stream().mapToDouble(LaneNorm::getPassCapacity).sum()).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//交通流量求和
laneNorm.setTrafficCapacity(collect.stream().mapToDouble(LaneNorm::getTrafficCapacity).sum());
//剩余承载力求和
double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
laneNorm.setResidualCapacity(new BigDecimal(sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//有效绿灯时间 相位下此方向非右转得绿灯有效时间之和
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((double)sum1/sum);
Double d = bd.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d>1d?1d:d);
laneNorm.setSpotSpaceLength(sum1);
//发送方向整体信息
out.collect(laneNorm);
}
}
}
/** /**
* @param map 车道信息 * @param map 车道信息
* @param cycle 周期信息 * @param cycle 周期信息
...@@ -196,10 +268,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -196,10 +268,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//车道属性 分别是车道编码 方向 //车道属性 分别是车道编码 方向
String lane_id = String.valueOf(map.get("lane_id")); String lane_id = String.valueOf(map.get("lane_id"));
String positionName = String.valueOf(map.get("positionName")); String positionName = String.valueOf(map.get("positionName"));
Double trafficCapacity = getTrafficCapacity(cycle, car, lane_id); Double trafficCapacity = getTrafficCapacity(cycle, car, lane_id);
Integer lane_length = Integer.parseInt(String.valueOf(map.get("lane_length"))); Integer lane_length = Integer.parseInt(String.valueOf(map.get("lane_length")));
//车道基本信息 //车道基本信息
laneNorm.setOrder(0); laneNorm.setOrder(0);
laneNorm.setCorssId(cycle.getCrossCode()); laneNorm.setCorssId(cycle.getCrossCode());
...@@ -209,11 +279,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -209,11 +279,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setLaneName(positionName); laneNorm.setLaneName(positionName);
laneNorm.setLaneLength(lane_length); laneNorm.setLaneLength(lane_length);
laneNorm.setTrafficCapacity(trafficCapacity); laneNorm.setTrafficCapacity(trafficCapacity);
laneNorm.setType("2");
//如果交通流量为0，则不统计此车道指标数据 //如果交通流量为0，则不统计此车道指标数据
if (trafficCapacity==0d) { if (trafficCapacity==0d) {
return laneNorm; return laneNorm;
} }
//车道饱和度 //车道饱和度
int saturation_flow_rate = getSaturation(map); int saturation_flow_rate = getSaturation(map);
//车道有效绿灯时间 //车道有效绿灯时间
...@@ -230,20 +300,27 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -230,20 +300,27 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setMaxQueueLength(maxQueueLength); laneNorm.setMaxQueueLength(maxQueueLength);
laneNorm.setEffectGreenTime(effcetGreenTime[0]); laneNorm.setEffectGreenTime(effcetGreenTime[0]);
laneNorm.setPassCapacity(v); //通行能力
Double saturation = 100 * (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime)); BigDecimal bd2 = new BigDecimal(v);
Double d3 = bd2.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setPassCapacity(d3);
Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime));
BigDecimal bd1 = new BigDecimal(saturation); BigDecimal bd1 = new BigDecimal(saturation);
Double d2 = bd1.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue(); Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSaturation(d2); laneNorm.setSaturation(d2);
//剩余承载力 //剩余承载力
Double residualCapacity = (saturation_flow_rate * laneEffectGreenTime) - trafficCapacity; Double residualCapacity = (saturation_flow_rate * laneEffectGreenTime) - trafficCapacity;
BigDecimal bd = new BigDecimal(residualCapacity); BigDecimal bd = new BigDecimal(residualCapacity);
Double d = bd.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue(); Double d = bd.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setResidualCapacity(d<0?0.00:d); laneNorm.setResidualCapacity(d<0?0.00:d);
//计算空间占有率 //计算空间占有率
BigDecimal spot = new BigDecimal( 100 * maxQueueLength / lane_length); BigDecimal spot = new BigDecimal( (double)maxQueueLength / lane_length);
Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue(); Double d1 = spot.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d1+"%("+maxQueueLength+"m)"); laneNorm.setSpotSpaceShare(d1>1d?1d:d1);
//计算空间占有长度
laneNorm.setSpotSpaceLength(maxQueueLength);
return laneNorm; return laneNorm;
} }
...@@ -278,7 +355,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -278,7 +355,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
private static Double getTrafficCapacity(Cycle cycle, Map<String, List<TravelEvent>> laneCarMap, String laneId) { private static Double getTrafficCapacity(Cycle cycle, Map<String, List<TravelEvent>> laneCarMap, String laneId) {
//排队车辆 //排队车辆
int i = 0; int i = 0;
Long aLong = DateUtils.dateDiff(cycle.getBeginDateTime(), cycle.getEndDateTime()); Long aLong =(Long.parseLong(cycle.getEndDateTime()) - Long.parseLong(cycle.getBeginDateTime())) ;
try { try {
if (laneCarMap.containsKey(laneId)) { if (laneCarMap.containsKey(laneId)) {
i = laneCarMap.get(laneId).size(); i = laneCarMap.get(laneId).size();
...@@ -289,7 +366,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -289,7 +366,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (i == 0) { if (i == 0) {
return 0d; return 0d;
} }
BigDecimal spot = new BigDecimal(60 * 60 / aLong); BigDecimal spot = new BigDecimal((60 * 60) / (aLong/1000));
Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue(); Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
return d1; return d1;
} }
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java
...@@ -32,8 +32,8 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> { ...@@ -32,8 +32,8 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties"; String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path); connection = MySQLUtils.getConnection(path);
insertPstmt = connection.prepareStatement("insert into " + 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," + "app_cross_dimension_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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?)"); "spot_space_share,dimension_name,`order`,update_time,`type`,spot_space_length) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
} }
@Override @Override
...@@ -60,10 +60,12 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> { ...@@ -60,10 +60,12 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
insertPstmt.setDouble(8,value.getTrafficCapacity()); insertPstmt.setDouble(8,value.getTrafficCapacity());
insertPstmt.setDouble(9,value.getSaturation()); insertPstmt.setDouble(9,value.getSaturation());
insertPstmt.setDouble(10,value.getResidualCapacity()); insertPstmt.setDouble(10,value.getResidualCapacity());
insertPstmt.setString(11,value.getSpotSpaceShare()); insertPstmt.setDouble(11,value.getSpotSpaceShare());
insertPstmt.setString(12,value.getLaneName()); insertPstmt.setString(12,value.getLaneName());
insertPstmt.setInt(13,value.getOrder()); insertPstmt.setInt(13,value.getOrder());
insertPstmt.setString(14, DateUtils.getTodayTime()); insertPstmt.setString(14, DateUtils.getTodayTime());
insertPstmt.setString(15,value.getType());
insertPstmt.setInt(16,value.getSpotSpaceLength());
insertPstmt.execute(); insertPstmt.execute();
} }
} }
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DateUtils.java
...@@ -242,10 +242,8 @@ public class DateUtils { ...@@ -242,10 +242,8 @@ public class DateUtils {
*/ */
public static boolean compareDataIsBefore10Min(String s){ public static boolean compareDataIsBefore10Min(String s){
try { try {
String res;
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = new Date(); Date date = new Date();
Date afterDate = new Date(simpleDateFormat.parse(s).getTime() + 600000); Date afterDate = new Date(Long.parseLong(s) + 600000);
return date.compareTo(afterDate)>0?false:true; return date.compareTo(afterDate)>0?false:true;
}catch (Exception e){ }catch (Exception e){
......
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