绿信比

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java

@@ -10,32 +10,32 @@ import java.util.List;
  */
 public class Cycle {
     // 周期编号
-    private int id;
+    private Long id;
 
     // 路口编号
     private String crossCode;
 
     // 周期开始时间
-    private Date beginTime;
+    private String beginTime;
 
     // 周期结束时间
-    private Date endTime;
+    private String endTime;
 
-    private Long duration;
+    private int duration;
 
     // 周期序号
     private int cycleOrder;
     //绿信比
-    private Double greenRadio;
+    private Double greenRatio;
 
     // 周期内相位列表（依次执行）
     private List<Phase> phaseList;
 
-    public int getId() {
+    public Long getId() {
         return id;
     }
 
-    public void setId(int id) {
+    public void setId(Long id) {
         this.id = id;
     }
 
@@ -47,27 +47,27 @@ public class Cycle {
         this.crossCode = crossCode;
     }
 
-    public Date getBeginTime() {
+    public String getBeginTime() {
         return beginTime;
     }
 
-    public void setBeginTime(Date beginTime) {
+    public void setBeginTime(String beginTime) {
         this.beginTime = beginTime;
     }
 
-    public Date getEndTime() {
+    public String getEndTime() {
         return endTime;
     }
 
-    public void setEndTime(Date endTime) {
+    public void setEndTime(String endTime) {
         this.endTime = endTime;
     }
 
-    public Long getDuration() {
+    public int getDuration() {
         return duration;
     }
 
-    public void setDuration(Long duration) {
+    public void setDuration(int duration) {
         this.duration = duration;
     }
 
@@ -87,11 +87,11 @@ public class Cycle {
         this.phaseList = phaseList;
     }
 
-    public Double getGreenRadio() {
-        return greenRadio;
+    public Double getGreenRatio() {
+        return greenRatio;
     }
 
-    public void setGreenRadio(Double greenRadio) {
-        this.greenRadio = greenRadio;
+    public void setGreenRatio(Double greenRatio) {
+        this.greenRatio = greenRatio;
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Phase.java

@@ -14,28 +14,28 @@ public class Phase {
     // 相位编号
     private String phaseValue;
     // 相位开始时间
-    private Date beginTime;
+    private String beginTime;
 
     // 相位结束时间
-    private Date endTime;
+    private String endTime;
 
     // 相位持续时长
-    private Long duration;
+    private int duration;
 
     // 相位绿灯时间
-    private Long green;
+    private int green;
 
     // 相位黄灯时间
-    private Long yellow;
+    private int yellow;
 
     // 相位全红时间
-    private Long allRed;
+    private int allRed;
 
     // 阶段绿信比
     private Double greenRatio;
 
     // 阶段有效绿灯时长
-    private Long validGreen;
+    private int validGreen;
 
     public String getPhaseValue() {
         return phaseValue;
@@ -45,51 +45,51 @@ public class Phase {
         this.phaseValue = phaseValue;
     }
 
-    public Date getBeginTime() {
+    public String getBeginTime() {
         return beginTime;
     }
 
-    public void setBeginTime(Date beginTime) {
+    public void setBeginTime(String beginTime) {
         this.beginTime = beginTime;
     }
 
-    public Date getEndTime() {
+    public String getEndTime() {
         return endTime;
     }
 
-    public void setEndTime(Date endTime) {
+    public void setEndTime(String endTime) {
         this.endTime = endTime;
     }
 
-    public Long getDuration() {
+    public int getDuration() {
         return duration;
     }
 
-    public void setDuration(Long duration) {
+    public void setDuration(int duration) {
         this.duration = duration;
     }
 
-    public Long getGreen() {
+    public int getGreen() {
         return green;
     }
 
-    public void setGreen(Long green) {
+    public void setGreen(int green) {
         this.green = green;
     }
 
-    public Long getYellow() {
+    public int getYellow() {
         return yellow;
     }
 
-    public void setYellow(Long yellow) {
+    public void setYellow(int yellow) {
         this.yellow = yellow;
     }
 
-    public Long getAllRed() {
+    public int getAllRed() {
         return allRed;
     }
 
-    public void setAllRed(Long allRed) {
+    public void setAllRed(int allRed) {
         this.allRed = allRed;
     }
 
@@ -117,11 +117,11 @@ public class Phase {
         this.greenRatio = greenRatio;
     }
 
-    public Long getValidGreen() {
+    public int getValidGreen() {
         return validGreen;
     }
 
-    public void setValidGreen(Long validGreen) {
+    public void setValidGreen(int validGreen) {
         this.validGreen = validGreen;
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/GreenRadioJob.java

@@ -27,44 +27,54 @@ public class GreenRadioJob {
         //                   {"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\":\"101\",\"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\",\"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<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();
-                long duration = cycle.getDuration();
+                int duration = cycle.getDuration();
                 String crossCode = cycle.getCrossCode();
-                double cycleGreenRadio = 0;
+                double cycleGreenRatio = 0;
                 List<Phase> phaseList = cycle.getPhaseList();
                 for (Phase phase : phaseList) {
                     //相位号
                     String phaseValue = phase.getPhaseValue();
                     // 绿灯时间
-                    long green = phase.getGreen();
+                    int green = phase.getGreen();
                     // 黄灯时间
-                    long yellow = phase.getYellow();
+                    int yellow = phase.getYellow();
                     // 全红时间
-                    long allRed = phase.getAllRed();
-                    // 启动损失时间取一般定义：2s
-                    long startLossTime = 2L;
+                    int allRed = phase.getAllRed();
+                    // 启动损失时间，一般定义：2s
+                    int startLossTime = 2;
                     //清场损失时间
-                    long clearLossTime = yellow + allRed - 2;
+                    int clearLossTime = yellow + allRed - 2;
+                    //阶段损失时间
+                    int phaseLossTime = startLossTime + clearLossTime;
                     //阶段有效绿灯时间
-                    long phaseValidGreen = green + yellow + allRed - startLossTime - clearLossTime;
+                    int phaseValidGreen = green + yellow + allRed - phaseLossTime;
+                    System.out.println("******duration:"+duration);
+                    System.out.println("******phaseValidGreen:"+phaseValidGreen);
+
                     //阶段绿信比
-                    double phaseGreenRadio = phaseValidGreen/duration;
+                    double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
+                    System.out.println("******phaseGreenRatio:"+phaseGreenRatio);
                     //组装阶段绿信比
-                    phase.setGreenRatio(phaseGreenRadio);
+                    phase.setGreenRatio(phaseGreenRatio);
                     phase.setValidGreen(phaseValidGreen);
                     //周期绿信比= 各个阶段绿信比之和
-                    cycleGreenRadio = cycleGreenRadio + phaseGreenRadio;
+                    cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
+                    System.out.println("******cycleGreenRatio:"+cycleGreenRatio);
                 }
                 // 组装周期绿信比
-                cycle.setGreenRadio(cycleGreenRadio);
+                cycle.setGreenRatio(cycleGreenRatio);
                 return cycle;
             }
         });

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.utils.MySQLUtils;
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
+import java.util.List;
 
 /**
  * domain traffic
  */
 public class GreenRadioSink extends RichSinkFunction<Cycle> {
     Connection connection;
-    PreparedStatement insertPstmt;
-    PreparedStatement updatePstmt;
+    PreparedStatement insertCyclePstmt;
+    PreparedStatement insertPhasePstmt;
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         // path暂定为固定值
-        String path = "F:\\workspace\\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);
-        insertPstmt = connection.prepareStatement("insert into app_cycle_green_ratio(id,cross_code,green_radio) values (?,?,?)");
-        //updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
+        String cycleSql="insert into app_cycle_green_ratio(cycle_id,cross_code,begin_time,end_time,duration,green_ratio) values (?,?,?,?,?,?)";
+        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 (?,?,?,?,?,?,?,?)";
+        insertPhasePstmt = connection.prepareStatement(phaseSql);
     }
     @Override
     public void close() throws Exception {
         super.close();
-        if(insertPstmt != null) insertPstmt.close();
-        if(updatePstmt != null) updatePstmt.close();
+        if(insertCyclePstmt != null) insertCyclePstmt.close();
+        if(insertPhasePstmt != null) insertPhasePstmt.close();
         if(connection != null) connection.close();
     }
 
@@ -39,11 +42,29 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
      * 1000w的数据  1000w次
      */
     @Override
-    public void invoke(Cycle value, Context context) throws Exception {
-        System.out.println("=====invoke======" + value.getId() + "-->" + value.getCrossCode());
-        insertPstmt.setInt(1, value.getId());
-        insertPstmt.setString(2, value.getCrossCode());
-        insertPstmt.setDouble(3,value.getGreenRadio());
-        insertPstmt.execute();
+    public void invoke(Cycle cycle, Context context) throws Exception {
+        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.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);
+            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.setInt(6,cycle.getDuration());
+            insertPhasePstmt.setDouble(7,phase.getGreenRatio());
+            insertPhasePstmt.setInt(8,phase.getValidGreen());
+            insertPhasePstmt.addBatch();
+        }
+        insertPhasePstmt.executeBatch();
+        insertPhasePstmt.clearBatch();
     }
 }