Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 0e8073be by 余根龙
Options

修改字段名称

parent 7bc29bf4
Showing with 45 additions and 49 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java
package com.zhht.irn.entity;
import java.util.Date;
import java.util.List;
/**
......@@ -16,10 +15,10 @@ public class Cycle {
private String crossCode;
// 周期开始时间
private String beginTime;
private String beginDateTime;
// 周期结束时间
private String endTime;
private String endDateTime;
private int duration;
......@@ -29,7 +28,7 @@ public class Cycle {
private Double greenRatio;
// 周期内相位列表（依次执行）
private List<Phase> phaseList;
private List<Stage> stageList;
public Long getId() {
return id;
......@@ -47,20 +46,20 @@ public class Cycle {
this.crossCode = crossCode;
}
public String getBeginTime() {
return beginTime;
public String getBeginDateTime() {
return beginDateTime;
}
public void setBeginTime(String beginTime) {
this.beginTime = beginTime;
public void setBeginDateTime(String beginDateTime) {
this.beginDateTime = beginDateTime;
}
public String getEndTime() {
return endTime;
public String getEndDateTime() {
return endDateTime;
}
public void setEndTime(String endTime) {
this.endTime = endTime;
public void setEndDateTime(String endDateTime) {
this.endDateTime = endDateTime;
}
public int getDuration() {
......@@ -79,12 +78,12 @@ public class Cycle {
this.cycleOrder = cycleOrder;
}
public List<Phase> getPhaseList() {
return phaseList;
public List<Stage> getStageList() {
return stageList;
}
public void setPhaseList(List<Phase> phaseList) {
this.phaseList = phaseList;
public void setStageList(List<Stage> stageList) {
this.stageList = stageList;
}
public Double getGreenRatio() {
......
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,7 +5,7 @@ import java.util.Date;
/**
* 相位实体类
*/
public class Phase {
public class Stage {
//周期ID
private Long cycleId;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/FlinkKafka2MysqlApp.java
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.Phase;
import com.zhht.irn.entity.Stage;
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.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import javax.xml.crypto.Data;
import java.util.List;
/**
......@@ -42,8 +39,8 @@ public class FlinkKafka2MysqlApp {
//绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
Cycle cycle = JSON.parseObject(s, Cycle.class);
Long allValidGreen = 0L;
List<Phase> phaseList = cycle.getPhaseList();
for (Phase phase : phaseList) {
List<Stage> phaseList = cycle.getStageList();
for (Stage phase : phaseList) {
// 绿灯时间
long green = phase.getGreen();
// 黄灯时间
......@@ -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);
}
});
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/GreenRadioJob.java
......@@ -2,7 +2,7 @@ package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
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 org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
......@@ -20,14 +20,14 @@ public class GreenRadioJob {
// DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
// 目前假设从Kafka接入的数据为json格式的字符串
// Cycle Topic 的消息格式如下：
// {"Id":"","CrossCode":"","BeginTime":"","EndTime":"","Duration":"","CycleOrder":"",
// "detail":[{"PhaseValue":"101","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"102","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"103","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"104","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}]
// {"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\":\"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\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"stageList\":[{\"phaseValue\":\"101\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"beginDateTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"102\",\"beginDateTime\":\"2020-08-02 08:00:00\",\"beginDateTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
DataStream<Cycle> lxbStream = stream.map(new MapFunction<String, Cycle>() {
@Override
public Cycle map(String s) throws Exception {
......@@ -42,16 +42,16 @@ public class GreenRadioJob {
int duration = cycle.getDuration();
String crossCode = cycle.getCrossCode();
double cycleGreenRatio = 0;
List<Phase> phaseList = cycle.getPhaseList();
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
int startLossTime = 2;
//清场损失时间
......@@ -67,8 +67,8 @@ public class GreenRadioJob {
double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
System.out.println("******phaseGreenRatio:"+phaseGreenRatio);
//组装阶段绿信比
phase.setGreenRatio(phaseGreenRatio);
phase.setValidGreen(phaseValidGreen);
stage.setGreenRatio(phaseGreenRatio);
stage.setValidGreen(phaseValidGreen);
//周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
System.out.println("******cycleGreenRatio:"+cycleGreenRatio);
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java
package com.zhht.irn.sink;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
......@@ -46,19 +46,19 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
System.out.println("=====invoke======" + cycle.getId() + "-->" + cycle.getCrossCode());
insertCyclePstmt.setLong(1, cycle.getId());
insertCyclePstmt.setString(2, cycle.getCrossCode());
insertCyclePstmt.setString(3, cycle.getBeginTime());
insertCyclePstmt.setString(4, cycle.getEndTime());
insertCyclePstmt.setString(3, cycle.getBeginDateTime());
insertCyclePstmt.setString(4, cycle.getEndDateTime());
insertCyclePstmt.setInt(5,cycle.getDuration());
insertCyclePstmt.setDouble(6,cycle.getGreenRatio());
insertCyclePstmt.execute();
List<Phase> phaseList = cycle.getPhaseList();
for(int i=0;i<phaseList.size();i++){
Phase phase = phaseList.get(i);
List<Stage> stageList = cycle.getStageList();
for(int i=0;i<stageList.size();i++){
Stage phase = stageList.get(i);
insertPhasePstmt.setLong(1,cycle.getId());
insertPhasePstmt.setString(2,phase.getCrossCode());
insertPhasePstmt.setString(3,phase.getPhaseValue());
insertPhasePstmt.setString(4, cycle.getBeginTime());
insertPhasePstmt.setString(5, cycle.getEndTime());
insertPhasePstmt.setString(4, cycle.getBeginDateTime());
insertPhasePstmt.setString(5, cycle.getEndDateTime());
insertPhasePstmt.setInt(6,cycle.getDuration());
insertPhasePstmt.setDouble(7,phase.getGreenRatio());
insertPhasePstmt.setInt(8,phase.getValidGreen());
......
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