Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 038fb0cc by 芮蒙
Options

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

parent 2930fc23
Showing with 824 additions and 68 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/Location.java
......@@ -2,6 +2,8 @@ package com.zhht.irn.entity.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
/**
* 位置信息
......@@ -11,6 +13,7 @@ import lombok.Data;
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Location implements Comparable<Location>{
private Double longitude ;//可选 经度
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelInfo.java
......@@ -11,7 +11,7 @@ import java.util.List;
* @create 2022-11-14 13:33{
**/
@Data
public class TravelInfo {
public class TravelInfo implements Comparable<TravelInfo> {
private Integer id ;//必填 ID
private String crossId ;//必填 路口ID
......@@ -45,4 +45,9 @@ public class TravelInfo {
private Double crossingTime ;//可选 旅行时长，单位 秒 (s)（驶离时刻-到达时刻）
private List<Location> locations ;//必填 数组，参考 位置信息字段
private String remark ;//可选 备注
@Override
public int compareTo(TravelInfo o) {
return (int)(inCrossTime-o.getInCrossTime());
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelCarInfoCoProcessFunction.java 0 → 100644
package com.zhht.irn.functions;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.zhht.irn.entity.dto.*;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.*;
/**
* 行车情况数据处理函数，双流connect后处理
*
* @author ruimeng
* @create 2022-11-14 17:16
**/
public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo, CycleSignalData, Object> {
//定义状态，用于存储旅行信息，当周期信号数据过来的时候，触发计算
//状态数据存储结构为 Map（路口id--》（Map（车道id-》（List（旅行信息））））） 一辆车只有一条旅行信息
// 这样就把每个路口的数据进行分流，同时，对每个路口的数据，也按照了不同的车道进行了分流，同一个路口，同一个车道的数据存在一个list中
private MapState<String, Map<String, List<TravelInfo>>> travelInfoState;
private MapState<Integer, CycleSignalData> cycleSignalDataState; //缓存周期数据（避免周期数据先到，而旅行数数据后到的情况）
private MapState<Integer, CycleSignalData> alreadyHandleCycleSignalDataState; //已经处理过的周期数据，避免重复触发
private ListState<Long> cycleSectionState ;
private transient DruidPooledConnection connection;
private static Map<String, Map<String, String>> dim_cnt_cross_lane_position;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
MapStateDescriptor<String, Map<String, List<TravelInfo>>> travelInfoStateDescriptor =
new MapStateDescriptor("travelInfoState",
TypeInformation.of(String.class), TypeInformation.of(Map.class));
MapStateDescriptor<Integer, CycleSignalData> cycleSignalDataStateDescriptor =
new MapStateDescriptor("cycleSignalDataState",
TypeInformation.of(Integer.class), TypeInformation.of(CycleSignalData.class));
MapStateDescriptor<Integer, CycleSignalData> alreadyHandleCycleSignalDataStateDescriptor =
new MapStateDescriptor("alreadyHandleCycleSignalDataState",
TypeInformation.of(Integer.class), TypeInformation.of(CycleSignalData.class));
ListStateDescriptor cycleSectionStateDescriptor = new ListStateDescriptor("cycleSectionState", TypeInformation.of(Long.class));
StateTtlConfig stateTtlConfig4TravelInfo = StateTtlConfig
// 状态有效时间 12小时
.newBuilder(Time.minutes(15))
// 设置状态的更新类型
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
// 已过期还未被清理掉的状态数据不返回给用户
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
// 过期对象的清理策略 全量清理
.cleanupFullSnapshot()
.build();
travelInfoStateDescriptor.enableTimeToLive(stateTtlConfig4TravelInfo);
StateTtlConfig stateTtlConfig4CycleData = StateTtlConfig
// 状态有效时间 3小时
.newBuilder(Time.minutes(3))
// 设置状态的更新类型
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
// 已过期还未被清理掉的状态数据不返回给用户
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
// 过期对象的清理策略 全量清理
.cleanupFullSnapshot()
.build();
cycleSignalDataStateDescriptor.enableTimeToLive(stateTtlConfig4CycleData);
travelInfoState = getRuntimeContext().getMapState(travelInfoStateDescriptor);
cycleSignalDataState = getRuntimeContext().getMapState(cycleSignalDataStateDescriptor);
alreadyHandleCycleSignalDataState = getRuntimeContext().getMapState(alreadyHandleCycleSignalDataStateDescriptor);
cycleSectionState = getRuntimeContext().getListState(cycleSectionStateDescriptor);
// connection = DruidConnectPoolUtils.getConnection();
connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://localhost:3309/test?characterEncoding=utf8",
"root","mima").getConnection();
dim_cnt_cross_lane_position = getLaneDimData(connection);
}
@Override
public void close() throws Exception {
super.close();
connection.close();
}
//把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
// 所有旅行信息存储在状态中，当周期数据过来的时候，去状态中挑选属于这个周期内的旅行信息的数据即可
@Override
public void processElement1(TravelInfo value, CoProcessFunction<TravelInfo, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
String crossId = value.getCrossId();
Iterator<Map<String, List<TravelInfo>>> iterator = travelInfoState.values().iterator();
//如果状态为空，则进行初始化
if (!iterator.hasNext()) {
Map<String, Map<String, List<TravelInfo>>> initMap = new HashMap<>();
travelInfoState.putAll(initMap);
}
//获取当前路口的数据
Map<String, List<TravelInfo>> crossListMap = travelInfoState.get(crossId);
//当前路口没有数据进来或者当前路口有数据，但是，当前车道没有数据
if (crossListMap == null || !crossListMap.containsKey(value.getInCrossLineId())) {
Map<String, List<TravelInfo>> crossMap = new HashMap<>();
List<TravelInfo> list = new ArrayList<>();
list.add(value);
//车道编号，当前车道的旅行信息
crossMap.put(value.getInCrossLineId(), list);
travelInfoState.put(crossId, crossMap);
//当前路口有数据，且 该路口的当前车道也有数据进来了,则把这条旅行信息，放入对应路口下的对应的车道中去
} else {
List<TravelInfo> lineTravelInfos = crossListMap.get(value.getInCrossLineId());
//如果，当前车道的数据，积压超过2000条，则进行清理操作
if(lineTravelInfos.size()>2000){
System.out.println("移除积压的数据。。。。");
List<TravelInfo> travelInfoList = orderByInCrossTime(lineTravelInfos);
List<TravelInfo> travelInfoListNew = travelInfoList.subList(500, travelInfoList.size());
travelInfoListNew.add(value);
crossListMap.put(value.getInCrossLineId(), travelInfoListNew);
travelInfoState.put(crossId, crossListMap);
} else {
lineTravelInfos.add(value);
crossListMap.put(value.getInCrossLineId(), lineTravelInfos);
travelInfoState.put(crossId, crossListMap);
}
}
}
//来一条周期数据，触发这个周期内的旅行数据
@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());
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();
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());
}
}
}
// 拿到这个周期内的当前路口的所有的车道的旅行信息数据 并移除之前周期的数据
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);
}
// 注册定时器
//获取当前的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 {
super.onTimer(timestamp, ctx, out);
//更新维表数据 只在每天12点的时候更新
SimpleDateFormat sdf = new SimpleDateFormat("HH");
if(sdf.format(timestamp).equals("12")){
dim_cnt_cross_lane_position = getLaneDimData(connection);
}
dim_cnt_cross_lane_position = getLaneDimData(connection);
}
/**
* 处理2个流的数据，并sink
* @param thisCycleDataListMap
* @param cycleOrder
* @param crossCode
* @throws SQLException
*/
private void handTwoStreamData(Map<String, List<TravelInfo>> thisCycleDataListMap, Integer cycleOrder,String crossCode
,Long beginDateTime ,Long endDateTime) throws Exception {
Set<String> laneIds = thisCycleDataListMap.keySet();
List<TravelLineSinkInfo> sinkList = new ArrayList<>();
for (String laneId : laneIds) {
// 当前车道的所有的旅行信息 ，一辆车对应1条信息
List<TravelInfo> travelInfos = thisCycleDataListMap.get(laneId);
//需要将这个车道上的所有的车辆，按照进入路口的时间依次排序才能计算出车头时距
List<TravelInfo> orderedTravelInfoList = orderByInCrossTime(travelInfos);
List<TravelCarInfo> list = new ArrayList<>();
//这里的旅行信息的数据，一定是完成了从路口出现到丢失的整个过程的数据，一辆车只有一条数据
//根据周期数据及旅行信息，查询维表，将每一条旅行信息转换为行车情况明细数据(1对1转换)
for (int i = 0; i < orderedTravelInfoList.size(); i++) {
TravelInfo travel = orderedTravelInfoList.get(i);
Long beforeCarInCrossTime;
if (i != 0) {
beforeCarInCrossTime = travelInfos.get(i - 1).getInCrossTime();
} else {
beforeCarInCrossTime = 0L;
}
TravelCarInfo travelCarInfo = trunTravelAndCycleToTravelCarInfo(travel, cycleOrder, beforeCarInCrossTime,beginDateTime,endDateTime);
list.add(travelCarInfo);
}
// list存的是一个周期内的当前车道 旅行信息数据 行车情况的明细数据
// 根据一个车道内一个周期的行车情况的明细数据，计算具体的指标值，这里按照流向进行预聚合统计
//对当前list中的数据按照流向进行拆分
Map<String, List<TravelCarInfo>> map = new HashMap<>();
for (TravelCarInfo t : list) {
List<TravelCarInfo> lineList;
if (!map.containsKey(t.getFlowDirection())) {
lineList = new ArrayList<>();
} else {
lineList = map.get(t.getFlowDirection());
}
lineList.add(t);
map.put(t.getFlowDirection(), lineList);
}
// 按照流向分组完成后，对每个流向的数据进行聚合统计操作，并记录数据，准备写入数据库
Set<String> flowDirections = map.keySet();
for (String flowDirection : flowDirections) {
//当前流向 所有的车辆行车情况信息
List<TravelCarInfo> next = map.get(flowDirection);
//在循环一个车道某个流向过程中需要计算的数据
//1、通过时间之和
//2、每辆车通过平均车速之和
//3、控制延误之和
//4、停车次数之和
//5、停车延误之和
//6、车头时距之和
//7、最后一辆车进入路口的时间（即inCrossTime最大值，记录此值是为了，后续有迟到数据，计算车头时距的时候用的）
Long sumPassTime = 0L;
double sumSpeed = 0.0;
Long sumControlDelay = 0L;
Integer sumStopTimes = 0;
Long sumStopDelay = 0L;
Long sumCarHeadTimeGap = 0L;
Long maxInCrossTime = 0L;
for (TravelCarInfo t : next) {
sumPassTime = sumPassTime + t.getPassTime();
sumSpeed = sumSpeed + t.getAverageSpeed();
sumControlDelay = sumControlDelay + t.getControlDelayTime();
sumStopTimes = sumStopTimes + t.getStopTimes();
sumStopDelay = sumStopDelay + t.getStopDelayTime();
sumCarHeadTimeGap = sumCarHeadTimeGap + t.getCarHeadTimeGap();
if (t.getInCrossTime() > maxInCrossTime) {
maxInCrossTime = t.getInCrossTime();
}
}
TravelLineSinkInfo travelLineSinkInfo = new TravelLineSinkInfo();
travelLineSinkInfo.setCross_id(crossCode);
travelLineSinkInfo.setCycle_id(cycleOrder);
travelLineSinkInfo.setDirection(next.get(0).getDirection());
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.setLast_car_inCross_time(maxInCrossTime);
travelLineSinkInfo.setCycle_begin_time(beginDateTime);
travelLineSinkInfo.setCycle_end_time(endDateTime);
sinkList.add(travelLineSinkInfo);
}
}
sinkTravelLineSinkInfoToMysql(sinkList);
}
/**
* 过滤出旅行信息状态中，属于这个周期内的计算数据，返回，对于已经过期的数据进行清理
* @param beginDateTime
* @param endDateTime
* @return
*/
private Map<String,List<TravelInfo>> filterThisCycleData(String crossCode,Long beginDateTime ,Long endDateTime) throws Exception {
Map<String, List<TravelInfo>> stringListMap = travelInfoState.get(crossCode);
Map<String, List<TravelInfo>> thisCycleData = new HashMap<>();
Map<String, List<TravelInfo>> leftCycleData = new HashMap<>();
if (stringListMap != null) {
//所有的车道数据
Set<String> laneIds = stringListMap.keySet();
for (String laneId : laneIds) {
List<TravelInfo> allTravelInfos = stringListMap.get(laneId);
List<TravelInfo> leftData = new ArrayList<>(); //剩下的数据，不在，当前周期的，属于下个周期的
List<TravelInfo> thisData = new ArrayList<>(); //属于当前周期的数据
for (TravelInfo t : allTravelInfos) {
//进入路口的时间属于这个周期
if (t.getInCrossTime() >= beginDateTime && t.getInCrossTime() <= endDateTime) {
//当前周期用到的数据
thisData.add(t);
//如果，进入路口时距小于周期开始时间，且已经超过30min，则清理这个数据
} else if(t.getInCrossTime()<beginDateTime && beginDateTime - t.getInCrossTime() > 1000*60*30 ) {
System.out.println("移除当前旅行信息-超过30分钟都没有触发计算"+t);
} else {
//当前周期没有用的的数据
leftData.add(t);
}
}
thisCycleData.put(laneId, thisData);
leftCycleData.put(laneId,leftData);
//更新旅行信息状态中的数据，只保留没有使用的旅行信息数据
travelInfoState.put(crossCode, leftCycleData);
return thisCycleData;
}
}
return null;
}
private List<TravelInfo> orderByInCrossTime(List<TravelInfo> travelInfos) {
List<TravelInfo> travelInfoList = new ArrayList<>();
Set<TravelInfo> set = new TreeSet<>();
for (TravelInfo t:travelInfos) {
set.add(t);
}
travelInfoList.addAll(set);
return travelInfoList ;
}
/**
* 将旅行信息转换为行车情况明细信息数据
*
* @param travel
* @return
*/
private TravelCarInfo trunTravelAndCycleToTravelCarInfo(TravelInfo travel, Integer cycleOrder, Long beforeCarInCrossTime,Long beginDateTime ,Long endDateTime) throws Exception {
//1、计算进入时间
long arriveTime = travel.getInCrossTime() - travel.getArrivedTime();
//2、获取方向
String direction = dim_cnt_cross_lane_position.get(travel.getCrossId()).get(travel.getInCrossLineId()+"direction");
//2.1 获取流向
String flow_direction = getFlowDirection(travel) ;
//3、获取平均速度 只看在旅行开始时间和旅行结束时间之间的数据
Double averageSpeed = getAverageSpeed(travel.getLocations(),travel.getTravelBeginTime(),travel.getTravelEndTime());
//5、计算停车次数(这里使用停车次数，非实际停车次数)
Integer stopTimes = getStopTimes(travel.getLocations(),beginDateTime,endDateTime)==0?0:1;
//6、计算停车延误时间
Long stopDelayTime = getStopDelayTime(travel.getLocations());
//4、控制延迟时间 暂时不算,使用停车延误时间
Long controlDelayTime = -stopDelayTime;
//7、计算车头时距 要分车道
Long carHeadTimeGap;
if (beforeCarInCrossTime == 0L) {
carHeadTimeGap = 0L;
} else {
carHeadTimeGap = travel.getInCrossTime() - beforeCarInCrossTime;
}
TravelCarInfo travelCarInfo = new TravelCarInfo(travel.getCrossId(), cycleOrder,
travel.getCarId()
, arriveTime,travel.getInCrossTime(), travel.getInCrossLineId(), direction, flow_direction,averageSpeed,
controlDelayTime, stopTimes, stopDelayTime, carHeadTimeGap);
return travelCarInfo;
}
/**
* 查询车道维表数据-数据存在map中，key是路口编号，
*
*
* @param connection
* @return
* @throws SQLException
*/
private Map<String, Map<String, String>> getLaneDimData(DruidPooledConnection connection) throws SQLException {
String sql = "select cross_id, lane_id,position,concat(position , name ) as flow_direction from test.dim_cnt_cross_lane_position ";
PreparedStatement ps = connection.prepareStatement(sql);
ResultSet resultSet = ps.executeQuery();
Map<String, Map<String, String>> dimData = new HashMap<>();
while (resultSet.next()) {
//当前这个路口的车道维度数据不存在
if (dimData.get(resultSet.getString("cross_id")) == null) {
Map<String, String> map = new HashMap<>();
map.put(resultSet.getString("lane_id")+"direction", resultSet.getString("position"));
map.put(resultSet.getString("lane_id")+"flow_direction", resultSet.getString("flow_direction"));
dimData.put(resultSet.getString("cross_id"), map);
} else {
Map<String, String> map = dimData.get(resultSet.getString("cross_id"));
map.put(resultSet.getString("lane_id")+"direction", resultSet.getString("position"));
map.put(resultSet.getString("lane_id")+"flow_direction", resultSet.getString("flow_direction"));
dimData.put(resultSet.getString("cross_id"), map);
}
}
return dimData;
}
/**
* 计算平均速度，传入一个list集合，计算每个元素中速度的平均值
* 只看旅行期间的平均速度，（照片时间在旅行开始和结束之间）
* @param locations
* @return
*/
private Double getAverageSpeed(List<Location> locations, Long travelBeginTime, Long travelEndTime) {
Double sumSpeed = 0.0;
if(travelBeginTime!=null && travelEndTime!=null) {
int i =0 ;
for (Location location : locations) {
if(location.getDtTranjectory()>=travelBeginTime && location.getDtTranjectory()<=travelEndTime) {
sumSpeed = location.getSpeed() + sumSpeed;
i++;
}
}
return sumSpeed / i;
//如果，没有旅行开始和结束时间，就算全部的
} else {
for (Location location : locations) {
sumSpeed = location.getSpeed() + sumSpeed;
}
return sumSpeed / locations.size();
}
}
/**
* 计算停车次数 根据传入的locations集合，通过按序排列每帧信息，获取到停车次数
* 根据捕获时间排序，从小到大
* 这里求的是实际停车次数
* @param locations
* @return
*/
private Integer getStopTimes(List<Location> locations,Long beginDateTime ,Long endDateTime) throws Exception {
//周期的开始时间
Iterator<Long> iterator1 = cycleSectionState.get().iterator();
Set<Long> set0 = new TreeSet<>();
List<Long> beginTimes ; // 最近3个周期开始时间
while(iterator1.hasNext()){
set0.add(iterator1.next());
}
List<Long> list = new ArrayList<>();
for (Long l:set0) {
list.add(l);
}
if(list.size() > 3) {
beginTimes = list.subList(list.size() - 3, list.size());
cycleSectionState.update(beginTimes);
} else {
beginTimes = list;
}
// 根据最近的几个周期的开始时间对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) {
realStopTimes++;
//从第二帧开始，只有前一帧速度大于10，且这一帧数据小于3的才算一次停车，连续小于3的视为一次停车
} else {
if (speedList.get(i).getSpeed() <= 3.0 && speedList.get(i - 1).getSpeed() > 10.0) {
realStopTimes++;
}
}
}
return stopTimes;
}
/**
* 计算停车延误时间 根据传入的locations集合，通过按序排列每帧信息，计算正常行驶开始时间置上一个停止时间的时间差，进行累计即为累计的停车延误时间
* 根据捕获时间排序，从小到大
*
* @param locations
* @return
*/
private Long getStopDelayTime(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());
}
Long stopDelayTime = 0L;
Long stopStart = 0L;
for (int i = 0; i < speedList.size(); i++) {
//第一帧数据小于3视为停车
if (i == 0 && speedList.get(i).getSpeed() < 3.0) {
stopStart = speedList.get(i).getDtTranjectory();
//从第二帧开始，只有前一帧速度大于10，且这一帧数据小于10的才算一次停车，连续小于3的视为一次停车
} else {
//发生停车事件
if (speedList.get(i).getSpeed() < 3.0 && speedList.get(i - 1).getSpeed() >= 10.0) {
stopStart = speedList.get(i).getDtTranjectory();
}
//开始正常行驶
else {
if (stopStart != 0L && speedList.get(i).getSpeed() >= 3.0) {
stopDelayTime = (speedList.get(i).getDtTranjectory() - stopStart) + stopDelayTime;
//开始正常行驶了，把这个停止起始时间置为0L
stopStart = 0L;
}
// 如果，是最后一帧数据并且是停止状态，且前面也是停止状态，则进行计算延迟时间
if (i == speedList.size() - 1 && speedList.get(i).getSpeed() < 3.0 && stopStart != 0L) {
stopDelayTime = (speedList.get(i).getDtTranjectory() - stopStart) + stopDelayTime;
}
}
}
}
return stopDelayTime;
}
/**
* 根据进入路口车道，和驶出路口车道结合起来，判断具体的流向
* 根据进入路口的方法向和驶出路口的方向，判断流向
* @return
*/
private String getFlowDirection(TravelInfo travel){
//根据车道id，查询方向
String inLineId = travel.getInCrossLineId()!=null?travel.getInCrossLineId():travel.getArrivedLineId();
String outLineId = travel.getOutCrossLineId()!=null?travel.getOutCrossLineId():travel.getAwayLineId();
String inDirection = dim_cnt_cross_lane_position.get(travel.getCrossId()).get(inLineId+"direction");
String outDirection = dim_cnt_cross_lane_position.get(travel.getCrossId()).get(outLineId+"direction");
if(inDirection!=null&&inDirection.equals(outDirection)){
return inDirection+"掉头";
} else {
switch (inDirection){
case "东":
if(outDirection.equals("南")){return "东左转";}
if(outDirection.equals("西")){return "东直行";}
if(outDirection.equals("北")){return "东右转";}
case "南":
if(outDirection.equals("东")){return "南右转";}
if(outDirection.equals("西")){return "南左转";}
if(outDirection.equals("北")){return "南直行";}
case "西":
if(outDirection.equals("东")){return "西直行";}
if(outDirection.equals("南")){return "西右转";}
if(outDirection.equals("北")){return "西左转";}
case "北":
if(outDirection.equals("东")){return "北左转";}
if(outDirection.equals("南")){return "北直行";}
if(outDirection.equals("西")){return "北右转";}
}
return "未知";
}
}
/**
* 将每个车道流向预聚合的数据存入mysql中去
* @param list
* @throws SQLException
*/
private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
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 :list) {
// 使用事件数据的时间划分数据分区，这样可以进行补数操作
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();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelEventAndCycleCoProcessFunction.java
......@@ -38,9 +38,10 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
private MapState<String, Map<String, Map<String,List<TravelEvent>>>> leftDataEventState; //缓存当前周期没有进入路口的车辆的旅行事件
private MapState<String, TravelEvent> outCrossEventState; //缓存驶出路口的旅行事件
private MapState<Integer, CycleSignalData> cycleSignalDataState; //缓存周期数据（避免周期数据先到，而旅行数数据后到的情况）
private MapState<Integer, CycleSignalData> alreadyHandleCycleSignalDataState; //已经处理过的周期数据，避免重复触发
private MapState<String, Long> carRecordState; //记录车辆的 车辆id --> 进入时间 （后面，根据这个时间，和车辆id，移除过期车辆的数据
private DruidDataSource dataSource;
// private DruidDataSource dataSource;
private transient DruidPooledConnection connection;
private static Map<String, Map<String, String>> dim_cnt_cross_lane_position;
......@@ -64,6 +65,10 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
new MapStateDescriptor("cycleSignalDataState",
TypeInformation.of(Integer.class), TypeInformation.of(CycleSignalData.class));
MapStateDescriptor<Integer, CycleSignalData> alreadyHandleCycleSignalDataStateDescriptor =
new MapStateDescriptor("alreadyHandleCycleSignalDataState",
TypeInformation.of(Integer.class), TypeInformation.of(CycleSignalData.class));
MapStateDescriptor<String, Long> carRecordStateDescriptor =
new MapStateDescriptor("carRecordState",
TypeInformation.of(String.class), TypeInformation.of(Long.class));
......@@ -78,36 +83,58 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
// 过期对象的清理策略 全量清理
.cleanupFullSnapshot()
.build();
carRecordStateDescriptor.enableTimeToLive(stateTtlConfig);
carRecordStateDescriptor.enableTimeToLive(stateTtlConfig); //设置车辆过期时间
StateTtlConfig stateTtlConfig2 = StateTtlConfig
// 状态有效时间 300min
.newBuilder(Time.minutes(300))
// 设置状态的更新类型
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
// 已过期还未被清理掉的状态数据不返回给用户
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
// 过期对象的清理策略 全量清理
.cleanupFullSnapshot()
.build();
alreadyHandleCycleSignalDataStateDescriptor.enableTimeToLive(stateTtlConfig2); //设置已处理周期过期时间
travelEventState = getRuntimeContext().getMapState(travelEventStateDescriptor);
leftDataEventState = getRuntimeContext().getMapState(leftDataEventStateDescriptor);
outCrossEventState = getRuntimeContext().getMapState(outCrossEventStateDescriptor);
cycleSignalDataState = getRuntimeContext().getMapState(cycleSignalDataStateDescriptor);
alreadyHandleCycleSignalDataState = getRuntimeContext().getMapState(alreadyHandleCycleSignalDataStateDescriptor);
carRecordState = getRuntimeContext().getMapState(carRecordStateDescriptor);
connection = DruidConnectPoolUtils.getConnection();
// ("jdbc:mysql://localhost:9030/demo?characterEncoding=utf8",
/* dataSource = DruidConnectPoolUtils.getDataSource
("jdbc:mysql://10.243.0.22:9030/qxlk?characterEncoding=utf8",
// ("jdbc:mysql://localhost:9030/qxlk?characterEncoding=utf8",
// ("jdbc:mysql://localhost:3307/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",
// "root123");
"mima");
*/
// dataSource.getConnection();
connection = dataSource.getConnection();*/
connection=DruidConnectPoolUtils.getConnection();
dim_cnt_cross_lane_position = getLaneDimData(connection);
}
@Override
public void close() throws Exception {
super.close();
if(connection!=null){
connection.close();
}
}
//把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
@Override
public void processElement1(TravelEvent value, CoProcessFunction<TravelEvent, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
System.out.println("处理的数据有"+value.getEventType());
// System.out.println("处理的数据有"+value.getEventType());
//记录当前车辆的进入路口的时间，车辆id --> 进入时间 （后面，根据这个时间，和车辆id，移除过期车辆的数据）
if("INCROSS".equals(value.getEventType())){
......@@ -192,6 +219,12 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
public void processElement2(CycleSignalData value, CoProcessFunction<TravelEvent, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
//判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
if(alreadyHandleCycleSignalDataState.contains(value.getCycleOrder())){
System.out.println("周期数据重复----当前路口id是"+value.getCrossCode()+"当前周期已经计算完成了"+value.getCycleOrder());
return;
}
// 注册定时器
//获取当前的ProcessingTime
long currentProcessingTime = ctx.timerService().currentProcessingTime();
......@@ -210,7 +243,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
List<Integer> cycles = new ArrayList<>();
while(iterator.hasNext()){
CycleSignalData next = iterator.next();
System.out.println("当前处理周期"+next.getCycleOrder()+"<---->路口"+next.getCrossCode());
// System.out.println("当前处理周期"+next.getCycleOrder()+"<---->路口"+next.getCrossCode());
handleTwoStream(next);
cycles.add(next.getCycleOrder());
}
......@@ -222,14 +255,13 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
// 如果，周期数据先到，旅行事件数据后到的话，把周期数据暂存到状态中去，
} else {
System.out.println("路口暂无旅行事件信息-等待下次触发");
// System.out.println(value.getCrossCode()+"路口暂无旅行事件信息-等待下次触发");
cycleSignalDataState.put(cycleOrder,value);
}
}
private void handleTwoStream(CycleSignalData value) throws Exception {
System.out.println("处理的数据有" + value.getCrossCode()+"--->"+value.getCycleOrder());
String crossCode = value.getCrossCode();
Integer cycleOrder = value.getCycleOrder();
......@@ -240,8 +272,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
//travelEventState 是一个map结构，key是每一个路口id，根据路口id，获取到当前路口的所有的数据
Map<String, Map<String, List<TravelEvent>>> crossMap = travelEventState.get(crossCode);
// System.out.println("旅行事件数据有：" + crossMap);
//
Map<String, Map<String, List<TravelEvent>>> leftDataMap = leftDataEventState.get(crossCode);
// System.out.println("上个周期剩余未计算的数据有：" + leftDataMap);
......@@ -269,10 +299,13 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
}
//如果，一条数据都没sink，说明周期数据先来的，事件数据迟到了
if(sinkCounts==0){
System.out.println("当前路口"+crossCode+"当前周期，没有sink的数据"+cycleOrder);
// cycleSignalDataState.put(cycleOrder,value);
// System.out.println("当前路口"+crossCode+"当前周期，没有sink的数据"+cycleOrder);
cycleSignalDataState.put(cycleOrder,value);
}
// 当前周期已经处理完成了，，kafka中再次来了该周期的数据不需要再次触发计算了
alreadyHandleCycleSignalDataState.put(cycleOrder,value);
}
}
......@@ -289,7 +322,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
super.onTimer(timestamp, ctx, out);
System.out.println("触发定时器了。。。。");
// System.out.println("触发定时器了。。。。");
Iterator<String> carIds = carRecordState.keys().iterator();
Iterator<Long> eventTimes = carRecordState.values().iterator();
// 获取到最新的的到时间，以便作为参考依据移除过期数据，这里不使用系统的或者服务器的时间的原因，是避免服务器时间与外界数据时间不一致的情况
......@@ -322,11 +355,18 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
//清空当前车辆的数据
for (String carId:needRemoveCars) {
next1.remove(carId);
System.out.println("移除当前车辆数据"+carId);
}
}
}
//更新维表数据 只在每天12点的时候更新
long l = System.currentTimeMillis();
SimpleDateFormat sdf = new SimpleDateFormat("HH");
if(sdf.format(l).equals("12")){
dim_cnt_cross_lane_position = getLaneDimData(connection);
}
}
/**
......@@ -353,8 +393,8 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
Map<String, List<TravelEvent>> leftLineData;
if(leftDataCross!=null){
leftDataCross = leftDataEventState.get(crossCode);
System.out.println("当前路口是"+crossCode+"当前车道id是"+lineId);
System.out.println("当前路口,当前车道存在未计算的数据有"+leftDataCross.get(lineId));
/* System.out.println("当前路口是"+crossCode+"当前车道id是"+lineId);
System.out.println("当前路口,当前车道存在未计算的数据有"+leftDataCross.get(lineId));*/
leftLineData = leftDataCross.get(lineId);
} else {
leftLineData = new HashMap<>();
......@@ -407,7 +447,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
// ，后面触发计算时，进行合并到新来的数据暂存状态中去）
//TODO 处理无法完成计算的数据 事件没有到齐的情况
if(!allEventOfCar.containsKey("INCROSS")||!allEventOfCar.containsKey("ARRIVED")){
System.out.println("当前数据不完整，无法计算。。。。。"+carId);
// System.out.println("当前数据不完整，无法计算。。。。。"+carId);
//如果，当前车道 存在，不能完成计算的数据 （数据不全，没有进入路口的数据，则等待下次触发计算）
//获取当前车道
String eventLineId = allEventOfCar.values().iterator().next().getEventLineId();
......@@ -433,8 +473,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
continue;
}
System.out.println("当前车辆所有事件数据有：==========》\r\n"+allEventOfCar);
//1、通过时间
Long passTime = allEventOfCar.get("INCROSS").getEventTime()-allEventOfCar.get("ARRIVED").getEventTime();
......@@ -506,12 +544,8 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
List<TravelLineSinkInfo> sinkList = new ArrayList<>();
for (String flowDirection:flowDirectionSet) {
System.out.println("当前数据流向是------》"+flowDirection);
//一个流向上每个车辆整合后的数据
Set<TravelCarInfo> travelCarInfosOfOneFlow = flowMap.get(flowDirection);
// sum
//在循环一个车道某个流向过程中需要计算的数据
//1、通过时间之和
......@@ -580,9 +614,8 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
}
Map<String, Map<String, List<TravelEvent>>> leftDataMap = leftDataEventState.get(crossCode);
System.out.println("这个周期剩余未计算的数据有："+leftDataMap);
// Map<String, Map<String, List<TravelEvent>>> leftDataMap = leftDataEventState.get(crossCode);
// System.out.println("这个周期剩余未计算的数据有："+leftDataMap);
return sinkList ;
}
......@@ -601,8 +634,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
return map;
}
//SELECT id, id_old, lane_id, cross_id, name, out_in_type, `position`, out_in_name, lane_length
//FROM test.dim_cnt_cross_lane_position;
......@@ -634,7 +665,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
dimData.put(resultSet.getString("cross_id"), map);
}
}
connection.close();
// connection.close();
return dimData;
}
......@@ -665,8 +696,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
* @return
*/
private Long getStopDelayTime(List<TravelEvent> travelEvents) {
//按照事件发送的事件顺序进行排序后，再计算停车延误时间
Set<TravelEvent> set = new TreeSet<>();
for (TravelEvent t:travelEvents) {
......@@ -692,8 +721,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
* @return
*/
private String getFlowDirection(Map<String, TravelEvent> allEventOfCar) throws Exception {
// 进来的数据都有INCROSS
TravelEvent inCross = allEventOfCar.get("INCROSS");
// 驶出路口的数据，单独存放在一个状态存储中
......@@ -701,7 +728,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
// 当前车辆存在驶出路口的事件
if(outCross!=null){
//根据车道id，查询方向
String inLineId = inCross.getEventLineId()!=null?inCross.getEventLineId():inCross.getEventLineId();
String outLineId = outCross.getEventLineId()!=null?outCross.getEventLineId():outCross.getEventLineId();
String inDirection = dim_cnt_cross_lane_position.get(inCross.getCrossId()).get(inLineId+"direction");
......@@ -754,9 +780,8 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
*/
private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
String sql = "INSERT INTO demo.app_cross_line_travel_car_info" +
String sql = "INSERT INTO qxlk.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, " +
......@@ -770,10 +795,11 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
String statistic_time = sdf.format(date);
String record_date = sdf2.format(date);
for (TravelLineSinkInfo t :list) {
// 使用事件数据的时间划分数据分区，这样可以进行补数操作
String record_date = sdf2.format(t.getLast_car_inCross_time());
preparedStatement.setString(1, record_date);
preparedStatement.setString(2, t.getCross_id());
......@@ -792,12 +818,10 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
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();
// connection.close();
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelSituationAnalysisJob.java
......@@ -2,9 +2,12 @@ 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.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 org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
......@@ -14,8 +17,9 @@ import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
......@@ -26,7 +30,6 @@ import java.util.Properties;
* @create 2022-11-14 11:43
**/
public class TravelSituationAnalysisJob {
public static void main(String[] args) {
try {
......@@ -46,7 +49,7 @@ public class TravelSituationAnalysisJob {
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//设置重启策略
env.setRestartStrategy( RestartStrategies.fixedDelayRestart(5, Time.seconds(10)));
env.setRestartStrategy( RestartStrategies.fixedDelayRestart(1, Time.seconds(10)));
/*
关于flink checkpoint存储的设置统一在集群配置文件中设置
......@@ -55,52 +58,57 @@ public class TravelSituationAnalysisJob {
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("bootstrap.servers", "dn1.zhht:9092,dn2.zhht:9092,dn3.zhht:9092");
// kafkaProperties.setProperty("bootstrap.servers", "172.25.1.251:9092,172.25.1.122:9092,172.25.1.67: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", "dn1.zhht:9092,dn2.zhht:9092,dn3.zhht:9092");
// kafkaProperties2.setProperty("bootstrap.servers", "172.25.1.251:9092,172.25.1.122:9092,172.25.1.67:9092");
kafkaProperties2.setProperty("group.id", "cycle_signal");
// 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");
//旅行事件
FlinkKafkaConsumerBase<TravelEvent> travelEvent =
/* FlinkKafkaConsumer<TravelEvent> travelEvent =
new FlinkKafkaConsumer<TravelEvent>("trips_event_info",
new TravelEventKafkaSchema(),
kafkaProperties) {
};
DataStreamSource<TravelEvent> travelEventStream = env.addSource(travelEvent);
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);
DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
travelEventStream.print();
travelInfoStream.print();
//周期信号
FlinkKafkaConsumer<CycleSignalData> cycleSignalData =
new FlinkKafkaConsumer<>("signal_cycle_data",
new CycleSignalKafkaSchema(),
kafkaProperties2);
// FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
cycleSignalDataStream.print();
//2、使用connect，把2个流联合处理
ConnectedStreams<TravelEvent, CycleSignalData> connect = travelEventStream
ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
.connect(cycleSignalDataStream);
ConnectedStreams<TravelEvent, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
(KeySelector<TravelEvent, String>) travelInfo1 -> travelInfo1.getCrossId(),
ConnectedStreams<TravelInfo, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
(KeySelector<TravelInfo, String>) travelInfo1 -> travelInfo1.getCrossId(),
(KeySelector<CycleSignalData, String>) cycleSignalData1 -> cycleSignalData1.getCrossCode()
);
travelInfoCycleSignalDataConnectedStreams.process(new TravelEventAndCycleCoProcessFunction());
travelInfoCycleSignalDataConnectedStreams.process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
env.execute();
env.execute("TravelSituationAnalysisJob");
} 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