Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 68ada241 by 芮蒙
Options

2022-12-14 切换使用旅行信息进行计算行车情况相关指标

parent 038fb0cc
Showing with 299 additions and 107 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelCarInfoCoProcessFunction.java
......@@ -25,7 +25,7 @@ import java.util.*;
* @author ruimeng
* @create 2022-11-14 17:16
**/
public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo, CycleSignalData, Object> {
public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>> {
//定义状态，用于存储旅行信息，当周期信号数据过来的时候，触发计算
//状态数据存储结构为 Map（路口id--》（Map（车道id-》（List（旅行信息））））） 一辆车只有一条旅行信息
......@@ -91,7 +91,9 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
alreadyHandleCycleSignalDataState = getRuntimeContext().getMapState(alreadyHandleCycleSignalDataStateDescriptor);
cycleSectionState = getRuntimeContext().getListState(cycleSectionStateDescriptor);
// connection = DruidConnectPoolUtils.getConnection();
connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://localhost:3309/test?characterEncoding=utf8",
// connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://localhost:3309/test?characterEncoding=utf8",
connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
"root","mima").getConnection();
dim_cnt_cross_lane_position = getLaneDimData(connection);
......@@ -107,8 +109,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
//把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
// 所有旅行信息存储在状态中，当周期数据过来的时候，去状态中挑选属于这个周期内的旅行信息的数据即可
@Override
public void processElement1(TravelInfo value, CoProcessFunction<TravelInfo, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
public void processElement1(TravelInfo value, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.Context ctx,
Collector<List<TravelLineSinkInfo>> out) throws Exception {
String crossId = value.getCrossId();
Iterator<Map<String, List<TravelInfo>>> iterator = travelInfoState.values().iterator();
......@@ -119,19 +121,27 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
}
//获取当前路口的数据
Map<String, List<TravelInfo>> crossListMap = travelInfoState.get(crossId);
//当前路口没有数据进来或者当前路口有数据，但是，当前车道没有数据
if (crossListMap == null || !crossListMap.containsKey(value.getInCrossLineId())) {
//当前路口没有数据
if(crossListMap == null ){
Map<String, List<TravelInfo>> crossMap = new HashMap<>();
List<TravelInfo> list = new ArrayList<>();
list.add(value);
//车道编号，当前车道的旅行信息
crossMap.put(value.getInCrossLineId(), list);
travelInfoState.put(crossId, crossMap);
}
//当前路口有数据，但是当前车道没有数据
if (crossListMap != null && !crossListMap.containsKey(value.getInCrossLineId())) {
List<TravelInfo> list = new ArrayList<>();
list.add(value);
//车道编号，当前车道的旅行信息
crossListMap.put(value.getInCrossLineId(), list);
travelInfoState.put(crossId, crossListMap);
//当前路口有数据，且 该路口的当前车道也有数据进来了,则把这条旅行信息，放入对应路口下的对应的车道中去
} else {
}
//当前路口有数据，而且 当前车道 也有数据
if(crossListMap != null && crossListMap.containsKey(value.getInCrossLineId())) {
List<TravelInfo> lineTravelInfos = crossListMap.get(value.getInCrossLineId());
//如果，当前车道的数据，积压超过2000条，则进行清理操作
if(lineTravelInfos.size()>2000){
System.out.println("移除积压的数据。。。。");
......@@ -152,67 +162,75 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
//来一条周期数据，触发这个周期内的旅行数据
@Override
public void processElement2(CycleSignalData value, CoProcessFunction<TravelInfo, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
//判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
if(alreadyHandleCycleSignalDataState.contains(value.getCycleOrder())){
System.out.println("周期数据重复----当前路口id是"+value.getCrossCode()+"当前周期已经计算完成了"+value.getCycleOrder());
return;
}
cycleSectionState.add(value.getBeginDateTime());
public void processElement2(CycleSignalData value, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.Context ctx,
Collector<List<TravelLineSinkInfo>> out) {
try {
//判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
if (alreadyHandleCycleSignalDataState.contains(value.getCycleOrder())) {
System.out.println("周期数据重复----当前路口id是" + value.getCrossCode() + "当前周期已经计算完成了" + value.getCycleOrder());
return;
}
cycleSectionState.add(value.getBeginDateTime());
String crossCode = value.getCrossCode();
Integer cycleOrder = value.getCycleOrder();
//周期开始时间，周期结束时间
Long beginDateTime = value.getBeginDateTime();
Long endDateTime = value.getEndDateTime();
//如果，存在之前周期没有计算的，先计算之前的周期的数据
if (!cycleSignalDataState.isEmpty()) {
Iterator<CycleSignalData> iterator = cycleSignalDataState.values().iterator();
List<Integer> listToRemove = new ArrayList<>();
while (iterator.hasNext()) {
CycleSignalData beforeCycle = iterator.next();
Map<String, List<TravelInfo>> beforeCycleDataListMap = filterThisCycleData(crossCode, beforeCycle.getBeginDateTime(), beforeCycle.getEndDateTime());
if (beforeCycleDataListMap != null) {
List<TravelLineSinkInfo> travelLineSinkInfos = handTwoStreamData(beforeCycleDataListMap, beforeCycle.getCycleOrder(),
crossCode, beforeCycle.getBeginDateTime(), beforeCycle.getEndDateTime());
out.collect(travelLineSinkInfos);
// 当前周期已经处理完成了，kafka中再次来了该周期的数据不需要再次触发计算了
alreadyHandleCycleSignalDataState.put(beforeCycle.getCycleOrder(), beforeCycle);
listToRemove.add(beforeCycle.getCycleOrder());
}
}
for (Integer i:listToRemove) {
cycleSignalDataState.remove(i);
}
}
String crossCode = value.getCrossCode();
Integer cycleOrder = value.getCycleOrder();
//周期开始时间，周期结束时间
Long beginDateTime = value.getBeginDateTime();
Long endDateTime = value.getEndDateTime();
// 拿到这个周期内的当前路口的所有的车道的旅行信息数据 并移除之前周期的数据
Map<String, List<TravelInfo>> thisCycleDataListMap = filterThisCycleData(crossCode, beginDateTime, endDateTime);
//如果，有旅行信息数据就计算，否则就不计算
if (thisCycleDataListMap!=null && !thisCycleDataListMap.values().isEmpty()) {
List<TravelLineSinkInfo> travelLineSinkInfos = handTwoStreamData(thisCycleDataListMap, cycleOrder, crossCode, beginDateTime, endDateTime);
out.collect(travelLineSinkInfos);
// 当前周期已经处理完成了，，kafka中再次来了该周期的数据不需要再次触发计算了
alreadyHandleCycleSignalDataState.put(cycleOrder, value);
cycleSignalDataState.remove(cycleOrder);
} else {
cycleSignalDataState.put(cycleOrder, value);
}
//如果，存在之前周期没有计算的，先计算之前的周期的数据
if(!cycleSignalDataState.isEmpty()){
Iterator<CycleSignalData> iterator = cycleSignalDataState.values().iterator();
while(iterator.hasNext()){
CycleSignalData beforeCycle = iterator.next();
Map<String, List<TravelInfo>> beforeCycleDataListMap = filterThisCycleData(crossCode, beforeCycle.getBeginDateTime(), beforeCycle.getEndDateTime());
// System.out.println("当前处理周期"+beforeCycle.getCycleOrder()+"<---->路口"+next.getCrossCode());
if(beforeCycleDataListMap!=null){
handTwoStreamData(beforeCycleDataListMap,beforeCycle.getCycleOrder(),crossCode,beforeCycle.getBeginDateTime(), beforeCycle.getEndDateTime());
// 当前周期已经处理完成了，kafka中再次来了该周期的数据不需要再次触发计算了
alreadyHandleCycleSignalDataState.put(beforeCycle.getCycleOrder(),beforeCycle);
cycleSignalDataState.remove(beforeCycle.getCycleOrder());
}
// 注册定时器
//获取当前的ProcessingTime
long currentProcessingTime = ctx.timerService().currentProcessingTime();
// 只在某个时间段注册定时器
SimpleDateFormat sdf = new SimpleDateFormat("HH");
if (sdf.format(currentProcessingTime).equals("11")) {
ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 30 * 60 * 1000);
}
}
// 拿到这个周期内的当前路口的所有的车道的旅行信息数据 并移除之前周期的数据
Map<String, List<TravelInfo>> thisCycleDataListMap = filterThisCycleData(crossCode, beginDateTime, endDateTime);
//如果，有旅行信息数据就计算，否则就不计算
if(thisCycleDataListMap!=null){
handTwoStreamData(thisCycleDataListMap,cycleOrder,crossCode,beginDateTime,endDateTime);
// 当前周期已经处理完成了，，kafka中再次来了该周期的数据不需要再次触发计算了
alreadyHandleCycleSignalDataState.put(cycleOrder,value);
cycleSignalDataState.remove(cycleOrder);
} else {
cycleSignalDataState.put(cycleOrder,value);
} catch (Exception e) {
e.printStackTrace();
}
// 注册定时器
//获取当前的ProcessingTime
long currentProcessingTime = ctx.timerService().currentProcessingTime();
// 只在某个时间段注册定时器
SimpleDateFormat sdf = new SimpleDateFormat("HH");
if(sdf.format(currentProcessingTime).equals("11")) {
ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+30*60*1000);
}
}
@Override
public void onTimer(long timestamp, CoProcessFunction<TravelInfo, CycleSignalData, Object>.OnTimerContext ctx, Collector<Object> out) throws Exception {
public void onTimer(long timestamp, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.OnTimerContext ctx, Collector<List<TravelLineSinkInfo>> out) throws Exception {
super.onTimer(timestamp, ctx, out);
......@@ -231,7 +249,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
* @param crossCode
* @throws SQLException
*/
private void handTwoStreamData(Map<String, List<TravelInfo>> thisCycleDataListMap, Integer cycleOrder,String crossCode
private List<TravelLineSinkInfo> handTwoStreamData(Map<String, List<TravelInfo>> thisCycleDataListMap, Integer cycleOrder,String crossCode
,Long beginDateTime ,Long endDateTime) throws Exception {
......@@ -244,7 +262,6 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
List<TravelInfo> travelInfos = thisCycleDataListMap.get(laneId);
//需要将这个车道上的所有的车辆，按照进入路口的时间依次排序才能计算出车头时距
List<TravelInfo> orderedTravelInfoList = orderByInCrossTime(travelInfos);
List<TravelCarInfo> list = new ArrayList<>();
//这里的旅行信息的数据，一定是完成了从路口出现到丢失的整个过程的数据，一辆车只有一条数据
......@@ -313,7 +330,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
maxInCrossTime = t.getInCrossTime();
}
}
int size = next.size();
TravelLineSinkInfo travelLineSinkInfo = new TravelLineSinkInfo();
travelLineSinkInfo.setCross_id(crossCode);
travelLineSinkInfo.setCycle_id(cycleOrder);
......@@ -321,12 +338,12 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
travelLineSinkInfo.setFlow_direction(flowDirection);
travelLineSinkInfo.setLane_id(next.get(0).getInCrossLineId());
travelLineSinkInfo.setPass_numbers(next.size());
travelLineSinkInfo.setAverage_pass_time(sumPassTime / (next.size()) * 1.0);
travelLineSinkInfo.setAverage_pass_speed(sumSpeed / (next.size()));
travelLineSinkInfo.setAverage_control_delay(sumControlDelay / (next.size()) * 1.0);
travelLineSinkInfo.setAverage_stop_times(sumStopTimes / (next.size()) * 1.0);
travelLineSinkInfo.setAverage_stop_delay(sumStopDelay / (next.size()) * 1.0);
travelLineSinkInfo.setAverage_car_head_time_gap(sumCarHeadTimeGap / (next.size()) * 1.0);
travelLineSinkInfo.setAverage_pass_time(sumPassTime/(size*1.000*1000));
travelLineSinkInfo.setAverage_pass_speed(sumSpeed / (size));
travelLineSinkInfo.setAverage_control_delay(-1.0);
travelLineSinkInfo.setAverage_stop_times(sumStopTimes/(size*1.000));
travelLineSinkInfo.setAverage_stop_delay(sumStopDelay/(size*1.000*1000));
travelLineSinkInfo.setAverage_car_head_time_gap(sumCarHeadTimeGap/(size*1.000*1000));
travelLineSinkInfo.setLast_car_inCross_time(maxInCrossTime);
travelLineSinkInfo.setCycle_begin_time(beginDateTime);
travelLineSinkInfo.setCycle_end_time(endDateTime);
......@@ -334,7 +351,10 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
}
}
sinkTravelLineSinkInfoToMysql(sinkList);
return sinkList ;
// sinkTravelLineSinkInfoToMysql(sinkList);
}
......@@ -374,12 +394,14 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
leftData.add(t);
}
}
thisCycleData.put(laneId, thisData);
if(thisData.size()!=0){
thisCycleData.put(laneId, thisData);
}
leftCycleData.put(laneId,leftData);
//更新旅行信息状态中的数据，只保留没有使用的旅行信息数据
travelInfoState.put(crossCode, leftCycleData);
return thisCycleData;
}
//更新旅行信息状态中的数据，只保留没有使用的旅行信息数据
travelInfoState.put(crossCode, leftCycleData);
return thisCycleData;
}
return null;
......@@ -417,10 +439,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
//3、获取平均速度 只看在旅行开始时间和旅行结束时间之间的数据
Double averageSpeed = getAverageSpeed(travel.getLocations(),travel.getTravelBeginTime(),travel.getTravelEndTime());
//5、计算停车次数(这里使用停车次数，非实际停车次数)
Integer stopTimes = getStopTimes(travel.getLocations(),beginDateTime,endDateTime)==0?0:1;
Integer stopTimes = getStopTimes(travel.getLocations());
//6、计算停车延误时间
Long stopDelayTime = getStopDelayTime(travel.getLocations());
......@@ -506,13 +526,13 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
}
/**
* 计算停车次数 根据传入的locations集合，通过按序排列每帧信息，获取到停车次数
* 计算停车次数 根据传入的locations集合，划分到不同的周期中去，在每个周期统计实际停车次数
* 根据捕获时间排序，从小到大
* 这里求的是实际停车次数
* @param locations
* @return
*/
private Integer getStopTimes(List<Location> locations,Long beginDateTime ,Long endDateTime) throws Exception {
private Integer getStopTimes(List<Location> locations) throws Exception {
//周期的开始时间
Iterator<Long> iterator1 = cycleSectionState.get().iterator();
......@@ -535,37 +555,112 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
// 根据最近的几个周期的开始时间对locations数据进行划分，划分到不同的周期
Map<Long, List<Location>> longListMap = divideLocations(beginTimes, locations);
Set<Long> longs = longListMap.keySet();
int stopTimes =0 ;
for (Long l:longs) {
List<Location> locations1 = longListMap.get(l);
Integer realStopTimes = getRealStopTimes(locations1);
// 一个周期内发生的实际停车次数大于0,停车次数加1
if(realStopTimes > 0){
stopTimes++;
}
}
return stopTimes;
}
/**
* 根据一系列的location信息，判断发生的实际停车次数
* @param locations
* @return
*/
private Integer getRealStopTimes(List<Location> locations){
// 按照捕获时间进行排序
Set<Location> set = new TreeSet<>();
for (Location location : locations) {
set.add(location);
}
// 按照捕获时间进行排序
Iterator<Location> iterator = set.iterator();
List<Location> speedList = new ArrayList<>();
while (iterator.hasNext()) {
speedList.add(iterator.next());
}
int stopTimes = 0;
int realStopTimes = 0;
for (int i = 0; i < speedList.size(); i++) {
//第一帧数据小于3视为停车
if (i == 0 && speedList.get(i).getSpeed() <= 3.0) {
if (i == 0 && speedList.get(i).getSpeed() < 3.0) {
realStopTimes++;
//从第二帧开始，只有前一帧速度大于10，且这一帧数据小于3的才算一次停车，连续小于3的视为一次停车
} else {
if (speedList.get(i).getSpeed() <= 3.0 && speedList.get(i - 1).getSpeed() > 10.0) {
if (speedList.get(i).getSpeed() < 3.0 && speedList.get(i - 1).getSpeed() >= 10.0) {
realStopTimes++;
}
}
}
return realStopTimes;
return stopTimes;
}
/**
* 根据最近的几个周期的开始时间，把对应的locations划分到不同的周期中去，以便计算停车次数
* @param beginTimes
* @param locations
* @return 返回值是map结构 key->list<Location> 周期开始时间 -> 该周期对应的位置信息
*/
private Map<Long,List<Location>> divideLocations(List<Long> beginTimes,List<Location> locations){
Map<Long,List<Location>> map = new HashMap<>();
if(beginTimes.size()==1){
List<Location> list = new ArrayList<>();
List<Location> list2 = new ArrayList<>();
for (Location location:locations) {
if (location.getDtTranjectory()<beginTimes.get(0)){
list.add(location);
} else {
list2.add(location);
}
}
map.put(beginTimes.get(0),list);
map.put(999999999L,list2);
}
for (int i=0;i<beginTimes.size();i++) {
List<Location> list = new ArrayList<>();
if(i==0){
for (Location location:locations) {
if (location.getDtTranjectory()<beginTimes.get(i)){
list.add(location);
}
}
map.put(beginTimes.get(i),list);
// 最后一个周期的数据
} else if(i==(beginTimes.size()-1)){
for (Location location:locations) {
if (location.getDtTranjectory()>beginTimes.get(i)){
list.add(location);
}
}
map.put(beginTimes.get(i),list);
} else {
for (Location location:locations) {
if (location.getDtTranjectory()<beginTimes.get(i)&&location.getDtTranjectory()>beginTimes.get(i-1)){
list.add(location);
}
}
map.put(beginTimes.get(i),list);
}
}
return map ;
}
/**
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelSituationAnalysisJob.java
......@@ -3,17 +3,21 @@ package com.zhht.irn.job;
import com.zhht.irn.entity.dto.CycleSignalData;
import com.zhht.irn.entity.dto.TravelEvent;
import com.zhht.irn.entity.dto.TravelInfo;
import com.zhht.irn.entity.dto.TravelLineSinkInfo;
import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
import com.zhht.irn.functions.TravelEventAndCycleCoProcessFunction;
import com.zhht.irn.schema.CycleSignalKafkaSchema;
import com.zhht.irn.schema.TravelEventKafkaSchema;
import com.zhht.irn.schema.TravelInfoKafkaSchema;
import com.zhht.irn.sink.TravelLaneInfoSink;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
......@@ -21,6 +25,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Properties;
/**
......@@ -39,17 +44,17 @@ public class TravelSituationAnalysisJob {
// checkpoint配置
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
// checkpoint时间间隔3分钟
checkpointConfig.setCheckpointInterval(1* 60 * 1000);
checkpointConfig.setCheckpointInterval(1 * 60 * 1000);
// 两次checkpoint中最短时间间隔1分钟
checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
// 同时同时允许1个checkpoint进行
checkpointConfig.setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//设置重启策略
env.setRestartStrategy( RestartStrategies.fixedDelayRestart(1, Time.seconds(10)));
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, Time.seconds(10)));
/*
关于flink checkpoint存储的设置统一在集群配置文件中设置
......@@ -58,17 +63,17 @@ public class TravelSituationAnalysisJob {
Properties kafkaProperties = new Properties();
// kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
// kafkaProperties.setProperty("group.id", "travel_event");
// kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
// kafkaProperties.setProperty("group.id", "travel_event");
kafkaProperties.setProperty("group.id", "trips_info2");
kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
kafkaProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
Properties kafkaProperties2 = new Properties();
// kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
// kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
kafkaProperties2.setProperty("group.id", "cycle_signal2");
kafkaProperties2.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
......@@ -82,37 +87,42 @@ public class TravelSituationAnalysisJob {
DataStreamSource<TravelEvent> travelEventStream = env.addSource(travelEvent).setParallelism(3);*/
// FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
// FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
// FlinkKafkaConsumer<String> travelInfo = new FlinkKafkaConsumer<>("t_info2", new SimpleStringSchema(), kafkaProperties);
FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
travelInfo.setStartFromEarliest();
DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
travelInfoStream.print();
// travelInfoStream.print();
//周期信号
// FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
// FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
cycleSignalData.setStartFromEarliest();
DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
cycleSignalDataStream.print();
// cycleSignalDataStream.print();
//2、使用connect，把2个流联合处理
ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
.connect(cycleSignalDataStream);
ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
.connect(cycleSignalDataStream);
ConnectedStreams<TravelInfo, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
(KeySelector<TravelInfo, String>) travelInfo1 -> travelInfo1.getCrossId(),
(KeySelector<CycleSignalData, String>) cycleSignalData1 -> cycleSignalData1.getCrossCode()
);
ConnectedStreams<TravelInfo, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
(KeySelector<TravelInfo, String>) travelInfo1 -> travelInfo1.getCrossId(),
(KeySelector<CycleSignalData, String>) cycleSignalData1 -> cycleSignalData1.getCrossCode()
);
travelInfoCycleSignalDataConnectedStreams.process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
SingleOutputStreamOperator<List<TravelLineSinkInfo>> sinkDataStream = travelInfoCycleSignalDataConnectedStreams
.process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
env.execute("TravelSituationAnalysisJob");
env.execute("TravelSituationAnalysisJob");
} catch (Exception e){
e.printStackTrace();
}
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelLaneInfoSink.java 0 → 100644
package com.zhht.irn.sink;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.zhht.irn.entity.dto.TravelLineSinkInfo;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* TravelLaneInfoSink
*
* @author Rui meng
* @description 行车情况信息sink
* @date 2022/12/15 9:39
*/
public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo>> {
private transient DruidPooledConnection connection;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
"root","mima").getConnection();
}
@Override
public void close() throws Exception {
super.close();
connection.close();
}
@Override
public void invoke(List<TravelLineSinkInfo> value, Context context) throws Exception {
super.invoke(value, context);
String sql = "INSERT INTO test.app_cross_line_travel_car_info" +
"(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
" flow_direction, pass_numbers," +
" last_car_inCross_time, average_pass_time, average_pass_speed, average_control_delay, " +
"average_stop_delay, average_stop_times, average_car_head_time_gap" +
")" +
"VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?,?,?,?) ";
PreparedStatement preparedStatement = connection.prepareStatement(sql);
//TODO 这里的时间使用数据携带的时间，使用最后进入路口车辆的时间，这样在补数据时，可以补充到对应的分区下
Date date = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
String statistic_time = sdf.format(date);
for (TravelLineSinkInfo t :value) {
// 使用事件数据的时间划分数据分区，这样可以进行补数操作
String record_date = sdf2.format(t.getLast_car_inCross_time());
preparedStatement.setString(1, record_date);
preparedStatement.setString(2, t.getCross_id());
preparedStatement.setString(3, statistic_time);
preparedStatement.setString(4, sdf.format(t.getCycle_begin_time()));
preparedStatement.setString(5, sdf.format(t.getCycle_end_time()));
preparedStatement.setString(6, t.getLane_id());
preparedStatement.setString(7, t.getCycle_id().toString());
preparedStatement.setString(8, t.getDirection());
preparedStatement.setString(9, t.getFlow_direction());
preparedStatement.setInt(10, t.getPass_numbers());
preparedStatement.setLong(11, t.getLast_car_inCross_time());
preparedStatement.setDouble(12, t.getAverage_pass_time());
preparedStatement.setDouble(13, t.getAverage_pass_speed());
preparedStatement.setDouble(14, t.getAverage_control_delay());
preparedStatement.setDouble(15, t.getAverage_stop_delay());
preparedStatement.setDouble(16, t.getAverage_stop_times());
preparedStatement.setDouble(17, t.getAverage_car_head_time_gap());
preparedStatement.addBatch();
}
preparedStatement.executeBatch();
}
}
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