Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 5e3ba4f6 by 吴延飞
Options

Merge remote-tracking branch 'origin/master'

parents f8275951 0feac360
Showing with 119 additions and 63 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java
package com.zhht.irn.entity; package com.zhht.irn.entity;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
...@@ -15,11 +14,14 @@ public class Cycle { ...@@ -15,11 +14,14 @@ public class Cycle {
// 路口编号 // 路口编号
private String crossCode; private String crossCode;
//信控机编号
private String signalCode;
// 周期开始时间 // 周期开始时间
private String beginTime; private String beginDateTime;
// 周期结束时间 // 周期结束时间
private String endTime; private String endDateTime;
private int duration; private int duration;
...@@ -27,9 +29,11 @@ public class Cycle { ...@@ -27,9 +29,11 @@ public class Cycle {
private int cycleOrder; private int cycleOrder;
//绿信比 //绿信比
private Double greenRatio; private Double greenRatio;
//损失时间
private int lossTime;
// 周期内相位列表（依次执行） // 周期内相位列表（依次执行）
private List<Phase> phaseList; private List<Stage> stageList;
public Long getId() { public Long getId() {
return id; return id;
...@@ -47,20 +51,20 @@ public class Cycle { ...@@ -47,20 +51,20 @@ public class Cycle {
this.crossCode = crossCode; this.crossCode = crossCode;
} }
public String getBeginTime() { public String getBeginDateTime() {
return beginTime; return beginDateTime;
} }
public void setBeginTime(String beginTime) { public void setBeginDateTime(String beginDateTime) {
this.beginTime = beginTime; this.beginDateTime = beginDateTime;
} }
public String getEndTime() { public String getEndDateTime() {
return endTime; return endDateTime;
} }
public void setEndTime(String endTime) { public void setEndDateTime(String endDateTime) {
this.endTime = endTime; this.endDateTime = endDateTime;
} }
public int getDuration() { public int getDuration() {
...@@ -79,12 +83,12 @@ public class Cycle { ...@@ -79,12 +83,12 @@ public class Cycle {
this.cycleOrder = cycleOrder; this.cycleOrder = cycleOrder;
} }
public List<Phase> getPhaseList() { public List<Stage> getStageList() {
return phaseList; return stageList;
} }
public void setPhaseList(List<Phase> phaseList) { public void setStageList(List<Stage> stageList) {
this.phaseList = phaseList; this.stageList = stageList;
} }
public Double getGreenRatio() { public Double getGreenRatio() {
...@@ -94,4 +98,20 @@ public class Cycle { ...@@ -94,4 +98,20 @@ public class Cycle {
public void setGreenRatio(Double greenRatio) { public void setGreenRatio(Double greenRatio) {
this.greenRatio = greenRatio; this.greenRatio = greenRatio;
} }
public int getLossTime() {
return lossTime;
}
public void setLossTime(int lossTime) {
this.lossTime = lossTime;
}
public String getSignalCode() {
return signalCode;
}
public void setSignalCode(String signalCode) {
this.signalCode = signalCode;
}
} }
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Phase.java → realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Stage.java
...@@ -5,10 +5,10 @@ import java.util.Date; ...@@ -5,10 +5,10 @@ import java.util.Date;
/** /**
* 相位实体类 * 相位实体类
*/ */
public class Phase { public class Stage {
//周期ID //周期ID
private Long cycleId; private Long cycleOrder;
// 路口编号 // 路口编号
private String crossCode; private String crossCode;
// 相位编号 // 相位编号
...@@ -31,12 +31,18 @@ public class Phase { ...@@ -31,12 +31,18 @@ public class Phase {
// 相位全红时间 // 相位全红时间
private int allRed; private int allRed;
//相位序号
private int phaseOrder;
// 阶段绿信比 // 阶段绿信比
private Double greenRatio; private Double greenRatio;
// 阶段有效绿灯时长 // 阶段有效绿灯时长
private int validGreen; private int validGreen;
//损失时间
private int lossTime;
public String getPhaseValue() { public String getPhaseValue() {
return phaseValue; return phaseValue;
} }
...@@ -93,12 +99,12 @@ public class Phase { ...@@ -93,12 +99,12 @@ public class Phase {
this.allRed = allRed; this.allRed = allRed;
} }
public Long getCycleId() { public Long getCycleOrder() {
return cycleId; return cycleOrder;
} }
public void setCycleId(Long cycleId) { public void setCycleOrder(Long cycleOrder) {
this.cycleId = cycleId; this.cycleOrder = cycleOrder;
} }
public String getCrossCode() { public String getCrossCode() {
...@@ -124,4 +130,20 @@ public class Phase { ...@@ -124,4 +130,20 @@ public class Phase {
public void setValidGreen(int validGreen) { public void setValidGreen(int validGreen) {
this.validGreen = validGreen; this.validGreen = validGreen;
} }
public int getPhaseOrder() {
return phaseOrder;
}
public void setPhaseOrder(int phaseOrder) {
this.phaseOrder = phaseOrder;
}
public int getLossTime() {
return lossTime;
}
public void setLossTime(int lossTime) {
this.lossTime = lossTime;
}
} }
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/FlinkKafka2MysqlApp.java
package com.zhht.irn.flow; package com.zhht.irn.flow;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle; import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase; import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.LXBSink; import com.zhht.irn.sink.LXBSink;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import javax.xml.crypto.Data;
import java.util.List; import java.util.List;
/** /**
...@@ -42,8 +39,8 @@ public class FlinkKafka2MysqlApp { ...@@ -42,8 +39,8 @@ public class FlinkKafka2MysqlApp {
//绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间 //绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
Cycle cycle = JSON.parseObject(s, Cycle.class); Cycle cycle = JSON.parseObject(s, Cycle.class);
Long allValidGreen = 0L; Long allValidGreen = 0L;
List<Phase> phaseList = cycle.getPhaseList(); List<Stage> phaseList = cycle.getStageList();
for (Phase phase : phaseList) { for (Stage phase : phaseList) {
// 绿灯时间 // 绿灯时间
long green = phase.getGreen(); long green = phase.getGreen();
// 黄灯时间 // 黄灯时间
...@@ -60,7 +57,7 @@ public class FlinkKafka2MysqlApp { ...@@ -60,7 +57,7 @@ public class FlinkKafka2MysqlApp {
} }
// 绿信比 // 绿信比
Double lxb = allValidGreen.doubleValue() / cycle.getDuration().doubleValue(); Double lxb = allValidGreen.doubleValue() / cycle.getDuration();
return Tuple2.of(cycle.getCrossCode(), lxb); return Tuple2.of(cycle.getCrossCode(), lxb);
} }
}); });
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/GreenRadioJob.java
...@@ -2,7 +2,7 @@ package com.zhht.irn.flow; ...@@ -2,7 +2,7 @@ package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle; import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase; import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.GreenRadioSink; import com.zhht.irn.sink.GreenRadioSink;
import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStream;
...@@ -20,14 +20,14 @@ public class GreenRadioJob { ...@@ -20,14 +20,14 @@ public class GreenRadioJob {
// DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class); // DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
// 目前假设从Kafka接入的数据为json格式的字符串 // 目前假设从Kafka接入的数据为json格式的字符串
// Cycle Topic 的消息格式如下： // Cycle Topic 的消息格式如下：
// {"Id":"","CrossCode":"","BeginTime":"","EndTime":"","Duration":"","CycleOrder":"", // {"id":"1","crossCode":"0001","beginDateTime":"2020-08-02 08:00:00","endDateTime":"2020-08-02 08:05:00","duration":"300","cycleOrder":"1",
// "detail":[{"PhaseValue":"101","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}, // "stageList":[
// {"PhaseValue":"102","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}, // {"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":"103","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}, // {"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"}
// {"PhaseValue":"104","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}] // ]
//}
// 造一条数据 // 造一条数据
DataStream<String> stream = env.fromElements("{\"id\":\"1\",\"crossCode\":\"0001\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"phaseList\":[{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"102\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}"); DataStream<String> stream = env.fromElements("{\"id\":\"1\",\"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>() { DataStream<Cycle> lxbStream = stream.map(new MapFunction<String, Cycle>() {
@Override @Override
public Cycle map(String s) throws Exception { public Cycle map(String s) throws Exception {
...@@ -42,16 +42,17 @@ public class GreenRadioJob { ...@@ -42,16 +42,17 @@ public class GreenRadioJob {
int duration = cycle.getDuration(); int duration = cycle.getDuration();
String crossCode = cycle.getCrossCode(); String crossCode = cycle.getCrossCode();
double cycleGreenRatio = 0; double cycleGreenRatio = 0;
List<Phase> phaseList = cycle.getPhaseList(); int cycleLossTime = 0;
for (Phase phase : phaseList) { List<Stage> stageList = cycle.getStageList();
for (Stage stage : stageList) {
//相位号 //相位号
String phaseValue = phase.getPhaseValue(); String phaseValue = stage.getPhaseValue();
// 绿灯时间 // 绿灯时间
int green = phase.getGreen(); int green = stage.getGreen();
// 黄灯时间 // 黄灯时间
int yellow = phase.getYellow(); int yellow = stage.getYellow();
// 全红时间 // 全红时间
int allRed = phase.getAllRed(); int allRed = stage.getAllRed();
// 启动损失时间，一般定义：2s // 启动损失时间，一般定义：2s
int startLossTime = 2; int startLossTime = 2;
//清场损失时间 //清场损失时间
...@@ -66,15 +67,19 @@ public class GreenRadioJob { ...@@ -66,15 +67,19 @@ public class GreenRadioJob {
//阶段绿信比 //阶段绿信比
double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration; double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
System.out.println("******phaseGreenRatio:"+phaseGreenRatio); System.out.println("******phaseGreenRatio:"+phaseGreenRatio);
//组装阶段绿信比 //组装阶段绿信比、有效绿灯时间、损失时间
phase.setGreenRatio(phaseGreenRatio); stage.setGreenRatio(phaseGreenRatio);
phase.setValidGreen(phaseValidGreen); stage.setValidGreen(phaseValidGreen);
stage.setLossTime(phaseLossTime);
//周期绿信比= 各个阶段绿信比之和 //周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio; cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
//周期损失时间= 各个阶段损失时间之和
cycleLossTime = cycleLossTime + phaseLossTime;
System.out.println("******cycleGreenRatio:"+cycleGreenRatio); System.out.println("******cycleGreenRatio:"+cycleGreenRatio);
} }
// 组装周期绿信比 // 组装周期绿信比、损失时间
cycle.setGreenRatio(cycleGreenRatio); cycle.setGreenRatio(cycleGreenRatio);
cycle.setLossTime(cycleLossTime);
return cycle; return cycle;
} }
}); });
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java
package com.zhht.irn.sink; package com.zhht.irn.sink;
import com.zhht.irn.entity.Cycle; import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase; import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils; import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection; import java.sql.Connection;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List; import java.util.List;
/** /**
...@@ -17,15 +20,17 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> { ...@@ -17,15 +20,17 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
Connection connection; Connection connection;
PreparedStatement insertCyclePstmt; PreparedStatement insertCyclePstmt;
PreparedStatement insertPhasePstmt; PreparedStatement insertPhasePstmt;
Date date = new Date();
SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd :hh:mm:ss");
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
// path暂定为固定值 // path暂定为固定值
String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties"; String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path); connection = MySQLUtils.getConnection(path);
String cycleSql="insert into app_cycle_green_ratio(cycle_id,cross_code,begin_time,end_time,duration,green_ratio) 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); insertCyclePstmt = connection.prepareStatement(cycleSql);
String phaseSql="insert into app_phase_green_ratio(cycle_id,cross_code,phase_value,begin_time,end_time,duration,green_ratio,valid_green) values (?,?,?,?,?,?,?,?)"; 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); insertPhasePstmt = connection.prepareStatement(phaseSql);
} }
@Override @Override
...@@ -46,22 +51,29 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> { ...@@ -46,22 +51,29 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
System.out.println("=====invoke======" + cycle.getId() + "-->" + cycle.getCrossCode()); System.out.println("=====invoke======" + cycle.getId() + "-->" + cycle.getCrossCode());
insertCyclePstmt.setLong(1, cycle.getId()); insertCyclePstmt.setLong(1, cycle.getId());
insertCyclePstmt.setString(2, cycle.getCrossCode()); insertCyclePstmt.setString(2, cycle.getCrossCode());
insertCyclePstmt.setString(3, cycle.getBeginTime()); insertCyclePstmt.setString(3, cycle.getSignalCode());
insertCyclePstmt.setString(4, cycle.getEndTime()); insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setInt(5,cycle.getDuration()); insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertCyclePstmt.setDouble(6,cycle.getGreenRatio()); insertCyclePstmt.setInt(6,cycle.getDuration());
insertCyclePstmt.setInt(7,cycle.getCycleOrder());
insertCyclePstmt.setDouble(8,cycle.getGreenRatio());
insertCyclePstmt.setInt(9,cycle.getLossTime());
insertCyclePstmt.setString(10, DateUtils.getTodayTime());
insertCyclePstmt.execute(); insertCyclePstmt.execute();
List<Phase> phaseList = cycle.getPhaseList(); List<Stage> stageList = cycle.getStageList();
for(int i=0;i<phaseList.size();i++){ for(int i=0;i<stageList.size();i++){
Phase phase = phaseList.get(i); Stage phase = stageList.get(i);
insertPhasePstmt.setLong(1,cycle.getId()); insertPhasePstmt.setLong(1,cycle.getCycleOrder());
insertPhasePstmt.setString(2,phase.getCrossCode()); insertPhasePstmt.setString(2,phase.getCrossCode());
insertPhasePstmt.setString(3,phase.getPhaseValue()); insertPhasePstmt.setString(3,phase.getPhaseValue());
insertPhasePstmt.setString(4, cycle.getBeginTime()); insertPhasePstmt.setString(4, cycle.getBeginDateTime());
insertPhasePstmt.setString(5, cycle.getEndTime()); insertPhasePstmt.setString(5, cycle.getEndDateTime());
insertPhasePstmt.setInt(6,cycle.getDuration()); insertPhasePstmt.setInt(6,cycle.getDuration());
insertPhasePstmt.setDouble(7,phase.getGreenRatio()); insertPhasePstmt.setInt(7,phase.getPhaseOrder());
insertPhasePstmt.setInt(8,phase.getValidGreen()); insertPhasePstmt.setDouble(8,phase.getGreenRatio());
insertPhasePstmt.setInt(9,phase.getValidGreen());
insertPhasePstmt.setInt(10,phase.getLossTime());
insertPhasePstmt.setString(11, DateUtils.getTodayTime());
insertPhasePstmt.addBatch(); insertPhasePstmt.addBatch();
} }
insertPhasePstmt.executeBatch(); insertPhasePstmt.executeBatch();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment