Skip to content

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

Merge remote-tracking branch 'origin/master'

parents f7289562 ff116bc4
Showing with 565 additions and 0 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/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