Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 044bd9c9 by 黄准
Options

车道指标相关计算逻辑

parent f868d445
Showing with 116 additions and 76 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/LaneNorm.java
......@@ -39,5 +39,9 @@ public class LaneNorm extends Stage{
private Integer order;
//类型
private String type;
//周期开始时间
private String cycleStartTime;
//周期结束时间
private String cycleEndTime;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/process/LaneNormProcessFunction.java → realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/LaneNormProcessFunction.java
package com.zhht.irn.process;
package com.zhht.irn.functions;
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.common.state.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
......@@ -34,23 +33,30 @@ import java.util.stream.Stream;
public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm> {
// state 管理当前周期中存储的旅行事件
private MapState<String, List<TravelEvent>> cycleTravel;
// 临时
private MapState<String, List<TravelEvent>> cycleTravelTemp;
//存储mysql维表数据
private MapState<String, List<Map>> dict;
//
private ValueState<Boolean> processCycle;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
cycleTravel = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravel", String.class, Object.class));
cycleTravelTemp = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravelTemp", String.class, Object.class));
dict = getRuntimeContext().getMapState(new MapStateDescriptor("dict", String.class, Object.class));
processCycle = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("processCycle", Boolean.class));
}
@Override
public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
processCycle.update(true);
Connection connection = getConnection();
//初始化字典数据
if (dict.get("phaseDirection") == null) {
//查询相位方向信息
dict.put("phaseDirection", getDictResult(connection,"SELECT\n" +
dict.put("phaseDirection", getDictResult(connection, "SELECT\n" +
" a.ID id,\n" +
" a.Direction direction,\n" +
" b.Name traffic_flow_direction,\n" +
......@@ -66,18 +72,18 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
// 代表周期已到，需要进行相应的指标计算了
//周期内所有过车数据
List<TravelEvent> travelEvents = cycleTravel.get(f1.getCrossCode());
if(travelEvents==null){
travelEvents=new ArrayList<>();
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> 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" +
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,GROUP_CONCAT(name) as name 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() + "'");
//查询路口所有的方向
......@@ -97,32 +103,51 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
list.add(laneNorm);
//交通流量为0得不参与计算
if(laneNorm!=null &&laneNorm.getTrafficCapacity()!=0d){
if (laneNorm != null && laneNorm.getTrafficCapacity() != 0d) {
out.collect(laneNorm);
}
}
//获取所有方向的指标
getPositionNorm(list,positionList,cycle,out);
getPositionNorm(list, positionList, cycle, out);
//获取所有流向的指标
getPositionNameNorm(list,positionNameList,cycle,out);
//过滤过期数据 没有进入事件的数据保留到下次计算
cycleTravel.put(f1.getCrossCode(),car.stream().filter(recode->StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList()));
getPositionNameNorm(list, positionNameList, cycle, out);
List<TravelEvent> collect = car.stream().filter(recode -> StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList());
List<TravelEvent> travelEvents1 = cycleTravelTemp.get(cycle.getCrossCode())==null?new ArrayList<>():cycleTravelTemp.get(cycle.getCrossCode());
//插入计算过程中缓存的数据
for (TravelEvent travelEvent:travelEvents1){
collect.add(travelEvent);
}
cycleTravel.put(f1.getCrossCode(),collect);
cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>());
processCycle.update(false);
//关闭mysql连接
if(connection!=null){
if (connection != null) {
connection.close();
}
}
@Override
public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
if(processCycle.value()==null){
processCycle.update(false);
}
if (cycleTravel.get(event.getCrossId()) == null) {
cycleTravel.put(event.getCrossId(), new ArrayList<>());
}
if (cycleTravelTemp.get(event.getCrossId()) == null) {
cycleTravelTemp.put(event.getCrossId(), new ArrayList<>());
}
//周期数据处理时，讲数据塞入临时内存中
if (processCycle.value()) {
cycleTravelTemp.get(event.getCrossId()).add(event);
} else {
cycleTravel.get(event.getCrossId()).add(event);
}
}
private static String gettEffectGreenTimeForPosition(List<Stage> stageList, String position) {
int AllGreenTime=0;
int AllGreenTime = 0;
for (Stage stage : stageList) {
List<StageForDirection> stageForDirectionList = stage.getStageForDirectionList();
for (StageForDirection dir : stageForDirectionList) {
......@@ -133,7 +158,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
}
}
return AllGreenTime+"";
return AllGreenTime + "";
}
/**
......@@ -160,47 +185,50 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
/**
* 生成流向指标
*
* @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){
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.setLaneName(position + name);
laneNorm.setOrder(0);
laneNorm.setCycleOrder(cycle.getCycleOrder());
laneNorm.setDirection(position);
laneNorm.setCorssId(cycle.getCrossCode());
laneNorm.setType("1");
laneNorm.setCycleStartTime(DateUtils.stampToTime(cycle.getBeginDateTime()));
laneNorm.setCycleEndTime(DateUtils.stampToTime(cycle.getEndDateTime()));
//获取所有这个方向得车道 交通流量为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){
List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection()) && recode.getLaneName().indexOf(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.setMaxQueueLength(collect.stream().max(Comparator.comparingInt(LaneNorm::getMaxQueueLength)).get().getMaxQueueLength());
//饱和度取方向车道最大值
laneNorm.setSaturation(collect.stream().max(Comparator.comparingDouble(LaneNorm ::getSaturation)).get().getSaturation());
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.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));
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 sum = list.stream().filter(recode -> position.equals(recode.getDirection()) && recode.getLaneName().indexOf(name) != -1).mapToInt(LaneNorm::getLaneLength).sum();
int sum1 = collect.stream().mapToInt(LaneNorm::getMaxQueueLength).sum();
BigDecimal bd = new BigDecimal((double)sum1/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.setSpotSpaceShare(d > 1d ? 1d : d);
laneNorm.setSpotSpaceLength(sum1);
//发送方向整体信息
out.collect(laneNorm);
......@@ -210,14 +238,15 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
/**
* 生成方向指标
*
* @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){
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("整体");
......@@ -226,29 +255,31 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setDirection(position);
laneNorm.setCorssId(cycle.getCrossCode());
laneNorm.setType("0");
laneNorm.setCycleStartTime(DateUtils.stampToTime(cycle.getBeginDateTime()));
laneNorm.setCycleEndTime(DateUtils.stampToTime(cycle.getEndDateTime()));
//获取所有这个方向得车道 交通流量为0得不参与计算
List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())&&recode.getTrafficCapacity()!=0d).collect(Collectors.toList());
if(collect.size()==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.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.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));
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);
BigDecimal bd = new BigDecimal((double) sum1 / sum);
Double d = bd.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d>1d?1d:d);
laneNorm.setSpotSpaceShare(d > 1d ? 1d : d);
laneNorm.setSpotSpaceLength(sum1);
//发送方向整体信息
out.collect(laneNorm);
......@@ -268,6 +299,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//车道属性 分别是车道编码 方向
String lane_id = String.valueOf(map.get("lane_id"));
String positionName = String.valueOf(map.get("positionName"));
String name = String.valueOf(map.get("name"));
Double trafficCapacity = getTrafficCapacity(cycle, car, lane_id);
Integer lane_length = Integer.parseInt(String.valueOf(map.get("lane_length")));
//车道基本信息
......@@ -276,12 +308,14 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setDirection(String.valueOf(map.get("position")));
laneNorm.setLaneId(lane_id);
laneNorm.setCycleOrder(cycle.getCycleOrder());
laneNorm.setLaneName(positionName);
laneNorm.setLaneName(lane_id+"-"+name);
laneNorm.setLaneLength(lane_length);
laneNorm.setTrafficCapacity(trafficCapacity);
laneNorm.setType("2");
laneNorm.setCycleStartTime(DateUtils.stampToTime(cycle.getBeginDateTime()));
laneNorm.setCycleEndTime(DateUtils.stampToTime(cycle.getEndDateTime()));
//如果交通流量为0，则不统计此车道指标数据
if (trafficCapacity==0d) {
if (trafficCapacity == 0d) {
return laneNorm;
}
//车道饱和度
......@@ -292,8 +326,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
double laneEffectGreenTime = Double.valueOf(effcetGreenTime[1]) / cycle.getDuration();
//最大排队长度
Integer maxQueueLength = getMaxQueueLength(car, lane_id);
if(maxQueueLength>lane_length){
maxQueueLength=lane_length;
if (maxQueueLength > lane_length) {
maxQueueLength = lane_length;
}
double v = saturation_flow_rate * laneEffectGreenTime;
//车道级别指标计算
......@@ -307,18 +341,18 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime));
BigDecimal bd1 = new BigDecimal(saturation);
BigDecimal bd1 = new BigDecimal(laneEffectGreenTime==0d?0:saturation);
Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSaturation(d2);
//剩余承载力
Double residualCapacity = (saturation_flow_rate * laneEffectGreenTime) - trafficCapacity;
BigDecimal bd = new BigDecimal(residualCapacity);
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( (double)maxQueueLength / lane_length);
BigDecimal spot = new BigDecimal((double) maxQueueLength / lane_length);
Double d1 = spot.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSpotSpaceShare(d1>1d?1d:d1);
laneNorm.setSpotSpaceShare(d1 > 1d ? 1d : d1);
//计算空间占有长度
laneNorm.setSpotSpaceLength(maxQueueLength);
return laneNorm;
......@@ -355,7 +389,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
private static Double getTrafficCapacity(Cycle cycle, Map<String, List<TravelEvent>> laneCarMap, String laneId) {
//排队车辆
int i = 0;
Long aLong =(Long.parseLong(cycle.getEndDateTime()) - Long.parseLong(cycle.getBeginDateTime())) ;
Long aLong = (Long.parseLong(cycle.getEndDateTime()) - Long.parseLong(cycle.getBeginDateTime()));
try {
if (laneCarMap.containsKey(laneId)) {
i = laneCarMap.get(laneId).size();
......@@ -366,7 +400,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (i == 0) {
return 0d;
}
BigDecimal spot = new BigDecimal((60 * 60) / (aLong/1000));
BigDecimal spot = new BigDecimal((60 * 60) / (aLong / 1000));
Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
return d1;
}
......@@ -378,7 +412,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
*/
private static String[] gettEffectGreenTime(List<Stage> stageList, String laneId, List<Map> laneList) {
String allGreenTime = "";
int AllGreenTime=0;
int AllGreenTime = 0;
//从字典中获取当前车道所有流向信息
List<Map> currentLane = laneList.stream().filter(a -> laneId.equals(String.valueOf(a.get("lane_id")))).collect(Collectors.toList());
//获取周期的相位信息
......@@ -395,12 +429,12 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
}
}
if(greeTime!=0){
if (greeTime != 0) {
allGreenTime = allGreenTime + greeTime + ",";
}
}
return allGreenTime.length()>1?new String[]{allGreenTime.substring(0,allGreenTime.length()-1),AllGreenTime+""}:new String[]{"0","0"};
return allGreenTime.length() > 1 ? new String[]{allGreenTime.substring(0, allGreenTime.length() - 1), AllGreenTime + ""} : new String[]{"0", "0"};
}
/**
......@@ -421,12 +455,12 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
return (i * 6);
}
private static Connection getConnection()throws Exception{
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 {
private static List<Map> getDictResult(Connection connection, String sql) throws Exception {
Statement statement = connection.createStatement();
ResultSet resultset = statement.executeQuery(sql);
List list = resultSetToList(resultset);
......@@ -457,28 +491,28 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
List<TravelEvent> lists = new ArrayList<TravelEvent>();
Map<String, TravelEvent> map = new HashMap<String, TravelEvent>();
for (TravelEvent event : cars) {
if(!DateUtils.compareDataIsBefore10Min(event.getEventTime())){
if (!DateUtils.compareDataIsBefore10Min(event.getEventTime())) {
continue;
}
if (!map.containsKey(event.getRecordId())) {
map.put(event.getRecordId(),event);
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);
map.put(event.getRecordId(), travelEvent);
}
//进入路口时间 数据更新
if (event.getEventType() == EventType.INCROSS) {
travelEvent.setArrayTime(travelEvent.getEventTime());
map.put(event.getRecordId(),travelEvent);
travelEvent.setInCorssTime(travelEvent.getEventTime());
map.put(event.getRecordId(), travelEvent);
}
//丢失事件也确认为进入路口
if(event.getEventType() == EventType.LAST){
travelEvent.setArrayTime(travelEvent.getEventTime());
map.put(event.getRecordId(),travelEvent);
if (event.getEventType() == EventType.LAST) {
travelEvent.setInCorssTime(travelEvent.getEventTime());
map.put(event.getRecordId(), travelEvent);
}
}
for (String key : map.keySet()) {
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java
......@@ -3,19 +3,16 @@ 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.process.LaneNormProcessFunction;
import com.zhht.irn.functions.LaneNormProcessFunction;
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.FlinkUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import java.time.Duration;
......@@ -30,6 +27,8 @@ public class LaneNormJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());
DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
......@@ -76,10 +75,11 @@ public class LaneNormJob {
//根据corssid 分组 保证同一路口的数据进行一个进程
DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy).keyBy(cycleKeySelect);
DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).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());
env.execute();
env.execute("laneNormJob");
}
/*获取信控指标 并计算相关属性*/
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java
......@@ -2,6 +2,7 @@ package com.zhht.irn.sink;
import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
......@@ -29,11 +30,10 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path);
connection = DorisUtils.getConnection();
insertPstmt = connection.prepareStatement("insert into " +
"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,dimension_name,`order`,update_time,`type`,spot_space_length) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
"app_cross_dimension_norm(cross_id,lane_id,direction,cycle_order,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," +
"spot_space_share,dimension_name,`order`,statistic_time,`type`,spot_space_length,cycle_start_time,cycle_end_time,record_date) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
}
@Override
......@@ -66,6 +66,9 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
insertPstmt.setString(14, DateUtils.getTodayTime());
insertPstmt.setString(15,value.getType());
insertPstmt.setInt(16,value.getSpotSpaceLength());
insertPstmt.setString(17,value.getCycleStartTime());
insertPstmt.setString(18,value.getCycleEndTime());
insertPstmt.setString(19,DateUtils.getTodayTime());
insertPstmt.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DateUtils.java
......@@ -261,7 +261,6 @@ public class DateUtils {
try {
Date date = new Date();
Date afterDate = new Date(Long.parseLong(s) + 600000);
return date.compareTo(afterDate)>0?false:true;
}catch (Exception e){
return false;
......
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