Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 2aad0262 by 黄准
Options

车道指标相关计算逻辑

parent 37d862da
Showing with 71 additions and 53 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java
......@@ -5,6 +5,7 @@ import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.functions.LaneNormProcessFunction;
import com.zhht.irn.sink.LaneNormSink;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
......@@ -26,60 +27,66 @@ 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>() {
@Override
public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode();
}
};
KeySelector<TravelEvent, String> carKeySelector = new KeySelector<TravelEvent, String>() {
@Override
public String getKey(TravelEvent car) throws Exception {
return car.toString();
}
};
MapFunction<String, Cycle> getCycleFunction = new MapFunction<String, Cycle>() {
@Override
public Cycle map(String cycle) throws Exception {
Cycle cycle1 = JSON.parseObject(cycle, Cycle.class);
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>() {
@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;
try {
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>() {
@Override
public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode();
}
};
KeySelector<TravelEvent, String> carKeySelector = new KeySelector<TravelEvent, String>() {
@Override
public String getKey(TravelEvent car) throws Exception {
return car.toString();
}
};
MapFunction<String, Cycle> getCycleFunction = new MapFunction<String, Cycle>() {
@Override
public Cycle map(String cycle) throws Exception {
Cycle cycle1 = JSON.parseObject(cycle, Cycle.class);
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);
}
}
};
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);
};
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;
}
}
};
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);
//程序异常 关闭数据库连接池
DruidConnectPoolUtils.colseDataSource();
}
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");
}
/*获取信控指标 并计算相关属性*/
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DruidConnectPoolUtils.java
......@@ -59,7 +59,7 @@ public class DruidConnectPoolUtils {
public static DruidPooledConnection getConnection(){
try {
if(pool==null){
InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("doris.properties");
InputStream inputStream = DruidConnectPoolUtils.class.getClassLoader().getResourceAsStream("doris.properties");
ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
// 根据路径的配置文件获取MySQL链接参数
String driver = tool.getRequired("driver");
......@@ -82,6 +82,17 @@ public class DruidConnectPoolUtils {
return null;
}
public static void colseDataSource() {
try {
if(pool!=null){
pool.close();
}
pool.close();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* CK close
* @param connection
......
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