Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit f18a37d2 by 黄准
Options

车道指标实时任务修改

parent c5b088a5
Showing with 96 additions and 21 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/LaneNormProcessFunction.java
......@@ -5,6 +5,7 @@ import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.utils.*;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
......@@ -42,6 +43,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
private Connection connection;
//每个路口允许的最大时间数据
private static Long maxTravel=50000L;
//每天的cycleOrder 不允许重复
private MapState<String,String> dayCycle;
@Override
public void open(Configuration parameters) throws Exception {
......@@ -51,6 +54,16 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
dict = getRuntimeContext().getMapState(new MapStateDescriptor("dict", String.class, Object.class));
processCycle = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("processCycle", Boolean.class));
connection=DorisUtils.getConnection();
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.hours(3))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor("dayCycle", String.class, String.class);
//
mapStateDescriptor.enableTimeToLive(ttlConfig);
dayCycle = getRuntimeContext().getMapState(mapStateDescriptor);
}
@Override
......@@ -65,6 +78,12 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
processCycle.update(true);
List<LaneNorm> laneNorms = new ArrayList<>();
if(!dayCycle.contains(DateUtils.toDateStr(new Date())+"-"+cycle.getCrossCode()+"-"+cycle.getCycleOrder())){
dayCycle.put(DateUtils.toDateStr(new Date())+"-"+cycle.getCrossCode()+"-"+cycle.getCycleOrder(),"true");
}else{
//如果内存中已经存在重复的周期 则不进行计算
return;
}
//初始化字典数据
if (dict.get("phaseDirection") == null || dict.get("laneList") == null || dict.get("dictResult") == null || dict.get("positionList") == null || dict.get("positionNameList") == null) {
if(connection==null){
......@@ -131,7 +150,7 @@ 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) {
laneNorms.add(laneNorm);
}
}
......@@ -268,8 +287,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
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(name) != -1 && recode.getTrafficCapacity() != 0d).collect(Collectors.toList());
//获取所有这个方向得车道
List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())&&recode.getLaneName().indexOf(name) != -1).collect(Collectors.toList());
if (collect.size() == 0) {
continue;
}
......@@ -282,8 +301,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//交通流量求和
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());
double sum2 = collect.stream().reduce(0.0,(x,y)->x+(y.getPassCapacity()-y.getTrafficCapacity()),Double::sum);
//double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
laneNorm.setResidualCapacity(new BigDecimal(sum2<=0?0d:sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//laneNorm.setResidualCapacity(new BigDecimal(sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//有效绿灯时间 相位下此方向非右转的绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForDirction(cycle.getStageList(), position,name));
//识别空间空间占有率 车道长度需要交通流量为0得数据
......@@ -320,8 +341,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
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());
//获取所有这个方向得车道
List<LaneNorm> collect = list.stream().filter(recode -> position.equals(recode.getDirection())).collect(Collectors.toList());
if (collect.size() == 0) {
continue;
}
......@@ -333,8 +354,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//交通流量求和
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());
double sum2 = collect.stream().reduce(0.0,(x,y)->x+(y.getPassCapacity()-y.getTrafficCapacity()),Double::sum);
//double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
laneNorm.setResidualCapacity(new BigDecimal(sum2<=0?0d:sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//有效绿灯时间 相位下此方向非右转得绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForPosition(cycle.getStageList(), position));
//识别空间空间占有率 车道长度需要交通流量为0得数据
......@@ -350,6 +372,20 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
}
public static void main(String[] args) {
List<LaneNorm> objects = new ArrayList<LaneNorm>();
LaneNorm laneNorm = new LaneNorm();
laneNorm.setTrafficCapacity(100d);
laneNorm.setPassCapacity(50d);
LaneNorm laneNorm1 = new LaneNorm();
laneNorm1.setTrafficCapacity(100d);
laneNorm1.setPassCapacity(130d);
objects.add(laneNorm);
objects.add(laneNorm1);
Double reduce = objects.stream().reduce(0.0, (x, y) -> x + (y.getPassCapacity() - y.getTrafficCapacity()), Double::sum);
System.out.println(reduce);
}
/**
* @param map 车道信息
* @param cycle 周期信息
......@@ -378,10 +414,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setType("2");
laneNorm.setCycleStartTime(DateUtils.stampToTime(cycle.getBeginDateTime()));
laneNorm.setCycleEndTime(DateUtils.stampToTime(cycle.getEndDateTime()));
//如果交通流量为0，则不统计此车道指标数据
/*//如果交通流量为0，则不统计此车道指标数据
//2.0版本 参与计算
if (trafficCapacity == 0d) {
return laneNorm;
}
}*/
//车道饱和度
int saturation_flow_rate = getSaturation(map);
//车道有效绿灯时间
......@@ -397,12 +434,12 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setMaxQueueLength(maxQueueLength);
laneNorm.setEffectGreenTime(effcetGreenTime[0]);
//通行能力
double v = saturation_flow_rate * laneEffectGreenTime;
double v = saturation_flow_rate * laneEffectGreenTime>1d?1: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));
Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime>1d?1:laneEffectGreenTime));
BigDecimal bd1 = new BigDecimal(laneEffectGreenTime==0d?0:saturation);
Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setSaturation(d2);
......@@ -478,6 +515,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
int AllGreenTime = 0;
//从字典中获取当前车道所有流向信息
List<Map> currentLane = laneList.stream().filter(a -> laneId.equals(String.valueOf(a.get("lane_id")))).collect(Collectors.toList());
//已经参与计算的相位，不累计计算
Map<Object, Object> HasStage = new HashMap<>();
//获取周期的相位信息
for (Map map : currentLane) {
int greeTime = 0;
......@@ -487,8 +526,12 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
List<StageForDirection> stageForDirectionList = stage.getStageForDirectionList();
for (StageForDirection dir : stageForDirectionList) {
if (position.equals(dir.getDirection()) && name.equals(dir.getTraffic_flow_direction())) {
//已经参与计算的相位，不参与累积绿灯时间计算
if(!HasStage.containsKey(stage.getPhaseValue())){
AllGreenTime += stage.getValidGreen();
HasStage.put(stage.getPhaseValue(),"1");
}
greeTime += stage.getValidGreen();
AllGreenTime += stage.getValidGreen();
}
}
}
......@@ -522,7 +565,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
maxQueneLength=queneLength;
}
}
//旧版逻辑 根据有到达时间 没有进入时间的车辆取 最大排队长度
//todo 旧版逻辑根据有到达时间 没有进入时间的车辆取 最大排队长度
/*i = laneCarMap.get(laneId).stream().
filter(a -> StringUtils.isNotEmpty(a.getArrayTime()) &&
StringUtils.isEmpty(a.getInCorssTime())).collect(Collectors.toList()).size();*/
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java
......@@ -33,8 +33,8 @@ public class LaneNormJob {
//env.setRestartStrategy(RestartStrategies.noRestart());
//设置重启策略
env.setRestartStrategy( RestartStrategies.fixedDelayRestart(10, Time.seconds(10)));
DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"signal_cycle_data","laneNormJob1");
DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"trips_event_info","laneNormJob1");
KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
@Override
public String getKey(Cycle cycle) throws Exception {
......@@ -52,14 +52,14 @@ public class LaneNormJob {
public Cycle map(String cycle) throws Exception {
Cycle cycle1 = JSON.parseObject(cycle, Cycle.class);
cycle1 = getCycle(cycle1);
System.out.println("周期数据->"+cycle);
//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);
//System.out.println("旅行事件数据->"+travelEvent);
return JSON.parseObject(travelEvent, TravelEvent.class);
}
};
......@@ -82,7 +82,7 @@ public class LaneNormJob {
ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
SingleOutputStreamOperator<List<LaneNorm>> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
process.addSink(new LaneNormSink()).name("laneNormJobSink");
process.addSink(new LaneNormSink()).name(" ");
env.execute("laneNormJob");
}catch (Exception e){
System.out.println("程序异常关闭->"+e);
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
......@@ -27,7 +27,7 @@ public class FlinkUtils {
private static final Logger logger = LoggerFactory.getLogger(FlinkUtils.class);
public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
private static String defaultBootstrapServers;
private static String defaultInputTopics;
......@@ -159,4 +159,36 @@ public class FlinkUtils {
DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
return rawStringStream;
}
public static DataStream<String> createKafkaStreamHuangzhun(String[] args, StreamExecutionEnvironment environment, String topics, String groupId) {
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String servers = kafkaTool.get("bootstrap.servers", defaultBootstrapServers);
String topics2 = kafkaTool.get("kafka.input.topics", topics);
String groupId2 = kafkaTool.get("group.id", groupId);
String autoOffsetReset = kafkaTool.get("auto.offset.reset", defaultAutoOffsetReset);
String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
String enableAutoCommit = kafkaTool.get("enable.auto.commit", "false");
return buildKafkaDataStreamHuangzhun(environment, servers, topics2, groupId2, autoOffsetReset, checkpointPath, checkpointInterval, enableAutoCommit);
}
private static DataStream<String> buildKafkaDataStreamHuangzhun(StreamExecutionEnvironment env, String servers, String topics, String groupId,
String autoOffsetReset, String checkpointPath, long checkpointInterval, String enableAutoCommit) {
List<String> topicList = Arrays.asList(topics.split(","));
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", enableAutoCommit);
properties.setProperty("auto.offset.reset", autoOffsetReset);
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
// 基于Kafka 原始JSON格式的字符串，获取消息流
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
kafkaConsumer.setStartFromLatest();
DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
return rawStringStream;
}
}
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