Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 10121561 by Ruidaimeng
Options

2022-11-30 新增行车情况指标开发任务

parent 0d48bc8f
Showing with 1369 additions and 2 deletions
realtime/hologram-streaming/pom.xml
...@@ -79,6 +79,46 @@ ...@@ -79,6 +79,46 @@
<artifactId>flink-table-api-java-bridge_2.11</artifactId> <artifactId>flink-table-api-java-bridge_2.11</artifactId>
<version>1.12.0</version> <version>1.12.0</version>
</dependency> </dependency>
<!-- 导入 druid 的 jar 包 -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.2.8</version>
</dependency>
</dependencies> </dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project> </project>
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/CycleSignalData.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.Data;
import java.util.List;
/**
* 信号数据
*
* @author ruimeng
* @create 2022-11-14 13:33
**/
@Data
public class CycleSignalData {
private Integer id ;//必填 ID
private String crossCode ;//必填 路口ID
private String signalCode ;//必填 信控机编号
private String beginControlModeCode ;//必填 开始模式编码
private String endControlModeCode ;//必填 结束模式编码
private Integer beginPlanId ;//可选 开始方案ID
private Integer endPlanId ;//可选 结束方案ID
private Long beginDateTime ;//必填 周期开始时间,毫秒时间戳
private Long endDateTime ;//必填 周期结束时间,毫秒时间戳
private Integer duration ;//必填 周期时长，单位 秒
private Integer cycleOrder ;//必填 周期顺序
private List<StageInfo> stageList ;//必填 相位列表，参考 相位
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/Location.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
/**
* 位置信息
*
* @author ruimeng
* @create 2022-11-14 14:50
**/
@Data
@AllArgsConstructor
public class Location implements Comparable<Location>{
private Double longitude ;//可选 经度
private Double latitude ;//可选 纬度
private Double speed ;//必填 时速，单位km/h
private Long dtTranjectory ;//必填 捕获时间，事件发生的时间，毫秒级别时间戳
private String lineId ;//可选 车道ID
@Override
public int compareTo(Location o) {
return (int)(dtTranjectory-o.getDtTranjectory());
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/StageInfo.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.Data;
/**
* 相位信息
*
* @author ruimeng
* @create 2022-11-14 14:59
**/
@Data
public class StageInfo {
private Integer id ;//必填 ID
private Integer phaseValue ;//必填 相位值
private Long beginDateTime ;//必填 开始时间,毫秒时间戳
private Long endDateTime ;//必填 结束时间,毫秒时间戳
private String beginControlModeCode ;//必填 开始模式编码
private String endControlModeCode ;//必填 结束模式编码
private Integer beginPlanId ;//可选 开始方案ID
private Integer endPlanId ;//可选 结束方案ID
private Integer duration ;//必填 相位时长，单位秒 s
private Integer green ;//必填 绿灯时长，单位秒 s
private Integer yellow ;//必填 黄灯时长，单位秒 s
private Integer allRed ;//必填 全红时长， 单位秒 s
private Integer phaseOrder ;//必填 相位序号
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelCarInfo.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 一辆车的行车情况信息
* 作为明细数据写入
* @author ruimeng
* @create 2022-11-14 17:09
**/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class TravelCarInfo implements Comparable<TravelCarInfo> {
private String crossId ;//必填 路口ID
private Integer cycleOrder ;//必填 周期顺序
private String carId ;//可选 车辆ID
private Long passTime ;//通过时间 =进入时间戳-到达时间戳
private Long inCrossTime ;//进入路口时间
private String inCrossLineId ;//可选 进入路口时车道编号
private String direction ; // 方向，根据路口id和进入路口时车道编号查询方向
private String flowDirection ; // 方向，根据路口id和进入路口时车道编号查询方向
private Double averageSpeed ; //平均速度 （根据locations的每帧图片数据计算平均速度）
private Long controlDelayTime ; //控制延迟时间 -> 暂时不计算，因为车辆的自由流入时间暂时获取不到
private Integer stopTimes ; //停车次数
private Long stopDelayTime ; //停车延迟时间 根据locations中的数据进行计算，locations是一个list，每100ms一个数据，当这个数据中的speed小于3km时，即视为停车
private Long carHeadTimeGap ; //车头时距 相邻2辆车到达同一观测线的时间差，第一辆车数据没有，后面每来一辆车，都要基于前一辆车的数据进行计算，因此，需要保存状态
@Override
public int compareTo(TravelCarInfo o) {
return (int)(inCrossTime-o.getInCrossTime());
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelEvent.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.Data;
/**
* 旅行事件
*
* @author ruimeng
* @create 2022-11-18 11:12
**/
@Data
public class TravelEvent implements Comparable<TravelEvent>{
private Integer id ;//必填 ID
private String crossId ;//必填 路口ID
private String recordId ;//必填 记录ID
private Long recordTime ;//必填 记录时间，毫秒级别时间戳
private String eventLineId ;//必填 车道编号
private Long eventTime ;//必填 事件时间，毫秒级别时间戳
private Double eventSpeed ;//必填 事件发生时速度，单位km/h
private String eventType ;//必填 事件类型
private String remark ;//可选 备注
@Override
public int compareTo(TravelEvent o) {
return (int)(eventTime-o.getEventTime());
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelInfo.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.Data;
import java.util.List;
/**
* 旅行信息
*
* @author ruimeng
* @create 2022-11-14 13:33{
**/
@Data
public class TravelInfo {
private Integer id ;//必填 ID
private String crossId ;//必填 路口ID
private String recordId ;//必填 记录ID
private Long recordTime ;//必填 记录时间（该记录产生的时间）毫秒级别时间戳
private Long travelBeginTime ;//必填 旅行开始时刻（该记录产生的时间）毫秒级别时间戳
private Long travelEndTime ;//必填 旅行结束时刻（该记录产生的时间）毫秒级别时间戳
private String carId ;//可选 车辆ID
private String carColor ;//可选 车辆颜色
private String plate ;//可选 车牌号码
private String plateColor ;//可选 车牌颜色，参考字典表
private Integer carType ;//可选 车辆类型, 参考字典表
private String firstLineId ;//必填 捕获时车道编号
private Long firstTime ;//必填 捕获时间，毫秒级别时间戳
private Double firstSpeed ;//必填 捕获时速度，单位km/h
private String arrivedLineId ;//可选 到达时车道编号
private Long arrivedTime ;//可选 到达时间，毫秒级别时间戳
private Double arrivedSpeed ;//可选 到达时速度，单位km/h
private String inCrossLineId ;//可选 进入路口时车道编号
private Long inCrossTime ;//可选 进入路口时间，毫秒级别时间戳
private Double inSpeed ;//可选 进入时速度，单位km/h
private String outCrossLineId ;//可选 通过路口时车道编号
private Long outCrossTime ;//可选 通过路口时间，毫秒级别时间戳
private Double outSpeed ;//可选 通过时速度，单位km/h
private String awayLineId ;//可选 驶离时车道编号
private Long awayTime ;//可选 驶离路口时间，毫秒级别时间戳
private Double awaySpeed ;//可选 驶离时速度，单位km/h
private String lastLineId ;//必填 丢失时车道编号
private Long lastTime ;//必填 丢失时间，毫秒级别时间戳
private Double lastSpeed ;//必填 丢失时速度，单位km/h
private Double crossingTime ;//可选 旅行时长，单位 秒 (s)（驶离时刻-到达时刻）
private List<Location> locations ;//必填 数组，参考 位置信息字段
private String remark ;//可选 备注
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelLineSinkInfo.java 0 → 100644
package com.zhht.irn.entity.dto;
import lombok.Data;
/**
* 行车情况按车道流向预聚合信息
*
* @author ruimeng
* @create 2022-11-17 14:11
**/
@Data
public class TravelLineSinkInfo {
private String cross_id ;//varchar(64) NOT NULL COMMENT '路口编号',
private String lane_id ;//varchar(64) DEFAULT NULL COMMENT '车道编号',
private Integer cycle_id ;//varchar(64) DEFAULT NULL COMMENT '周期编号',
private String direction;// varchar(16) DEFAULT NULL COMMENT '方向',
private String flow_direction ;// varchar(64) DEFAULT NULL COMMENT '流向',
private Integer pass_numbers;// int(32) DEFAULT NULL COMMENT '通过车辆数',
private Long last_car_inCross_time;// long DEFAULT NULL COMMENT '最后一辆车进入路口时间,(用于处理迟到数据的车头时距的计算)',
private Double average_pass_time ;//double DEFAULT NULL COMMENT '平均通过时间',
private Double average_pass_speed;// double DEFAULT NULL COMMENT '平均车速',
private Double average_control_delay ;//double DEFAULT NULL COMMENT '平均控制延误',
private Double average_stop_delay ;//double DEFAULT NULL COMMENT '平均停车延误',
private Double average_stop_times ;// int DEFAULT NULL COMMENT '平均停车次数',
private Double average_car_head_time_gap;// int DEFAULT NULL COMMENT '平均车头时距'
private Long cycle_begin_time; // 周期开始时间
private Long cycle_end_time; // 周期结束时间
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelEventAndCycleCoProcessFunction.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.CycleSignalData;
import com.zhht.irn.entity.dto.TravelCarInfo;
import com.zhht.irn.entity.dto.TravelEvent;
import com.zhht.irn.entity.dto.TravelLineSinkInfo;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
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后处理
* 处理旅行事件topic
* 周期信号topic
* @author ruimeng
* @create 2022-11-18 14:16
**/
public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<TravelEvent, CycleSignalData, Object> {
//定义状态，用于存储旅行信息，当周期信号数据过来的时候，触发计算
//状态数据存储结构为 Map（路口id--》（Map（车道id-》 Map（车辆id--》List（旅行事件）））））
// 这样就把每个路口的旅行事件数据进行分流，同时，对每个路口的数据，也按照了不同的车道进行了分流，同一个路口，同一个车道的数据存在一个车道信息记录中
private MapState<String, Map<String, Map<String,List<TravelEvent>>>> travelEventState; //缓存旅行事件的数据
private MapState<String, Map<String, Map<String,List<TravelEvent>>>> leftDataEventState; //缓存当前周期没有进入路口的车辆的旅行事件
private MapState<String, TravelEvent> outCrossEventState; //缓存驶出路口的旅行事件
private MapState<Integer, CycleSignalData> cycleSignalDataState; //缓存周期数据（避免周期数据先到，而旅行数数据后到的情况）
private MapState<String, Long> carRecordState; //记录车辆的 车辆id --> 进入时间 （后面，根据这个时间，和车辆id，移除过期车辆的数据
private DruidDataSource dataSource;
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,Map<String,List<TravelEvent>>>> travelEventStateDescriptor =
new MapStateDescriptor("travelEventState",
TypeInformation.of(String.class), TypeInformation.of(Map.class));
MapStateDescriptor<String, Map<String,Map<String,List<TravelEvent>>>> leftDataEventStateDescriptor =
new MapStateDescriptor("leftDataEventState",
TypeInformation.of(String.class), TypeInformation.of(Map.class));
MapStateDescriptor<String, TravelEvent> outCrossEventStateDescriptor =
new MapStateDescriptor("outCrossEventState",
TypeInformation.of(String.class), TypeInformation.of(TravelEvent.class));
MapStateDescriptor<Integer, CycleSignalData> cycleSignalDataStateDescriptor =
new MapStateDescriptor("cycleSignalDataState",
TypeInformation.of(Integer.class), TypeInformation.of(CycleSignalData.class));
MapStateDescriptor<String, Long> carRecordStateDescriptor =
new MapStateDescriptor("carRecordState",
TypeInformation.of(String.class), TypeInformation.of(Long.class));
StateTtlConfig stateTtlConfig = StateTtlConfig
// 状态有效时间 15min
.newBuilder(Time.minutes(15))
// 设置状态的更新类型
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
// 已过期还未被清理掉的状态数据不返回给用户
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
// 过期对象的清理策略 全量清理
.cleanupFullSnapshot()
.build();
carRecordStateDescriptor.enableTimeToLive(stateTtlConfig);
travelEventState = getRuntimeContext().getMapState(travelEventStateDescriptor);
leftDataEventState = getRuntimeContext().getMapState(leftDataEventStateDescriptor);
outCrossEventState = getRuntimeContext().getMapState(outCrossEventStateDescriptor);
cycleSignalDataState = getRuntimeContext().getMapState(cycleSignalDataStateDescriptor);
carRecordState = getRuntimeContext().getMapState(carRecordStateDescriptor);
dataSource = DruidConnectPoolUtils.getDataSource
// ("jdbc:mysql://localhost:9030/demo?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",
"root",
// "root123");
"mima");
connection = dataSource.getConnection();
dim_cnt_cross_lane_position = getLaneDimData(connection);
}
//把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
@Override
public void processElement1(TravelEvent value, CoProcessFunction<TravelEvent, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
System.out.println("处理的数据有"+value.getEventType());
//记录当前车辆的进入路口的时间，车辆id --> 进入时间 （后面，根据这个时间，和车辆id，移除过期车辆的数据）
if("INCROSS".equals(value.getEventType())){
if(carRecordState.values()==null){
Map<String,Long> map = new HashMap<>();
carRecordState.putAll(map);
}
carRecordState.put(value.getRecordId(), value.getEventTime());
}
// TODO 对于 驶出路口事件单独处理
if("OUTCROSS".equals(value.getEventType())){
if(outCrossEventState.values()==null){
Map<String,TravelEvent> map = new HashMap<>();
outCrossEventState.putAll(map);
}
outCrossEventState.put(value.getRecordId(), value);
}
//对于丢失和驶离的事件不做处理
else if(!"AWAY".equals(value.getEventType())&&!"LAST".equals(value.getEventType())){
String crossId = value.getCrossId();
String lineId = value.getEventLineId();
String carId = value.getRecordId();
Iterator<Map<String, Map<String, List<TravelEvent>>>> iterator = travelEventState.values().iterator();
//如果状态为空，则进行初始化
if (!iterator.hasNext()) {
Map<String, Map<String, Map<String, List<TravelEvent>>>> initMap = new HashMap<>();
travelEventState.putAll(initMap);
}
//MapState<String, Map<String, CarLineTravelEvent>> travelEventState;
//获取当前路口的数据 <Map<路口id，CarLineTravelEvent（String-车道id->(车辆id--》List《旅行事件》)>>）
// （路口的数据）车道id 车辆id 事件列表
Map<String, Map<String, List<TravelEvent>>> crossListMap = travelEventState.get(crossId); //（路口的数据）车道id 车辆id 事件列表
//当前路口没有数据
if (crossListMap == null ) {
Map<String, Map<String, List<TravelEvent>>> crossListMap2 = new HashMap<>();
Map<String, List<TravelEvent>> lineMap = new HashMap<>();
List<TravelEvent> carTravelEventsList = new ArrayList<>();
carTravelEventsList.add(value);
lineMap.put(carId,carTravelEventsList); //当前车道所有车的旅行事件数据，存在一个map里面，key对应的是车辆id
crossListMap2.put(lineId, lineMap);
travelEventState.put(crossId, crossListMap2);
//当前路口有数据，
} else {
//获取当前车道的数据
Map<String, List<TravelEvent>> lineMap = crossListMap.get(lineId);
// 当前车道有数据
if(lineMap!=null){
// 车道 根据车辆id 找到 当前车辆的数据
List<TravelEvent> carTravelEventsList = lineMap.get(carId);
//当前车辆有数据
if(carTravelEventsList!=null){
carTravelEventsList.add(value);
lineMap.put(carId,carTravelEventsList);
crossListMap.put(lineId,lineMap);
travelEventState.put(crossId, crossListMap);
//当前车辆没有数据
} else {
List<TravelEvent> carEventsList =new ArrayList<>();
carEventsList.add(value);
lineMap.put(carId,carEventsList);
crossListMap.put(lineId,lineMap);
travelEventState.put(crossId, crossListMap);
}
//当前车道没有数据
} else {
Map<String, List<TravelEvent>> lineMap2 = new HashMap<>();
List<TravelEvent> carEventsList =new ArrayList<>();
carEventsList.add(value);
lineMap2.put(carId,carEventsList);
crossListMap.put(lineId,lineMap2);
travelEventState.put(crossId, crossListMap);
}
}
}
}
//来一条周期数据，触发这个周期内的旅行数据
@Override
public void processElement2(CycleSignalData value, CoProcessFunction<TravelEvent, CycleSignalData, Object>.Context ctx,
Collector<Object> out) throws Exception {
// 注册定时器
//获取当前的ProcessingTime
long currentProcessingTime = ctx.timerService().currentProcessingTime();
ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+15*60*1000);
String crossCode = value.getCrossCode();
Integer cycleOrder = value.getCycleOrder();
Map<String, Map<String, List<TravelEvent>>> crossMap = travelEventState.get(crossCode);
// System.out.println("旅行事件数据有："+crossMap);
// 只有旅行事件有数据过来了的时候，才进行关联sink
if(crossMap!=null) {
//如果，存在周期数据没有算的，需要计算一下
if(!cycleSignalDataState.isEmpty()){
Iterator<CycleSignalData> iterator = cycleSignalDataState.values().iterator();
List<Integer> cycles = new ArrayList<>();
while(iterator.hasNext()){
CycleSignalData next = iterator.next();
System.out.println("当前处理周期"+next.getCycleOrder()+"<---->路口"+next.getCrossCode());
handleTwoStream(next);
cycles.add(next.getCycleOrder());
}
for (Integer c:cycles) {
cycleSignalDataState.remove(c);
}
}
handleTwoStream(value);
// 如果，周期数据先到，旅行事件数据后到的话，把周期数据暂存到状态中去，
} else {
System.out.println("路口暂无旅行事件信息-等待下次触发");
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();
//周期开始时间，周期结束时间
Long beginDateTime = value.getBeginDateTime();
Long endDateTime = value.getEndDateTime();
//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);
// 只有旅行事件有数据过来了的时候，才进行关联sink
if (crossMap != null) {
int sinkCounts = 0 ;
//该路口 的 所有车道id的集合
Set<String> lineIdSet = crossMap.keySet();
//拿到每一个车道上的数据
for (String lineId : lineIdSet) {
// System.out.println("当前车道id是" + lineId);
//车道上所有的数据
Map<String, List<TravelEvent>> lineMap = crossMap.get(lineId);
System.out.println(lineMap);
//把一个车道的数据，按照流向进行拆分统计，生成sink数据
List<TravelLineSinkInfo> sinkData = getSinkData(lineMap, cycleOrder, crossCode, lineId, beginDateTime, endDateTime);
// System.out.println("准备sink的数据有" + sinkData.size() + "条");
// System.out.println("准备sink的数据有" + sinkData);
if (sinkData.size() > 0) {
sinkTravelLineSinkInfoToMysql(sinkData);
}
sinkCounts = sinkData.size()+ sinkCounts;
}
//如果，一条数据都没sink，说明周期数据先来的，事件数据迟到了
if(sinkCounts==0){
System.out.println("当前路口"+crossCode+"当前周期，没有sink的数据"+cycleOrder);
// cycleSignalDataState.put(cycleOrder,value);
}
}
}
/**
* 注册定时器，用于定期清空状态存储中存储的旅行事件的数据
* @param timestamp
* @param ctx
* @param out
* @throws Exception
*/
@Override
public void onTimer(long timestamp, CoProcessFunction<TravelEvent, CycleSignalData, Object>.OnTimerContext ctx, Collector<Object> out) throws Exception {
super.onTimer(timestamp, ctx, out);
System.out.println("触发定时器了。。。。");
Iterator<String> carIds = carRecordState.keys().iterator();
Iterator<Long> eventTimes = carRecordState.values().iterator();
// 获取到最新的的到时间，以便作为参考依据移除过期数据，这里不使用系统的或者服务器的时间的原因，是避免服务器时间与外界数据时间不一致的情况
Long maxEventTime = 0L;
while(eventTimes.hasNext()){
Long eventTime = eventTimes.next();
if(eventTime>maxEventTime){
maxEventTime= eventTime;
}
}
List<String> needRemoveCars = new ArrayList<>();
while(carIds.hasNext()){
String carId = carIds.next();
Long eventTime = carRecordState.get(carId);
if((maxEventTime-1000*60*5)>eventTime) {
needRemoveCars.add(carId);
}
}
MapState<String, Map<String, Map<String, List<TravelEvent>>>> travelEventState = this.travelEventState;
Iterator<Map<String, Map<String, List<TravelEvent>>>> iterator = travelEventState.values().iterator();
while(iterator.hasNext()){
Map<String, Map<String, List<TravelEvent>>> next = iterator.next();
Iterator<Map<String, List<TravelEvent>>> iterator1 = next.values().iterator();
while(iterator1.hasNext()){
//每一个车道上的所有数据，key是每一辆车的id
Map<String, List<TravelEvent>> next1 = iterator1.next();
//清空当前车辆的数据
for (String carId:needRemoveCars) {
next1.remove(carId);
System.out.println("移除当前车辆数据"+carId);
}
}
}
}
/**
*
* 按照流向作为最小粒度统计
* @param lineMap 传入的参数是 一个车道上的所有数据
* (车辆id ->当前车辆所有的事件信息) 存放的是一个车道上的所有数据
* @return
* FIRST 发现/捕获事件
* ARRIVED 到达事件
* INCROSS 进入事件
* OUTCROSS 通过事件
* AWAY 驶离事件
* LAST 丢失事件
* STOP 停车事件
* STARTUP 启动事件
*
*/
private List<TravelLineSinkInfo> getSinkData(Map<String, List<TravelEvent>> lineMap,Integer cycleOrder ,
String crossCode,String lineId,Long beginDateTime ,Long endDateTime ) throws Exception {
Map<String, Map<String, List<TravelEvent>>> leftDataCross = leftDataEventState.get(crossCode);
Map<String, List<TravelEvent>> leftLineData;
if(leftDataCross!=null){
leftDataCross = leftDataEventState.get(crossCode);
System.out.println("当前路口是"+crossCode+"当前车道id是"+lineId);
System.out.println("当前路口,当前车道存在未计算的数据有"+leftDataCross.get(lineId));
leftLineData = leftDataCross.get(lineId);
} else {
leftLineData = new HashMap<>();
}
Set<String> carIds = lineMap.keySet();
//一辆车在一个车道上发生进入事件后，整合成为一条数据
List<TravelCarInfo> list = new ArrayList<>();
//记录每一辆车id，即其到达的时间
Map<String,Long> carIdArriveTime = new HashMap<>();
TreeMap<Long,String> arriveTimeCarId = new TreeMap<>();
//拿到每一辆车的所有的旅行事件数据
// 需要分开计算
Map<String ,Map<String,List<TravelEvent>>> leftData = new HashMap<>();
for (String carId:carIds) {
// 停车次数
Integer stopCounts ;
// 当前这个周期内来的数据
List<TravelEvent> travelEvents = lineMap.get(carId);
//TODO 结合上，上个周期内没有处理的数据 ，（找到同一辆车的数据即可）
if(leftLineData!=null&&leftLineData.containsKey(carId)){
//上一个周期的所有数据
List<TravelEvent> lastCycleTravelEvents = leftLineData.get(carId);
//上一周期所有事件
Map<String, TravelEvent> lastCycleAllEventOfCar = getAllEventOfCar(lastCycleTravelEvents);
//本周期所有事件
Map<String, TravelEvent> thisCycleAllEventOfCar = getAllEventOfCar(travelEvents);
Integer lastStopTimes = lastCycleAllEventOfCar.containsKey("STOP")?1:0;
Integer thisStopTimes = thisCycleAllEventOfCar.containsKey("STOP")?1:0;
stopCounts = lastStopTimes+thisStopTimes ;
travelEvents.addAll(lastCycleTravelEvents);
//移除 上周期剩余 本周期计算完成的数据
leftDataEventState.get(crossCode).get(lineId).remove(carId);
} else {
//本周期所有事件
Map<String, TravelEvent> thisCycleAllEventOfCar = getAllEventOfCar(travelEvents);
stopCounts = thisCycleAllEventOfCar.containsKey("STOP")?1:0;
}
Map<String, TravelEvent> allEventOfCar = getAllEventOfCar(travelEvents);
//在循环一个车道某个流向过程中需要计算的数据
//如果，在一个周期内，没有进入路口（通过停止线），则这条数据，放到下一个周期去计算,这辆车就不算了
// 直接将没有计算的数据，放到状态中去，注意是新定义一个状态，还是覆盖原来的状态，（专门定义一个状态存储，剩余未计算的数据
// ，后面触发计算时，进行合并到新来的数据暂存状态中去）
//TODO 处理无法完成计算的数据 事件没有到齐的情况
if(!allEventOfCar.containsKey("INCROSS")||!allEventOfCar.containsKey("ARRIVED")){
System.out.println("当前数据不完整，无法计算。。。。。"+carId);
//如果，当前车道 存在，不能完成计算的数据 （数据不全，没有进入路口的数据，则等待下次触发计算）
//获取当前车道
String eventLineId = allEventOfCar.values().iterator().next().getEventLineId();
if(leftData.containsKey(eventLineId)) {
// System.out.println("当前车道已存在，不完整的车辆数据。。。yes");
Map<String, List<TravelEvent>> arrived = leftData.get(eventLineId);
arrived.put(carId,travelEvents);
leftData.put(eventLineId,arrived);
} else {
// System.out.println("当前车道不存在，不完整的车辆数据。。。。no");
Map<String, List<TravelEvent>> arrived = new HashMap<>();
arrived.put(carId,travelEvents);
leftData.put(eventLineId,arrived);
}
leftDataEventState.put(crossCode,leftData);
continue ;
}
// 排除上一个周期已经算过的数据
if(allEventOfCar.get("INCROSS").getEventTime()<beginDateTime ||
allEventOfCar.get("INCROSS").getEventTime() > endDateTime){
continue;
}
System.out.println("当前车辆所有事件数据有：==========》\r\n"+allEventOfCar);
//1、通过时间
Long passTime = allEventOfCar.get("INCROSS").getEventTime()-allEventOfCar.get("ARRIVED").getEventTime();
//2、每辆车通过平均车速
Double carSpeed = getAverageSpeed(allEventOfCar);
//2、获取方向
String direction = dim_cnt_cross_lane_position.get(crossCode)
.get(allEventOfCar.get("INCROSS").getEventLineId()+"direction");
// System.out.println("方向是==========="+direction);
//2.1 获取流向
String flowDirection = getFlowDirection(allEventOfCar);
//3、控制延误
//4、停车次数 这里使用的是停车次数，不管停车几次只算一次（并非实际停车次数）
Integer stopTimes = stopCounts;
// System.out.println(carId+"当前停车次数-------------》"+stopTimes);
//5、停车延误之
Long stopDelayTime = getStopDelayTime(travelEvents);
//6、车头时距之和
//7、最后一辆车进入路口的时间（即inCrossTime最大值，记录此值是为了，后续有迟到数据，计算车头时距的时候用的）
Long inCrossTime = allEventOfCar.get("INCROSS").getEventTime();
carIdArriveTime.put(allEventOfCar.get("ARRIVED").getRecordId(),allEventOfCar.get("ARRIVED").getEventTime());
arriveTimeCarId.put(allEventOfCar.get("ARRIVED").getEventTime(),allEventOfCar.get("ARRIVED").getRecordId());
TravelCarInfo travelCarInfo = new TravelCarInfo();
travelCarInfo.setCarId(carId);
travelCarInfo.setAverageSpeed(carSpeed);
travelCarInfo.setCycleOrder(cycleOrder);
travelCarInfo.setDirection(direction);
travelCarInfo.setFlowDirection(flowDirection);
travelCarInfo.setCrossId(crossCode);
travelCarInfo.setInCrossTime(inCrossTime);
travelCarInfo.setStopDelayTime(stopDelayTime);
travelCarInfo.setStopTimes(stopTimes);
travelCarInfo.setPassTime(passTime);
list.add(travelCarInfo);
}
//对list中一辆车的旅行整合后的数据，进行分流，按照流向分成不同的集合，并按照进入世界计算相邻2辆车的车头时距
Set<String> flowDirectionSet = new HashSet<>();
//每个流向的车辆数据，存在一个排序的Set中，这样自动按照进入时间依次排好序了
Map<String,Set<TravelCarInfo>> flowMap = new HashMap<>();
for (TravelCarInfo t:list) {
if(flowDirectionSet.add(t.getFlowDirection())){
Set<TravelCarInfo> set = new TreeSet<>();
set.add(t);
flowMap.put(t.getFlowDirection(),set);
} else {
Set<TravelCarInfo> set = flowMap.get(t.getFlowDirection());
set.add(t);
flowMap.put(t.getFlowDirection(),set);
}
}
// 循环结束，一个车道上的所有车辆的整合后的数据，已经按照流向分开，装载在flowMap中了
// 下面处理每一个流向上的数据，每一个流向上的数据，会聚合生成一条数据，然后，sink
List<TravelLineSinkInfo> sinkList = new ArrayList<>();
for (String flowDirection:flowDirectionSet) {
System.out.println("当前数据流向是------》"+flowDirection);
//一个流向上每个车辆整合后的数据
Set<TravelCarInfo> travelCarInfosOfOneFlow = flowMap.get(flowDirection);
// sum
//在循环一个车道某个流向过程中需要计算的数据
//1、通过时间之和
//2、每辆车通过平均车速之和
//3、控制延误之和
//4、停车次数之和
//5、停车延误之和
//6、车头时距之和
//7、最后一辆车进入路口的时间（即inCrossTime最大值，记录此值是为了，后续有迟到数据，计算车头时距的时候用的）
List<TravelCarInfo> listOfOneFlow = new ArrayList<>();
for (TravelCarInfo t:travelCarInfosOfOneFlow) {
listOfOneFlow.add(t);
}
Long sumPassTime = 0L ;
double sumSpeed = 0.0 ;
// Long sumControlDelay = 0L ;
Integer sumStopTimes = 0 ;
Long sumStopDelay = 0L ;
Long sumCarHeadTimeGap = 0L ;
Long maxInCrossTime = 0L ;
String direction = "";
for (int i=0;i<listOfOneFlow.size();i++) {
TravelCarInfo t = listOfOneFlow.get(i);
sumPassTime = sumPassTime+t.getPassTime();
sumSpeed=sumSpeed+t.getAverageSpeed();
// sumControlDelay=sumControlDelay+t.getControlDelayTime();
sumStopTimes=sumStopTimes+t.getStopTimes();
sumStopDelay=sumStopDelay+t.getStopDelayTime();
if(i-1>=0){
//相邻2车的进入时间差之和，即为车头时距
sumCarHeadTimeGap=sumCarHeadTimeGap+(t.getInCrossTime()-listOfOneFlow.get(i-1).getInCrossTime());
}
if(t.getInCrossTime()>maxInCrossTime){
maxInCrossTime=t.getInCrossTime() ;
}
direction=t.getDirection();
}
int size = travelCarInfosOfOneFlow.size();
TravelLineSinkInfo travelLineSinkInfo = new TravelLineSinkInfo();
travelLineSinkInfo.setCross_id(crossCode);
travelLineSinkInfo.setCycle_id(cycleOrder);
travelLineSinkInfo.setDirection(direction);
travelLineSinkInfo.setFlow_direction(flowDirection);
travelLineSinkInfo.setLane_id(lineId);
travelLineSinkInfo.setPass_numbers(size);
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);
sinkList.add(travelLineSinkInfo);
}
Map<String, Map<String, List<TravelEvent>>> leftDataMap = leftDataEventState.get(crossCode);
System.out.println("这个周期剩余未计算的数据有："+leftDataMap);
return sinkList ;
}
/**
* 把一辆车的所有事件数据，按照事件名称进行划分
* @param travelEvents
* @return
*/
private Map<String,TravelEvent> getAllEventOfCar(List<TravelEvent> travelEvents){
Map<String,TravelEvent> map = new HashMap<>();
for (TravelEvent t:travelEvents) {
map.put(t.getEventType(),t);
}
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;
/**
* 查询车道维表数据-数据存在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 , turn_direction ) as flow_direction from qxlk.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);
}
}
connection.close();
return dimData;
}
/**
* 计算平均速度，传入一个list集合，计算每个元素中速度的平均值
* 只看旅行期间的平均速度，（照片时间在旅行开始和结束之间）
* 计算平均速度，不包含驶离和丢失是的速度，启动和停止，只会取其中一个进行计算，其他事件的速度都会算进去，包括outCross事件
* @param
* @return
*/
private Double getAverageSpeed(Map<String, TravelEvent> allEventOfCar) {
//根据当前车辆的所有事件数据，计算平均速度
Iterator<TravelEvent> iterator = allEventOfCar.values().iterator();
double sumSpeed =0.0 ;
while(iterator.hasNext()){
sumSpeed= sumSpeed+iterator.next().getEventSpeed();
}
return sumSpeed/allEventOfCar.keySet().size() ;
}
/**
* 计算停车延误时间 根据传入的是车辆捕获事件的集合，
* @return
*/
private Long getStopDelayTime(List<TravelEvent> travelEvents) {
//按照事件发送的事件顺序进行排序后，再计算停车延误时间
Set<TravelEvent> set = new TreeSet<>();
for (TravelEvent t:travelEvents) {
set.add(t);
}
Long startStopTime =0L; //开始停车时间
Long stopDelayTime =0L;
for (TravelEvent t:set) {
if(t.getEventType().equals("STOP")){
startStopTime=t.getEventTime();
}
if(t.getEventType().equals("STARTUP")){
stopDelayTime=stopDelayTime+(t.getEventTime()-startStopTime);
}
}
return stopDelayTime;
}
/**
* 根据进入路口车道，和驶出路口车道结合起来，判断具体的流向
* 根据进入路口的方法向和驶出路口的方向，判断流向
* @return
*/
private String getFlowDirection(Map<String, TravelEvent> allEventOfCar) throws Exception {
// 进来的数据都有INCROSS
TravelEvent inCross = allEventOfCar.get("INCROSS");
// 驶出路口的数据，单独存放在一个状态存储中
TravelEvent outCross = outCrossEventState.get(inCross.getRecordId());
// 当前车辆存在驶出路口的事件
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");
String outDirection = dim_cnt_cross_lane_position.get(outCross.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 "北右转";}
}
}
} else {
TravelEvent travelEvent = allEventOfCar.get("INCROSS");
//根据入口车道判断流向
String inDirection = dim_cnt_cross_lane_position.get(allEventOfCar.get("INCROSS")
.getCrossId()).get(travelEvent.getEventLineId()+"flow_direction");
return inDirection;
}
return "未知";
}
/**
* 将每个车道流向预聚合的数据存入mysql中去
* @param list
* @throws SQLException
*/
private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
DruidPooledConnection connection = dataSource.getConnection();
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," +
" 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);
String record_date = sdf2.format(date);
for (TravelLineSinkInfo t :list) {
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/job/TravelSituationAnalysisJob.java 0 → 100644
package com.zhht.irn.job;
import com.zhht.irn.entity.dto.CycleSignalData;
import com.zhht.irn.entity.dto.TravelEvent;
import com.zhht.irn.functions.TravelEventAndCycleCoProcessFunction;
import com.zhht.irn.schema.CycleSignalKafkaSchema;
import com.zhht.irn.schema.TravelEventKafkaSchema;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
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.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 java.util.Properties;
/**
* 行车情况 指标分析任务
*
* @author ruimeng
* @create 2022-11-14 11:43
**/
public class TravelSituationAnalysisJob {
public static void main(String[] args) {
try {
//1、创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// checkpoint配置
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
// checkpoint时间间隔3分钟
checkpointConfig.setCheckpointInterval(1* 60 * 1000);
// 两次checkpoint中最短时间间隔1分钟
checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
// 同时同时允许1个checkpoint进行
checkpointConfig.setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//设置重启策略
env.setRestartStrategy( RestartStrategies.fixedDelayRestart(5, Time.seconds(10)));
/*
关于flink checkpoint存储的设置统一在集群配置文件中设置
checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setStateBackend(new RocksDBStateBackend("file:///opt/rocksDb/", true));*/
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(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(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//旅行事件
FlinkKafkaConsumerBase<TravelEvent> travelEvent =
new FlinkKafkaConsumer<TravelEvent>("trips_event_info",
new TravelEventKafkaSchema(),
kafkaProperties) {
};
DataStreamSource<TravelEvent> travelEventStream = env.addSource(travelEvent);
travelEventStream.print();
//周期信号
FlinkKafkaConsumer<CycleSignalData> cycleSignalData =
new FlinkKafkaConsumer<>("signal_cycle_data",
new CycleSignalKafkaSchema(),
kafkaProperties2);
DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
cycleSignalDataStream.print();
//2、使用connect，把2个流联合处理
ConnectedStreams<TravelEvent, CycleSignalData> connect = travelEventStream
.connect(cycleSignalDataStream);
ConnectedStreams<TravelEvent, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
(KeySelector<TravelEvent, String>) travelInfo1 -> travelInfo1.getCrossId(),
(KeySelector<CycleSignalData, String>) cycleSignalData1 -> cycleSignalData1.getCrossCode()
);
travelInfoCycleSignalDataConnectedStreams.process(new TravelEventAndCycleCoProcessFunction());
env.execute();
} catch (Exception e){
e.printStackTrace();
}
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/schema/CycleSignalKafkaSchema.java 0 → 100644
package com.zhht.irn.schema;
import com.alibaba.fastjson.JSONObject;
import com.zhht.irn.entity.dto.CycleSignalData;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
/**
* 周期信号数据kafka约束
*
* @author ruimeng
* @create 2022-11-14 13:47
**/
public class CycleSignalKafkaSchema implements KafkaDeserializationSchema<CycleSignalData> {
@Override
public boolean isEndOfStream(CycleSignalData signalData) {
return false;
}
@Override
public CycleSignalData deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
String recordDatas = new String(consumerRecord.value());
try {
if (recordDatas.length() >= 1) {
CycleSignalData cycleSignalData = JSONObject.parseObject(recordDatas, CycleSignalData.class);
return cycleSignalData;
}
} catch (Exception e) {
System.out.println("出现异常 ，消息体中的字符串是:");
System.out.println(recordDatas);
e.printStackTrace();
}
return null;
}
@Override
public TypeInformation<CycleSignalData> getProducedType() {
return TypeInformation.of(CycleSignalData.class);
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/schema/TravelEventKafkaSchema.java 0 → 100644
package com.zhht.irn.schema;
import com.alibaba.fastjson.JSONObject;
import com.zhht.irn.entity.dto.TravelEvent;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
/**
* 旅行事件kafka约束
*
* @author ruimeng
* @create 2022-11-18 11:16
**/
public class TravelEventKafkaSchema implements KafkaDeserializationSchema<TravelEvent> {
@Override
public boolean isEndOfStream(TravelEvent signalData) {
return false;
}
@Override
public TravelEvent deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) {
String recordDatas = new String(consumerRecord.value());
try {
if (recordDatas.length() >= 1) {
TravelEvent travelEvent = JSONObject.parseObject(recordDatas, TravelEvent.class);
return travelEvent;
}
} catch (Exception e) {
System.out.println("出现异常 ，消息体中的字符串是:");
System.out.println(recordDatas);
e.printStackTrace();
}
return null;
}
@Override
public TypeInformation<TravelEvent> getProducedType() {
return TypeInformation.of(TravelEvent.class);
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/schema/TravelInfoKafkaSchema.java 0 → 100644
package com.zhht.irn.schema;
import com.alibaba.fastjson.JSONObject;
import com.zhht.irn.entity.dto.TravelInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
/**
* 事件数据kafka约束
*
* @author ruimeng
* @create 2022-11-14 13:35
**/
public class TravelInfoKafkaSchema implements KafkaDeserializationSchema<TravelInfo> {
@Override
public boolean isEndOfStream(TravelInfo travelInfo) {
return false;
}
@Override
public TravelInfo deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) {
String recordDatas = new String(consumerRecord.value());
try {
if (recordDatas.length() >= 1) {
TravelInfo travelInfo = JSONObject.parseObject(recordDatas, TravelInfo.class);
return travelInfo;
}
} catch (Exception e) {
System.out.println("出现异常 ，消息体中的字符串是:");
System.out.println(recordDatas);
e.printStackTrace();
}
return null;
}
@Override
public TypeInformation<TravelInfo> getProducedType() {
return TypeInformation.of(TravelInfo.class);
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DruidConnectPoolUtils.java 0 → 100644
package com.zhht.irn.utils;
import com.alibaba.druid.pool.DruidDataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
/**
* @author Rui meng
* @description: TODO
* @date 2022-09-05 15:11:00
*/
public class DruidConnectPoolUtils {
/**
* 初始化连接池
* @param dbHosts
* @param username
* @param password
* @return
*/
public static DruidDataSource getDataSource(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.setMaxActive(30); //最大连接数，默认8
dataSource.setMinIdle(5); //最小闲置数
dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
dataSource.setTestOnBorrow(false); // 借出连接时不要测试，否则很影响性能
return dataSource;
}
/**
* CK close
* @param connection
* @param ps
*/
public static void closeConn(DruidDataSource dataSource, Connection connection, PreparedStatement... ps) {
try {
if (ps.length > 0) {
for (PreparedStatement p : ps) {
if (p != null) {
p.close();
}
}
}
if (connection != null) {
connection.close();
}
if (dataSource != null) {
dataSource.close();
}
} catch (SQLException 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