lxb

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

@@ -4,7 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Car;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.LaneNorm;
-import com.zhht.irn.entity.Phase;
+import com.zhht.irn.entity.Stage;
 import com.zhht.irn.sink.LXBSink;
 import com.zhht.irn.sink.LaneNormSink;
 import com.zhht.irn.source.CarSource;
@@ -55,7 +55,7 @@ public class LaneNormJob {
 
        //根据corssid 分组 保证同一路口的数据进行一个进程
         DataStream<Cycle> cycleDataStreamSource = env.addSource(new CirlceSource()).
-                assignTimestampsAndWatermarks(WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> DateUtils.dateToStamp(cycle.getEndTime())))
+                assignTimestampsAndWatermarks(WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> DateUtils.dateToStamp(cycle.getEndDateTime())))
                 .keyBy(new KeySelector<Cycle,String>() {
                     @Override
                     public String getKey(Cycle cycle) throws Exception {
@@ -144,8 +144,8 @@ public class LaneNormJob {
         double cycleGreenRatio = 0;
         //周期有效绿灯时间
         double cycleVaildGreen = 0;
-        List<Phase> phaseList = cycle.getPhaseList();
-        for (Phase phase : phaseList) {
+        List<Stage> phaseList = cycle.getStageList();
+        for (Stage phase : phaseList) {
             //相位号
             String phaseValue = phase.getPhaseValue();
             // 绿灯时间
@@ -175,7 +175,6 @@ public class LaneNormJob {
         // 组装周期绿信比
         cycle.setGreenRatio(cycleGreenRatio);
         //组装周期绿灯有效时间
-        cycle.setVaildGreen(cycleVaildGreen);
         return cycle;
     }
 }