Skip to content

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

Merge remote-tracking branch 'origin/master'

parents 0867941d 2930fc23
Showing with 265 additions and 139 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/LaneNormProcessFunction.java
package com.zhht.irn.functions; package com.zhht.irn.functions;
import com.alibaba.druid.pool.DruidPooledConnection;
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.utils.DateUtils; import com.zhht.irn.utils.*;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.MySQLUtils;
import com.zhht.irn.utils.StringUtils;
import org.apache.flink.api.common.state.*; import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
...@@ -31,7 +29,7 @@ import java.util.stream.Stream; ...@@ -31,7 +29,7 @@ import java.util.stream.Stream;
* @retrun: * @retrun:
* @Creat :2022-11-04-9:53 * @Creat :2022-11-04-9:53
**/ **/
public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm> { public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>> {
// state 管理当前周期中存储的旅行事件 // state 管理当前周期中存储的旅行事件
private MapState<String, List<TravelEvent>> cycleTravel; private MapState<String, List<TravelEvent>> cycleTravel;
// 临时 // 临时
...@@ -41,6 +39,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -41,6 +39,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
// //
private ValueState<Boolean> processCycle; private ValueState<Boolean> processCycle;
private Connection connection;
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
...@@ -48,14 +48,26 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -48,14 +48,26 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
cycleTravelTemp = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravelTemp", String.class, Object.class)); cycleTravelTemp = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravelTemp", String.class, Object.class));
dict = getRuntimeContext().getMapState(new MapStateDescriptor("dict", String.class, Object.class)); dict = getRuntimeContext().getMapState(new MapStateDescriptor("dict", String.class, Object.class));
processCycle = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("processCycle", Boolean.class)); processCycle = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("processCycle", Boolean.class));
connection=DorisUtils.getConnection();
} }
@Override @Override
public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception { public void close() throws Exception {
super.close();
if(connection!=null){
connection.close();
}
}
@Override
public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
processCycle.update(true); processCycle.update(true);
Connection connection = getConnection(); List<LaneNorm> laneNorms = new ArrayList<>();
//初始化字典数据 //初始化字典数据
if (dict.get("phaseDirection") == null) { if (dict.get("phaseDirection") == null || dict.get("laneList") == null || dict.get("dictResult") == null || dict.get("positionList") == null || dict.get("positionNameList") == null) {
if(connection==null){
connection=DorisUtils.getConnection();
}
//查询相位方向信息 //查询相位方向信息
dict.put("phaseDirection", getDictResult(connection, "SELECT\n" + dict.put("phaseDirection", getDictResult(connection, "SELECT\n" +
" a.ID id,\n" + " a.ID id,\n" +
...@@ -63,11 +75,32 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -63,11 +75,32 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
" b.Name traffic_flow_direction,\n" + " b.Name traffic_flow_direction,\n" +
" c.PID phase_no,\n" + " c.PID phase_no,\n" +
" c.PName phase_name\n" + " c.PName phase_name\n" +
"FROM tb_phase_movement a\n" + "FROM dim_phase_movement a\n" +
"LEFT JOIN trafficflowdirection b\n" + "LEFT JOIN dim_traffic_flow_direction b\n" +
" on a.TrafficFlowId = b.ID\n" + " on a.TrafficFlowId = b.ID\n" +
"LEFT JOIN dim_dic_phase c\n" + "LEFT JOIN dim_dic_phase c\n" +
" on a.PhaseId = c.ID")); " on a.PhaseId = c.ID"));
//车道道路流向关系
dict.put("laneList", getDictResult(connection, "SELECT cross_id,lane_id,turn_direction as name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in'"));
//获取所有车道信息
dict.put("dictResult", getDictResult(connection, "select a.*,b.saturation_flow_rate from (\n" +
" SELECT cross_id,lane_id,position,lane_length,GROUP_CONCAT(turn_direction) as name FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in' group By lane_id,cross_id,position,lane_length,cross_id order by cross_id,lane_id)a left join app_atomic_lane_saturation_flow_rate b\n" +
" on a.lane_id=b.lane_id and a.cross_id=b.cross_id\t\t" +
" where 1=1\n" +
" and b.inout_name='in' order by b.lane_id asc"));
//查询路口所有的方向
dict.put("positionList", getDictResult(connection, "SELECT cross_id,position FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in' GROUP BY position,cross_id"));
//查询路口所有的流向
dict.put("positionNameList", getDictResult(connection, " SELECT position,turn_direction as name,cross_id FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in' GROUP BY position,turn_direction,cross_id"));
long currentProcessingTime = ctx.timerService().currentProcessingTime();
//12小时触发一次
ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+12*60*60*1000);
//ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+1*60*1000);
} }
Cycle f1 = cycle; Cycle f1 = cycle;
// 代表周期已到，需要进行相应的指标计算了 // 代表周期已到，需要进行相应的指标计算了
...@@ -76,23 +109,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -76,23 +109,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (travelEvents == null) { if (travelEvents == null) {
travelEvents = new ArrayList<>(); travelEvents = new ArrayList<>();
} }
List<TravelEvent> car = getCar(travelEvents); List<TravelEvent> car = getCar(travelEvents,cycle);
//根据车道分割过车数据 //根据车道分割过车数据
Map<String, List<TravelEvent>> carForLane = getCarForLane(car); 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,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() + "'");
//查询路口所有的方向
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<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) {
...@@ -100,18 +119,25 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -100,18 +119,25 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} }
//获取所有车道的指标信息 //获取所有车道的指标信息
List<LaneNorm> list = new ArrayList<LaneNorm>(); List<LaneNorm> list = new ArrayList<LaneNorm>();
List<Map> laneList = dict.get("laneList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> dictResult = dict.get("dictResult").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> positionList = dict.get("positionList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> positionNameList = dict.get("positionNameList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
for (Map map : dictResult) { for (Map map : dictResult) {
LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList); LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
list.add(laneNorm); list.add(laneNorm);
//交通流量为0得不参与计算 //交通流量为0得不参与计算
if (laneNorm != null && laneNorm.getTrafficCapacity() != 0d) { if (laneNorm != null && laneNorm.getTrafficCapacity() != 0d) {
out.collect(laneNorm); laneNorms.add(laneNorm);
} }
} }
//获取所有方向的指标 //获取所有方向的指标
getPositionNorm(list, positionList, cycle, out); getPositionNorm(list, positionList, cycle, laneNorms);
//获取所有流向的指标 //获取所有流向的指标
getPositionNameNorm(list, positionNameList, cycle, out); getPositionNameNorm(list, positionNameList, cycle, laneNorms);
if(laneNorms.size()>0){
out.collect(laneNorms);
}
List<TravelEvent> collect = car.stream().filter(recode -> StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList()); 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()); List<TravelEvent> travelEvents1 = cycleTravelTemp.get(cycle.getCrossCode())==null?new ArrayList<>():cycleTravelTemp.get(cycle.getCrossCode());
//插入计算过程中缓存的数据 //插入计算过程中缓存的数据
...@@ -121,14 +147,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -121,14 +147,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
cycleTravel.put(f1.getCrossCode(),collect); cycleTravel.put(f1.getCrossCode(),collect);
cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>()); cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>());
processCycle.update(false); processCycle.update(false);
//关闭mysql连接
if (connection != null) {
connection.close();
}
} }
@Override @Override
public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception { public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
if(processCycle.value()==null){ if(processCycle.value()==null){
processCycle.update(false); processCycle.update(false);
} }
...@@ -147,6 +169,19 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -147,6 +169,19 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} }
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<LaneNorm>> out) throws Exception {
super.onTimer(timestamp, ctx, out);
System.out.println("触发定时器了。。。。");
//通过定时器清空字典表
dict.remove("phaseDirection");
dict.remove("laneList");
dict.remove("dictResult");
dict.remove("positionList");
dict.remove("positionNameList");
}
private static String gettEffectGreenTimeForPosition(List<Stage> stageList, String position) { private static String gettEffectGreenTimeForPosition(List<Stage> stageList, String position) {
int AllGreenTime = 0; int AllGreenTime = 0;
for (Stage stage : stageList) { for (Stage stage : stageList) {
...@@ -190,9 +225,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -190,9 +225,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
* @param list * @param list
* @param positionNameList * @param positionNameList
* @param cycle * @param cycle
* @param out * @param collectList
*/ */
private static void getPositionNameNorm(List<LaneNorm> list, List<Map> positionNameList, Cycle cycle, Collector<LaneNorm> out) { private static void getPositionNameNorm(List<LaneNorm> list, List<Map> positionNameList, Cycle cycle, List<LaneNorm> collectList) {
if (list.size() > 0) { if (list.size() > 0) {
for (Map map : positionNameList) { for (Map map : positionNameList) {
String position = String.valueOf(map.get("position")); String position = String.valueOf(map.get("position"));
...@@ -232,7 +267,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -232,7 +267,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setSpotSpaceShare(d > 1d ? 1d : d); laneNorm.setSpotSpaceShare(d > 1d ? 1d : d);
laneNorm.setSpotSpaceLength(sum1); laneNorm.setSpotSpaceLength(sum1);
//发送方向整体信息 //发送方向整体信息
out.collect(laneNorm); collectList.add(laneNorm);
} }
} }
} }
...@@ -243,9 +278,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -243,9 +278,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
* @param list * @param list
* @param positionList * @param positionList
* @param cycle * @param cycle
* @param out * @param collectList
*/ */
private static void getPositionNorm(List<LaneNorm> list, List<Map> positionList, Cycle cycle, Collector<LaneNorm> out) { private static void getPositionNorm(List<LaneNorm> list, List<Map> positionList, Cycle cycle, List<LaneNorm> collectList) {
if (list.size() > 0) { if (list.size() > 0) {
for (Map map : positionList) { for (Map map : positionList) {
String position = String.valueOf(map.get("position")); String position = String.valueOf(map.get("position"));
...@@ -283,7 +318,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -283,7 +318,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
laneNorm.setSpotSpaceShare(d > 1d ? 1d : d); laneNorm.setSpotSpaceShare(d > 1d ? 1d : d);
laneNorm.setSpotSpaceLength(sum1); laneNorm.setSpotSpaceLength(sum1);
//发送方向整体信息 //发送方向整体信息
out.collect(laneNorm); collectList.add(laneNorm);
} }
} }
} }
...@@ -299,9 +334,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -299,9 +334,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
LaneNorm laneNorm = new LaneNorm(); LaneNorm laneNorm = new LaneNorm();
//车道属性 分别是车道编码 方向 //车道属性 分别是车道编码 方向
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 name = String.valueOf(map.get("name")); String name = String.valueOf(map.get("name"));
//计算交通流量
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);
...@@ -325,22 +361,20 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -325,22 +361,20 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
String[] effcetGreenTime = gettEffectGreenTime(cycle.getStageList(), lane_id, laneList); String[] effcetGreenTime = gettEffectGreenTime(cycle.getStageList(), lane_id, laneList);
//车道绿信比 //车道绿信比
double laneEffectGreenTime = Double.valueOf(effcetGreenTime[1]) / cycle.getDuration(); double laneEffectGreenTime = Double.valueOf(effcetGreenTime[1]) / cycle.getDuration();
//车道级别指标计算
//最大排队长度 //最大排队长度
Integer maxQueueLength = getMaxQueueLength(car, lane_id); Integer maxQueueLength = getMaxQueueLength(cycle,car, lane_id);
if (maxQueueLength > lane_length) { if (maxQueueLength > lane_length) {
maxQueueLength = lane_length; maxQueueLength = lane_length;
} }
double v = saturation_flow_rate * laneEffectGreenTime;
//车道级别指标计算
laneNorm.setMaxQueueLength(maxQueueLength); laneNorm.setMaxQueueLength(maxQueueLength);
laneNorm.setEffectGreenTime(effcetGreenTime[0]); laneNorm.setEffectGreenTime(effcetGreenTime[0]);
//通行能力 //通行能力
double v = saturation_flow_rate * laneEffectGreenTime;
BigDecimal bd2 = new BigDecimal(v); BigDecimal bd2 = new BigDecimal(v);
Double d3 = bd2.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue(); Double d3 = bd2.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setPassCapacity(d3); laneNorm.setPassCapacity(d3);
//饱和流率
Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime)); Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime));
BigDecimal bd1 = new BigDecimal(laneEffectGreenTime==0d?0:saturation); BigDecimal bd1 = new BigDecimal(laneEffectGreenTime==0d?0:saturation);
Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue(); Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
...@@ -380,8 +414,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -380,8 +414,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} }
/** /**
* 计算通行能力 * 计算交通流量
* * 口径为 周期开始时间至周期结束时间内进入的车数量 单位为辆/h
* @param cycle * @param cycle
* @param laneCarMap * @param laneCarMap
* @param laneId * @param laneId
...@@ -398,10 +432,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -398,10 +432,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} catch (Exception e) { } catch (Exception e) {
} }
if (i == 0) {
if (i==0 || aLong == 0) {
return 0d; return 0d;
} }
BigDecimal spot = new BigDecimal((60 * 60) / (aLong / 1000)); BigDecimal spot = new BigDecimal(i * (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;
} }
...@@ -441,23 +476,40 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -441,23 +476,40 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
/** /**
* 根据过车查询车道的最大长度 * 根据过车查询车道的最大长度
*/ */
private static Integer getMaxQueueLength(Map<String, List<TravelEvent>> laneCarMap, String laneId) { private static Integer getMaxQueueLength(Cycle cycle,Map<String, List<TravelEvent>> laneCarMap, String laneId) {
//排队车辆 //排队车辆
int i = 0; long cycleStart = Long.parseLong(cycle.getBeginDateTime());
long cycleEnd = Long.parseLong(cycle.getEndDateTime());
int maxQueneLength = 0;
try { try {
if (laneCarMap.containsKey(laneId)) { if (laneCarMap.containsKey(laneId)) {
i = laneCarMap.get(laneId).stream(). //获取这个车道所有的 过车数据
List<TravelEvent> travelEvents = laneCarMap.get(laneId);
//去周期内 每一秒的排队长度 取最大值
for(Long start = cycleStart; start <= cycleEnd; start += 1000) {
long start1 = start;
//通过每一秒分析出该车道停留的车数量
int queneLength = travelEvents.stream().filter(a-> Long.parseLong(a.getArrayTime())<=start1
&& (StringUtils.isEmpty(a.getInCorssTime()) || start1<Long.parseLong(a.getInCorssTime()))).collect(Collectors.toList()).size();
if(maxQueneLength<queneLength){
maxQueneLength=queneLength;
}
}
//旧版逻辑 根据有到达时间 没有进入时间的车辆取 最大排队长度
/*i = laneCarMap.get(laneId).stream().
filter(a -> StringUtils.isNotEmpty(a.getArrayTime()) && filter(a -> StringUtils.isNotEmpty(a.getArrayTime()) &&
StringUtils.isEmpty(a.getInCorssTime())).collect(Collectors.toList()).size(); StringUtils.isEmpty(a.getInCorssTime())).collect(Collectors.toList()).size();*/
} }
} catch (Exception e) { } catch (Exception e) {
} }
return (i * 6); return (maxQueneLength * 6);
} }
private static Connection getConnection() throws Exception { private static DruidPooledConnection getConnection() throws Exception {
Connection connection = DorisUtils.getConnection(); //修改为从连接池中取连接
DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
//Connection connection = DorisUtils.getConnection();
return connection; return connection;
} }
...@@ -488,10 +540,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -488,10 +540,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} }
//处理过车数据 将其合并 //处理过车数据 将其合并
private static List<TravelEvent> getCar(List<TravelEvent> cars) { private static List<TravelEvent> getCar(List<TravelEvent> cars,Cycle cycle) {
List<TravelEvent> lists = new ArrayList<TravelEvent>(); List<TravelEvent> lists = new ArrayList<TravelEvent>();
Map<String, TravelEvent> map = new HashMap<String, TravelEvent>(); Map<String, TravelEvent> map = new HashMap<String, TravelEvent>();
for (TravelEvent event : cars) { for (TravelEvent event : cars) {
//十分钟之前的数据 直接废弃
if (!DateUtils.compareDataIsBefore10Min(event.getEventTime())) { if (!DateUtils.compareDataIsBefore10Min(event.getEventTime())) {
continue; continue;
} }
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelEventAndCycleCoProcessFunction.java
...@@ -87,16 +87,16 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav ...@@ -87,16 +87,16 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
carRecordState = getRuntimeContext().getMapState(carRecordStateDescriptor); carRecordState = getRuntimeContext().getMapState(carRecordStateDescriptor);
dataSource = DruidConnectPoolUtils.getDataSource connection = DruidConnectPoolUtils.getConnection();
// ("jdbc:mysql://localhost:9030/demo?characterEncoding=utf8", // ("jdbc:mysql://localhost:9030/demo?characterEncoding=utf8",
// ("jdbc:mysql://localhost:3307/test?characterEncoding=utf8", // ("jdbc:mysql://localhost:3307/test?characterEncoding=utf8",
// ("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8", // ("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
("jdbc:mysql://10.243.0.22:9030/demo?characterEncoding=utf8", /* ("jdbc:mysql://10.243.0.22:9030/demo?characterEncoding=utf8",
"root", "root",
// "root123"); // "root123");
"mima"); "mima");
*/
connection = dataSource.getConnection(); // dataSource.getConnection();
dim_cnt_cross_lane_position = getLaneDimData(connection); dim_cnt_cross_lane_position = getLaneDimData(connection);
} }
...@@ -754,7 +754,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav ...@@ -754,7 +754,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
*/ */
private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException { private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
DruidPooledConnection connection = dataSource.getConnection(); DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
String sql = "INSERT INTO demo.app_cross_line_travel_car_info" + String sql = "INSERT INTO demo.app_cross_line_travel_car_info" +
"(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," + "(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java
...@@ -5,6 +5,7 @@ import com.zhht.irn.entity.*; ...@@ -5,6 +5,7 @@ import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType; import com.zhht.irn.enums.EventType;
import com.zhht.irn.functions.LaneNormProcessFunction; import com.zhht.irn.functions.LaneNormProcessFunction;
import com.zhht.irn.sink.LaneNormSink; import com.zhht.irn.sink.LaneNormSink;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import com.zhht.irn.utils.FlinkUtils; import com.zhht.irn.utils.FlinkUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.FilterFunction;
...@@ -26,60 +27,64 @@ public class LaneNormJob { ...@@ -26,60 +27,64 @@ public class LaneNormJob {
public static void main(String[] args) throws Exception { public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart()); env.setRestartStrategy(RestartStrategies.noRestart());
DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob"); DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob"); DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
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<String, Cycle> getCycleFunction = new MapFunction<String, Cycle>() {
@Override @Override
public Cycle map(String cycle) throws Exception { public Cycle map(String cycle) throws Exception {
Cycle cycle1 = JSON.parseObject(cycle, Cycle.class); Cycle cycle1 = JSON.parseObject(cycle, Cycle.class);
cycle1 = getCycle(cycle1); cycle1 = getCycle(cycle1);
System.out.println("周期数据->"+cycle); System.out.println("周期数据->"+cycle);
return cycle1; return cycle1;
} }
}; };
MapFunction<String, TravelEvent> getTravelEventFunction = new MapFunction<String, TravelEvent>() { MapFunction<String, TravelEvent> getTravelEventFunction = new MapFunction<String, TravelEvent>() {
@Override @Override
public TravelEvent map(String travelEvent) throws Exception { public TravelEvent map(String travelEvent) throws Exception {
System.out.println("旅行事件数据->"+travelEvent); System.out.println("旅行事件数据->"+travelEvent);
return JSON.parseObject(travelEvent, TravelEvent.class); return JSON.parseObject(travelEvent, TravelEvent.class);
}
};
FilterFunction<TravelEvent> filterFunction = new FilterFunction<TravelEvent>() {
@Override
public boolean filter(TravelEvent car) throws Exception {
//只需要到达进入丢失事件
if (car.getEventType() == EventType.ARRIVED || car.getEventType() == EventType.INCROSS || car.getEventType() == EventType.LAST) {
return true;
} else {
return false;
} }
} };
}; FilterFunction<TravelEvent> filterFunction = new FilterFunction<TravelEvent>() {
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car, timestamp) -> Long.parseLong(car.getEventTime())); @Override
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> Long.parseLong(cycle.getEndDateTime())); public boolean filter(TravelEvent car) throws Exception {
//根据corssid 分组 保证同一路口的数据进行一个进程 //只需要到达进入丢失事件
DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy).keyBy(cycleKeySelect); if (car.getEventType() == EventType.ARRIVED || car.getEventType() == EventType.INCROSS || car.getEventType() == EventType.LAST) {
DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).keyBy(carKeySelector).filter(filterFunction); return true;
} else {
return false;
}
}
};
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car, timestamp) -> Long.parseLong(car.getEventTime()));
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> Long.parseLong(cycle.getEndDateTime()));
//根据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<List<LaneNorm>> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
process.addSink(new LaneNormSink()).name("laneNormJobSink");
env.execute("laneNormJob");
}catch (Exception e){
System.out.println("程序异常关闭->"+e);
}
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("laneNormJob");
} }
/*获取信控指标 并计算相关属性*/ /*获取信控指标 并计算相关属性*/
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/DirectionEvalSecondSink.java
...@@ -3,6 +3,7 @@ package com.zhht.irn.sink; ...@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
import com.zhht.irn.entity.metric.DirectionEvalSecondMetric; import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
import com.zhht.irn.utils.DateUtils; import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils; import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import com.zhht.irn.utils.MySQLUtils; import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java
...@@ -4,6 +4,7 @@ import com.zhht.irn.entity.Cycle; ...@@ -4,6 +4,7 @@ import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage; import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.DateUtils; import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils; import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import com.zhht.irn.utils.MySQLUtils; import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java
...@@ -3,6 +3,7 @@ package com.zhht.irn.sink; ...@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
import com.zhht.irn.entity.LaneNorm; import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.utils.DateUtils; import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils; import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import com.zhht.irn.utils.MySQLUtils; import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
...@@ -11,6 +12,7 @@ import org.apache.flink.streaming.api.functions.sink.SinkFunction; ...@@ -11,6 +12,7 @@ import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import java.sql.Connection; import java.sql.Connection;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
import java.util.List;
/** /**
* @Description: * @Description:
...@@ -19,7 +21,7 @@ import java.sql.PreparedStatement; ...@@ -19,7 +21,7 @@ import java.sql.PreparedStatement;
* @retrun: * @retrun:
* @Creat :2022-10-31-14:11 * @Creat :2022-10-31-14:11
**/ **/
public class LaneNormSink extends RichSinkFunction<LaneNorm> { public class LaneNormSink extends RichSinkFunction<List<LaneNorm>> {
Connection connection; Connection connection;
PreparedStatement insertPstmt; PreparedStatement insertPstmt;
...@@ -49,26 +51,32 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> { ...@@ -49,26 +51,32 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
* 1000w的数据 1000w次 * 1000w的数据 1000w次
*/ */
@Override @Override
public void invoke(LaneNorm value, SinkFunction.Context context) throws Exception { public void invoke(List<LaneNorm> laneNorms, SinkFunction.Context context) throws Exception {
insertPstmt.setString(1, value.getCorssId()); if(laneNorms.size()==0){
insertPstmt.setString(2, value.getLaneId()); return;
insertPstmt.setString(3, value.getDirection()); }
insertPstmt.setLong(4, value.getCycleOrder()); for (LaneNorm value: laneNorms){
insertPstmt.setDouble(5, value.getMaxQueueLength()); insertPstmt.setString(1, value.getCorssId());
insertPstmt.setString(6,value.getEffectGreenTime()); insertPstmt.setString(2, value.getLaneId());
insertPstmt.setDouble(7,value.getPassCapacity()); insertPstmt.setString(3, value.getDirection());
insertPstmt.setDouble(8,value.getTrafficCapacity()); insertPstmt.setLong(4, value.getCycleOrder());
insertPstmt.setDouble(9,value.getSaturation()); insertPstmt.setDouble(5, value.getMaxQueueLength());
insertPstmt.setDouble(10,value.getResidualCapacity()); insertPstmt.setString(6,value.getEffectGreenTime());
insertPstmt.setDouble(11,value.getSpotSpaceShare()); insertPstmt.setDouble(7,value.getPassCapacity());
insertPstmt.setString(12,value.getLaneName()); insertPstmt.setDouble(8,value.getTrafficCapacity());
insertPstmt.setInt(13,value.getOrder()); insertPstmt.setDouble(9,value.getSaturation());
insertPstmt.setString(14, DateUtils.getTodayTime()); insertPstmt.setDouble(10,value.getResidualCapacity());
insertPstmt.setString(15,value.getType()); insertPstmt.setDouble(11,value.getSpotSpaceShare());
insertPstmt.setInt(16,value.getSpotSpaceLength()); insertPstmt.setString(12,value.getLaneName());
insertPstmt.setString(17,value.getCycleStartTime()); insertPstmt.setInt(13,value.getOrder());
insertPstmt.setString(18,value.getCycleEndTime()); insertPstmt.setString(14, DateUtils.getTodayTime());
insertPstmt.setString(19,DateUtils.getTodayTime()); insertPstmt.setString(15,value.getType());
insertPstmt.execute(); insertPstmt.setInt(16,value.getSpotSpaceLength());
insertPstmt.setString(17,value.getCycleStartTime());
insertPstmt.setString(18,value.getCycleEndTime());
insertPstmt.setString(19,DateUtils.getTodayTime());
insertPstmt.addBatch();
}
insertPstmt.executeBatch();
} }
} }
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelMetricSink.java
...@@ -2,6 +2,7 @@ package com.zhht.irn.sink; ...@@ -2,6 +2,7 @@ package com.zhht.irn.sink;
import com.zhht.irn.entity.Travel; import com.zhht.irn.entity.Travel;
import com.zhht.irn.utils.DorisUtils; import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DruidConnectPoolUtils.java
package com.zhht.irn.utils; package com.zhht.irn.utils;
import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import org.apache.flink.api.java.utils.ParameterTool;
import java.io.InputStream;
import java.sql.Connection; import java.sql.Connection;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
import java.sql.SQLException; import java.sql.SQLException;
...@@ -12,6 +15,9 @@ import java.sql.SQLException; ...@@ -12,6 +15,9 @@ import java.sql.SQLException;
* @date 2022-09-05 15:11:00 * @date 2022-09-05 15:11:00
*/ */
public class DruidConnectPoolUtils { public class DruidConnectPoolUtils {
private static DruidDataSource pool=null;
/** /**
* 初始化连接池 * 初始化连接池
* @param dbHosts * @param dbHosts
...@@ -25,15 +31,66 @@ public class DruidConnectPoolUtils { ...@@ -25,15 +31,66 @@ public class DruidConnectPoolUtils {
dataSource.setUrl(dbHosts); dataSource.setUrl(dbHosts);
dataSource.setUsername(username); dataSource.setUsername(username);
dataSource.setPassword(password); dataSource.setPassword(password);
dataSource.setInitialSize(3); //初始连接数，默认0
dataSource.setMaxActive(10); //最大连接数，默认8
dataSource.setMinIdle(5); //最小闲置数
dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
dataSource.setTestOnBorrow(false); // 借出连接时不要测试，否则很影响性能
return dataSource;
}
public static void initDataSource(String dbHosts, String username, String password) {
DruidDataSource dataSource = new DruidDataSource();
dataSource.setDriverClassName("com.mysql.jdbc.Driver");
dataSource.setUrl(dbHosts);
dataSource.setUsername(username);
dataSource.setPassword(password);
dataSource.setInitialSize(3); //初始连接数，默认0 dataSource.setInitialSize(3); //初始连接数，默认0
dataSource.setMaxActive(30); //最大连接数，默认8 dataSource.setMaxActive(10); //最大连接数，默认8
dataSource.setMinIdle(5); //最小闲置数 dataSource.setMinIdle(5); //最小闲置数
dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒 dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除 dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
dataSource.setTestOnBorrow(false); // 借出连接时不要测试，否则很影响性能 dataSource.setTestOnBorrow(false); // 借出连接时不要测试，否则很影响性能
pool=dataSource;
}
return dataSource; public static DruidPooledConnection getConnection(){
try {
if(pool==null){
InputStream inputStream = DruidConnectPoolUtils.class.getClassLoader().getResourceAsStream("doris.properties");
ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
// 根据路径的配置文件获取MySQL链接参数
String driver = tool.getRequired("driver");
String username = tool.getRequired("username");
String password = tool.getRequired("password");
String host = tool.getRequired("host");
String port = tool.getRequired("port");
String database = tool.get("database", "default");
String jdbcUrl = String.format("jdbc:mysql://%s:%s/%s", host, port, database);
initDataSource(jdbcUrl,username,password);
}
}catch (Exception e){
e.printStackTrace();
}
try {
return pool.getConnection();
} catch (SQLException e) {
e.printStackTrace();
}
return null;
}
public static void colseDataSource() {
try {
if(pool!=null){
pool.close();
}
pool.close();
} catch (Exception e) {
e.printStackTrace();
}
} }
/** /**
......
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