实体类Phase改为Stage

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

@@ -6,7 +6,7 @@ import lombok.Data;
  * 相位实体类
  */
 @Data
-public class LaneNorm extends Phase{
+public class LaneNorm extends Stage{
     //路口编号
     private String corssId;
     //车道编号

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

@@ -14,7 +14,7 @@ import java.util.List;
  */
 public class TravelInfo {
     // ID
-    private Integer id;
+    private Long id;
 
     // 路口ID
     private String crossId;
@@ -23,7 +23,7 @@ public class TravelInfo {
     private String recordId;
 
     // 记录时间
-    private Date recordTime;
+    private String recordTime;
 
     // 车辆ID
     private Long carId;
@@ -41,16 +41,25 @@ public class TravelInfo {
     private String carType;
 
     // 捕获车道ID
-    private Long firstLineId;
+    private String firstLineId;
 
     // 捕获时速度
     private Double firstSpeed;
 
     // 捕获时间
-    private Date firstTime;
+    private String firstTime;
+
+    // 达到车道ID
+    private String arrivedLineId;
+
+    // 到达时速度
+    private Double arrivedSpeed;
+
+    // 到达时间
+    private String arrivedTime;
 
     // 进入车道ID
-    private Long inCrossLineId;
+    private String inCrossLineId;
 
     // 进入时速度
     private Double inSpeed;
@@ -59,34 +68,34 @@ public class TravelInfo {
     private String inCrossTime;
 
     // 通过车道ID
-    private long outCrossLineId;
+    private String outCrossLineId;
 
     // 通过时速度
-    private double outSpeed;
+    private Double outSpeed;
 
     // 通过时间
-    private Date outCrossTime;
+    private String outCrossTime;
 
     // 驶离车道ID
-    private long awayLineId;
+    private String awayLineId;
 
     // 驶离时速度
-    private double awaySpeed;
+    private Double awaySpeed;
 
     // 驶离时间
-    private Date awayTime;
+    private String awayTime;
 
     // 丢失车道ID
-    private long lastLineId;
+    private String lastLineId;
 
     // 丢失时速度
-    private double lastSpeed;
+    private Double lastSpeed;
 
     // 丢失时间
-    private Date lastTime;
+    private String lastTime;
 
     // 旅行时长，驶离时刻 -  到达时刻， 单位秒
-    private Double crossingTime;
+    private int crossingTime;
 
     // 旅行时点位集合
     private List<Location> locations;
@@ -94,11 +103,11 @@ public class TravelInfo {
     // 备注
     private String remark;
 
-    public Integer getId() {
+    public Long getId() {
         return id;
     }
 
-    public void setId(Integer id) {
+    public void setId(Long id) {
         this.id = id;
     }
 
@@ -118,11 +127,11 @@ public class TravelInfo {
         this.recordId = recordId;
     }
 
-    public Date getRecordTime() {
+    public String getRecordTime() {
         return recordTime;
     }
 
-    public void setRecordTime(Date recordTime) {
+    public void setRecordTime(String recordTime) {
         this.recordTime = recordTime;
     }
 
@@ -166,11 +175,11 @@ public class TravelInfo {
         this.carType = carType;
     }
 
-    public Long getFirstLineId() {
+    public String getFirstLineId() {
         return firstLineId;
     }
 
-    public void setFirstLineId(Long firstLineId) {
+    public void setFirstLineId(String firstLineId) {
         this.firstLineId = firstLineId;
     }
 
@@ -182,19 +191,43 @@ public class TravelInfo {
         this.firstSpeed = firstSpeed;
     }
 
-    public Date getFirstTime() {
+    public String getFirstTime() {
         return firstTime;
     }
 
-    public void setFirstTime(Date firstTime) {
+    public void setFirstTime(String firstTime) {
         this.firstTime = firstTime;
     }
 
-    public Long getInCrossLineId() {
+    public String getArrivedLineId() {
+        return arrivedLineId;
+    }
+
+    public void setArrivedLineId(String arrivedLineId) {
+        this.arrivedLineId = arrivedLineId;
+    }
+
+    public Double getArrivedSpeed() {
+        return arrivedSpeed;
+    }
+
+    public void setArrivedSpeed(Double arrivedSpeed) {
+        this.arrivedSpeed = arrivedSpeed;
+    }
+
+    public String getArrivedTime() {
+        return arrivedTime;
+    }
+
+    public void setArrivedTime(String arrivedTime) {
+        this.arrivedTime = arrivedTime;
+    }
+
+    public String getInCrossLineId() {
         return inCrossLineId;
     }
 
-    public void setInCrossLineId(Long inCrossLineId) {
+    public void setInCrossLineId(String inCrossLineId) {
         this.inCrossLineId = inCrossLineId;
     }
 
@@ -214,83 +247,83 @@ public class TravelInfo {
         this.inCrossTime = inCrossTime;
     }
 
-    public long getOutCrossLineId() {
+    public String getOutCrossLineId() {
         return outCrossLineId;
     }
 
-    public void setOutCrossLineId(long outCrossLineId) {
+    public void setOutCrossLineId(String outCrossLineId) {
         this.outCrossLineId = outCrossLineId;
     }
 
-    public double getOutSpeed() {
+    public Double getOutSpeed() {
         return outSpeed;
     }
 
-    public void setOutSpeed(double outSpeed) {
+    public void setOutSpeed(Double outSpeed) {
         this.outSpeed = outSpeed;
     }
 
-    public Date getOutCrossTime() {
+    public String getOutCrossTime() {
         return outCrossTime;
     }
 
-    public void setOutCrossTime(Date outCrossTime) {
+    public void setOutCrossTime(String outCrossTime) {
         this.outCrossTime = outCrossTime;
     }
 
-    public long getAwayLineId() {
+    public String getAwayLineId() {
         return awayLineId;
     }
 
-    public void setAwayLineId(long awayLineId) {
+    public void setAwayLineId(String awayLineId) {
         this.awayLineId = awayLineId;
     }
 
-    public double getAwaySpeed() {
+    public Double getAwaySpeed() {
         return awaySpeed;
     }
 
-    public void setAwaySpeed(double awaySpeed) {
+    public void setAwaySpeed(Double awaySpeed) {
         this.awaySpeed = awaySpeed;
     }
 
-    public Date getAwayTime() {
+    public String getAwayTime() {
         return awayTime;
     }
 
-    public void setAwayTime(Date awayTime) {
+    public void setAwayTime(String awayTime) {
         this.awayTime = awayTime;
     }
 
-    public long getLastLineId() {
+    public String getLastLineId() {
         return lastLineId;
     }
 
-    public void setLastLineId(long lastLineId) {
+    public void setLastLineId(String lastLineId) {
         this.lastLineId = lastLineId;
     }
 
-    public double getLastSpeed() {
+    public Double getLastSpeed() {
         return lastSpeed;
     }
 
-    public void setLastSpeed(double lastSpeed) {
+    public void setLastSpeed(Double lastSpeed) {
         this.lastSpeed = lastSpeed;
     }
 
-    public Date getLastTime() {
+    public String getLastTime() {
         return lastTime;
     }
 
-    public void setLastTime(Date lastTime) {
+    public void setLastTime(String lastTime) {
         this.lastTime = lastTime;
     }
 
-    public Double getCrossingTime() {
+    public int getCrossingTime() {
         return crossingTime;
     }
 
-    public void setCrossingTime(Double crossingTime) {
+    public void setCrossingTime(int crossingTime) {
         this.crossingTime = crossingTime;
     }
 

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelInfoJob.java

 package com.zhht.irn.job;
 
+import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.kafka.Location;
 import com.zhht.irn.entity.kafka.TravelInfo;
 import com.zhht.irn.entity.mysql.TripsMetric;
@@ -80,6 +81,8 @@ public class TravelInfoJob {
             @Override
             public TravelInfo map(String s) throws Exception {
                 // json 格式字符串转化为 TravelInfo 实体类
+                TravelInfo travelInfo = JSON.parseObject(s,TravelInfo.class);
+
                 return null;
             }
         }).filter(new FilterFunction<TravelInfo>() {

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TripsMetricSink.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.entity.mysql.TripsMetric;
 import com.zhht.irn.utils.MySQLUtils;
 import org.apache.flink.configuration.Configuration;

realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CarSource.java

@@ -3,7 +3,7 @@ package com.zhht.irn.source;
 import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Car;
 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 org.apache.flink.streaming.api.functions.source.SourceFunction;
 

realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CirlceSource.java

@@ -2,7 +2,7 @@ package com.zhht.irn.source;
 
 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.utils.DateUtils;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.util.TimeUtils;
@@ -36,13 +36,13 @@ public class CirlceSource implements SourceFunction<Cycle> {
         Cycle cycle = new Cycle();
         cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
         cycle.setCrossCode("13070201");
-        cycle.setBeginTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
-        cycle.setEndTime(DateUtils.formatTime(date));
+        cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
+        cycle.setEndDateTime(DateUtils.formatTime(date));
         cycle.setDuration(120);
         cycle.setCycleOrder(1);
-        ArrayList<Phase> phases = new ArrayList<>();
+        ArrayList<Stage> phases = new ArrayList<>();
         for(int i=0;i<3;i++){
-            Phase phase = new Phase();
+            Stage phase = new Stage();
             phase.setPhaseValue("10"+i);
             if(i==0){
                 phase.setPhaseValue("1");
@@ -73,7 +73,7 @@ public class CirlceSource implements SourceFunction<Cycle> {
             }
             phases.add(phase);
         }
-        cycle.setPhaseList(phases);
+        cycle.setStageList(phases);
         return cycle;
     }