Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 15a5fcc3 by 吴延飞
Options

提交绿性比、损失时间指标的计算代码，完善接入kafka数据流的工具类

parent f2717a21
Showing with 159 additions and 177 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Stage.java
......@@ -189,4 +189,28 @@ public class Stage {
public void setEndPlanId(Integer endPlanId) {
this.endPlanId = endPlanId;
}
@Override
public String toString() {
return "Stage{" +
"cycleOrder=" + cycleOrder +
", crossCode='" + crossCode + '\'' +
", phaseValue='" + phaseValue + '\'' +
", beginDateTime='" + beginDateTime + '\'' +
", endDateTime='" + endDateTime + '\'' +
", beginControlModelCode='" + beginControlModelCode + '\'' +
", endControlModelCode='" + endControlModelCode + '\'' +
", beginPlanId=" + beginPlanId +
", endPlanId=" + endPlanId +
", duration=" + duration +
", green=" + green +
", yellow=" + yellow +
", allRed=" + allRed +
", phaseOrder=" + phaseOrder +
", greenRatio=" + greenRatio +
", validGreen=" + validGreen +
", lossTime=" + lossTime +
", stageForDirectionList=" + stageForDirectionList +
'}';
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Travel.java
......@@ -398,4 +398,46 @@ public class Travel {
public void setMaxSpeed(double maxSpeed) {
this.maxSpeed = maxSpeed;
}
@Override
public String toString() {
return "Travel{" +
"id=" + id +
", crossId='" + crossId + '\'' +
", recordId='" + recordId + '\'' +
", recordTime='" + recordTime + '\'' +
", carId=" + carId +
", carColor='" + carColor + '\'' +
", plate='" + plate + '\'' +
", plateColor='" + plateColor + '\'' +
", carType='" + carType + '\'' +
", firstLineId='" + firstLineId + '\'' +
", firstSpeed=" + firstSpeed +
", firstTime='" + firstTime + '\'' +
", arrivedLineId='" + arrivedLineId + '\'' +
", arrivedSpeed=" + arrivedSpeed +
", arrivedTime='" + arrivedTime + '\'' +
", inCrossLineId='" + inCrossLineId + '\'' +
", inSpeed=" + inSpeed +
", inCrossTime='" + inCrossTime + '\'' +
", outCrossLineId='" + outCrossLineId + '\'' +
", outSpeed=" + outSpeed +
", outCrossTime='" + outCrossTime + '\'' +
", awayLineId='" + awayLineId + '\'' +
", awaySpeed=" + awaySpeed +
", awayTime='" + awayTime + '\'' +
", lastLineId='" + lastLineId + '\'' +
", lastSpeed=" + lastSpeed +
", lastTime='" + lastTime + '\'' +
", crossingTime=" + crossingTime +
", locations=" + locations +
", remark='" + remark + '\'' +
", stopCount=" + stopCount +
", stopTime=" + stopTime +
", avgSpeed=" + avgSpeed +
", minSpeed=" + minSpeed +
", maxSpeed=" + maxSpeed +
'}';
}
}
\ No newline at end of file
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/GreenRadioJob.java
......@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.GreenRadioSink;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
......@@ -11,80 +12,57 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.List;
/**
* 两阶段提交参考文档：
* https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
* 实时接入信控周期数据
* 计算绿信比、损失时间
*
* @author 余根龙
* @date 2022-11-15 16:25:00
*/
public class GreenRadioJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
// 目前假设从Kafka接入的数据为json格式的字符串
// Cycle Topic 的消息格式如下：
// {"id":"1","crossCode":"0001","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"300","cycleOrder":"1",
// "stageList":[
// {"phaseValue":"101","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:01:00","duration":"60","green":"30","yellow":"3","allRed":"0","phaseOrder":"1"},
// {"phaseValue":"102","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"240","green":"120","yellow":"3","allRed":"0","phaseOrder":"2"}
// ]
//}
// 造一条数据
DataStream<String> stream = env.fromElements("{\"id\":\"3\",\"crossCode\":\"0001\",\"signalCode\":\"1000001\",\"beginDateTime\":\"2020-08-02 08:10:03\",\"endDateTime\":\"2020-08-02 08:12:11\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"stageList\":[{\"phaseValue\":\"101\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endDateTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\",\"phaseOrder\":\"1\"},{\"phaseValue\":\"102\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endDateTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\",\"phaseOrder\":\"2\"}]}");
DataStream<Cycle> lxbStream = stream.map(new MapFunction<String, Cycle>() {
@Override
public Cycle map(String s) throws Exception {
//启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
//清场损失时间 = 黄灯时长 + 全红时长 - 2
//阶段损失时间 = 启动损失时间 + 清场损失时间
//阶段有效绿灯时间 = 绿灯时长+黄灯时长+全红时长 - 阶段损失时间
//阶段绿信比 = 阶段有效绿灯时间/周期时长
//周期绿信比 = 阶段绿信比之和
Cycle cycle = JSON.parseObject(s, Cycle.class);
long cycleId= cycle.getId();
int duration = cycle.getDuration();
String crossCode = cycle.getCrossCode();
double cycleGreenRatio = 0;
int cycleLossTime = 0;
FlinkUtils.createKafkaStream(args, env, "signal_cycle_data")
.map(record -> {
// 字符串解析为实体类
Cycle cycle = JSON.parseObject(record, Cycle.class);
System.out.println(cycle);
double cycleGreenRatioTotal = 0;
int cycleLossTimeTotal = 0;
// 先按阶段进行计算
List<Stage> stageList = cycle.getStageList();
for (Stage stage : stageList) {
//相位号
String phaseValue = stage.getPhaseValue();
// 绿灯时间
int green = stage.getGreen();
// 黄灯时间
int yellow = stage.getYellow();
// 全红时间
int allRed = stage.getAllRed();
// 启动损失时间，一般定义：2s
// 启动损失时间，一般定义：2s 从配置文件读取，正常情况下都是2，不排除个别城市有差异
int startLossTime = 2;
//清场损失时间
// 清场损失时间 = 黄灯时长 + 全红时长 - 2
int clearLossTime = yellow + allRed - 2;
//阶段损失时间
// 阶段损失时间启动损失时间 + 清场损失时间
int phaseLossTime = startLossTime + clearLossTime;
//阶段有效绿灯时间
// 阶段有效绿灯时间 = 绿灯时长+黄灯时长+全红时长 - 阶段损失时间
int phaseValidGreen = green + yellow + allRed - phaseLossTime;
System.out.println("******duration:"+duration);
System.out.println("******phaseValidGreen:"+phaseValidGreen);
//阶段绿信比
double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
System.out.println("******phaseGreenRatio:"+phaseGreenRatio);
//组装阶段绿信比、有效绿灯时间、损失时间
stage.setGreenRatio(phaseGreenRatio);
stage.setValidGreen(phaseValidGreen);
stage.setLossTime(phaseLossTime);
//周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
//周期损失时间= 各个阶段损失时间之和
cycleLossTime = cycleLossTime + phaseLossTime;
System.out.println("******cycleGreenRatio:"+cycleGreenRatio);
// 阶段绿信比 = 阶段有效绿灯时间/周期时长
double phaseGreenRatio = Double.valueOf(phaseValidGreen) / cycle.getDuration();
// 周期绿信比 = 阶段绿信比之和
cycleGreenRatioTotal = cycleGreenRatioTotal + phaseGreenRatio;
// 周期损失时间 = 各个阶段损失时间之和
cycleLossTimeTotal = cycleLossTimeTotal + phaseLossTime;
}
// 组装周期绿信比、损失时间
cycle.setGreenRatio(cycleGreenRatio);
cycle.setLossTime(cycleLossTime);
cycle.setGreenRatio(cycleGreenRatioTotal);
cycle.setLossTime(cycleLossTimeTotal);
return cycle;
}
});
lxbStream.addSink(new GreenRadioSink());
})
.addSink(new GreenRadioSink());
env.execute();
}
}
\ No newline at end of file
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelInfoJob.java
......@@ -15,7 +15,7 @@ import org.slf4j.LoggerFactory;
import java.util.List;
/**
* 旅行信心实时任务
* 旅行信息实时任务
*
* 计算车辆旅行过程中的指标
* 包括 最高、最低、平均速度，停车总时长，实际停车次数
......@@ -28,14 +28,13 @@ public class TravelInfoJob {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 通过kafka配置文件，获取String类型的实时数据流
// DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
DataStream<String> rawStringStream = env.addSource(new TravelMockSource());
// 开始计算旅行信息相关指标
rawStringStream.map((MapFunction<String, Travel>) s -> {
System.out.println(s);
// json 格式字符串转化为 Travel 实体类 @todo 不一定能正确解析
Travel travel = JSON.parseObject(s,Travel.class);
Travel travel = JSON.parseObject(s, Travel.class);
List<Location> locations = travel.getLocations();
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java
......@@ -14,38 +14,34 @@ import java.util.Date;
import java.util.List;
/**
* domain traffic
* 周期信控数据计算
*
* 结果表：app_cycle_green_ratio
* 计算结果指标包含 绿信比、损失时间
*/
public class GreenRadioSink extends RichSinkFunction<Cycle> {
Connection connection;
PreparedStatement insertCyclePstmt;
PreparedStatement insertPhasePstmt;
Date date = new Date();
SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd :hh:mm:ss");
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
String path = "F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path);
String cycleSql="insert into app_cycle_green_ratio(id,cross_code,signal_code,begin_time,end_time,duration,cycle_order,green_ratio,loss_time,update_time) values (?,?,?,?,?,?,?,?,?,?)";
String cycleSql="insert into app_cycle_green_ratio" +
"(id,cross_code,signal_code,begin_time,end_time,duration,cycle_order,green_ratio,loss_time,update_time) " +
"values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
insertCyclePstmt = connection.prepareStatement(cycleSql);
String phaseSql="insert into app_phase_green_ratio(cycle_order,cross_code,phase_value,begin_time,end_time,duration,phase_order,green_ratio,valid_green,loss_time,update_time) values (?,?,?,?,?,?,?,?,?,?,?)";
insertPhasePstmt = connection.prepareStatement(phaseSql);
}
@Override
public void close() throws Exception {
super.close();
if(insertCyclePstmt != null) insertCyclePstmt.close();
if(insertPhasePstmt != null) insertPhasePstmt.close();
if(connection != null) connection.close();
}
/**
* 来一条数据就执行一次
*
* 1000w的数据 1000w次
*/
@Override
public void invoke(Cycle cycle, Context context) throws Exception {
System.out.println("=====invoke======" + cycle.getId() + "-->" + cycle.getCrossCode());
......@@ -54,29 +50,11 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
insertCyclePstmt.setString(3, cycle.getSignalCode());
insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertCyclePstmt.setInt(6,cycle.getDuration());
insertCyclePstmt.setLong(7,cycle.getCycleOrder());
insertCyclePstmt.setDouble(8,cycle.getGreenRatio());
insertCyclePstmt.setInt(9,cycle.getLossTime());
insertCyclePstmt.setInt(6, cycle.getDuration());
insertCyclePstmt.setLong(7, cycle.getCycleOrder());
insertCyclePstmt.setDouble(8, cycle.getGreenRatio());
insertCyclePstmt.setInt(9, cycle.getLossTime());
insertCyclePstmt.setString(10, DateUtils.getTodayTime());
insertCyclePstmt.execute();
List<Stage> stageList = cycle.getStageList();
for(int i=0;i<stageList.size();i++){
Stage phase = stageList.get(i);
insertPhasePstmt.setLong(1,cycle.getCycleOrder());
insertPhasePstmt.setString(2,phase.getCrossCode());
insertPhasePstmt.setString(3,phase.getPhaseValue());
insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertPhasePstmt.setInt(6,cycle.getDuration());
insertPhasePstmt.setInt(7,phase.getPhaseOrder());
insertPhasePstmt.setDouble(8,phase.getGreenRatio());
insertPhasePstmt.setInt(9,phase.getValidGreen());
insertPhasePstmt.setInt(10,phase.getLossTime());
insertPhasePstmt.setString(11, DateUtils.getTodayTime());
insertPhasePstmt.addBatch();
}
insertPhasePstmt.executeBatch();
insertPhasePstmt.clearBatch();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelMetricSink.java
......@@ -11,7 +11,10 @@ import java.sql.PreparedStatement;
import java.util.Date;
/**
* domain traffic
* 旅行信息指标计算
*
* 结果表：app_travel_metric
* 计算结果指标包含 平均、最大、最小速度，停车次数、停车时长
*/
public class TravelMetricSink extends RichSinkFunction<Travel> {
Connection connection;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
......@@ -26,98 +26,39 @@ public class FlinkUtils {
public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
public static <T> DataStream<T> createKafkaStreamV3(String[] args, Class<? extends KafkaDeserializationSchema<T>> deser) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
String autoCommit = tool.get("enable.auto.commit", "false");
String offsetReset = tool.get("auto.offset.reset", "earliest");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", autoCommit);
properties.setProperty("auto.offset.reset",offsetReset);
int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
String checkpointPath = tool.get("checkpoint.path", "file:///Users/rocky/Desktop/Flink/workspace/imooc-flink/state");
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topics, deser.newInstance(), properties);
return env.addSource(kafkaConsumer);
}
public static <T> DataStream<T> createKafkaStreamV2(String[] args, Class<? extends DeserializationSchema<T>> deser) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
String autoCommit = tool.get("enable.auto.commit", "false");
String offsetReset = tool.get("auto.offset.reset", "earliest");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", autoCommit);
properties.setProperty("auto.offset.reset",offsetReset);
int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
String checkpointPath = tool.get("checkpoint.path", "file:///f/workspace/ZHHT-IRN-DB-ANALYSIS/state");
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topics, deser.newInstance(), properties);
return env.addSource(kafkaConsumer);
/**
* 根据JAVA命令行参数获取kafka配置地址，来获取相应的topic中的字符串数据流
* topic和 offset 均取配置文件中的值
*
* @param args
* @param environment
* @return
*/
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
String topics = kafkaTool.getRequired("kafka.input.topics");
return createKafkaStream(args, environment, topics);
}
public static DataStream<String> createKafkaStreamV1(String[] args) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
String autoCommit = tool.get("enable.auto.commit", "false");
String offsetReset = tool.get("auto.offset.reset", "earliest");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", autoCommit);
properties.setProperty("auto.offset.reset",offsetReset);
int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
String checkpointPath = tool.get("checkpoint.path", "file:///Users/rocky/Desktop/Flink/workspace/imooc-flink/state");
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
return env.addSource(kafkaConsumer);
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
return createKafkaStream(args, environment, topic, offsetReset);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment) {
// 获取Kafka配置文件
ParameterTool finkArgs = ParameterTool.fromArgs(args);
String kafkaConfigPath = finkArgs.getRequired("kafka.conf");
ParameterTool kafkaTool;
try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
} catch (IOException e) {
logger.error("kafka 配置文件未设置，请检查!");
throw new RuntimeException(e);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String offset) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
// 加载Kafka信息 { topic 相关 }
String groupId = kafkaTool.get("group.id", "test1");
String groupId = kafkaTool.get("group.id", "group-wuyanfei");
String servers = kafkaTool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(kafkaTool.getRequired("kafka.input.topics").split(","));
List<String> topics = Arrays.asList(topic.split(","));
String autoCommit = kafkaTool.get("enable.auto.commit", "false");
String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", autoCommit);
properties.setProperty("auto.offset.reset",offsetReset);
properties.setProperty("auto.offset.reset",offset);
int checkpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
......@@ -133,4 +74,18 @@ public class FlinkUtils {
DataStream<String> rawStringStream = environment.addSource(kafkaConsumer);
return rawStringStream;
}
private static ParameterTool initParameterToolByArgs(String[] args) {
// 获取Kafka配置文件
ParameterTool finkArgs = ParameterTool.fromArgs(args);
String kafkaConfigPath = finkArgs.getRequired("kafka.conf");
ParameterTool kafkaTool;
try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
} catch (IOException e) {
logger.error("kafka 配置文件未设置，请检查!");
throw new RuntimeException(e);
}
return kafkaTool;
}
}
realtime/hologram-streaming/src/main/resources/kafka.properties
bootstrap.servers=124.71.213.187:9092
kafka.input.topics=wyf-test-topic
\ No newline at end of file
bootstrap.servers=10.100.32.173:9092
kafka.input.topics=trips_info
auto.offset.reset=earliest
\ No newline at end of file
realtime/hologram-streaming/src/main/resources/mysql.properties
driver=com.mysql.jdbc.Driver
host=127.0.0.1
port=13306
port=13305
username=root
password=mima
database=test
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