Merge remote-tracking branch 'origin/master'

realtime/hologram-streaming/src/main/java/com/zhht/irn/consts/Constant.java

@@ -16,5 +16,5 @@ public class Constant {
     // 周期计算的延迟时间 15秒
     public static final int CYCLE_CALCULATE_LATENESS = 15 * 1000;
 
-    public static final int INSERT_BATCH_SIZE = 1000;
+    public static final int INSERT_BATCH_SIZE = 100;
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalDirectionSecondMetricFunction.java

 package com.zhht.irn.functions;
 
-import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
 import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
@@ -10,7 +9,7 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
 import org.apache.flink.api.common.state.MapState;
 import org.apache.flink.api.common.state.MapStateDescriptor;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
 import org.apache.flink.util.Collector;
 
 import java.util.*;
@@ -24,7 +23,7 @@ import java.util.stream.Collectors;
  * @author 吴延飞
  * @date 2022-11-15 11:23:00
  */
-public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle, TravelEvent, DirectionEvalSecondMetric> {
+public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, DirectionEvalSecondMetric> {
 
     private MapState<String, List<TravelEvent>> mapState;
 
@@ -38,27 +37,15 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
     }
 
     @Override
-    public void processElement(Cycle cycle, TravelEvent travelEvent, ProcessJoinFunction<Cycle, TravelEvent, DirectionEvalSecondMetric>.Context ctx, Collector<DirectionEvalSecondMetric> out) throws Exception {
-        // 进入该方法表明，周期已经到了，而且周期已经向过车记录interval join了[-9,1]分钟的旅行事件数据
-
-        // 先累计旅行事件数据到 mapState 中, 只累计到达和进入事件，其他类型计算用不上
-        List<TravelEvent> cachedTravelEvent = mapState.get(cycle.getCrossCode());
-        if(cachedTravelEvent == null) {
-            cachedTravelEvent = new ArrayList<>();
-            mapState.put(cycle.getCrossCode(), cachedTravelEvent);
-        }
-        if(travelEvent.getEventType() == EventType.ARRIVED || travelEvent.getEventType() == EventType.INCROSS
-            || travelEvent.getEventType() == EventType.STOP || travelEvent.getEventType() == EventType.STARTUP)
-        cachedTravelEvent.add(travelEvent);
-
-        // 确定触发条件
-        if(Long.parseLong(travelEvent.getEventTime()) - Constant.CYCLE_CALCULATE_LATENESS > Long.parseLong(cycle.getEndDateTime())) {
+    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, DirectionEvalSecondMetric>.Context context, Collector<DirectionEvalSecondMetric> collector) throws Exception {
+        // 周期数据进入，开始计算
         long cycleStart = Long.parseLong(cycle.getBeginDateTime());
         long cycleEnd = Long.parseLong(cycle.getEndDateTime());
-
+        List<TravelEvent> cachedTravelEvent = mapState.get(cycle.getCrossCode());
+        Map<String, Boolean> globalRecordStopStatusMap = new HashMap<>();
         // 计算每个车道的到达流量, 按秒计算
         for(long start = cycleStart; start <= cycleEnd; start += 1000) {
-                // 首先过滤出到当前秒的事件数据
+            // 首先过滤出到当前周期的事件数据
             long finalStart = start;
             if (cachedTravelEvent != null) {
                 List<TravelEvent> currentEventList = cachedTravelEvent.stream().filter(event -> {
@@ -78,7 +65,7 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                 });
 
 
-                    // 计算两个指标，一个 “发生到达，未进入的车辆数”， 一个“发生到达数”，粒度计算到车道级
+                // 计算一个“发生到达数”，粒度计算到车道级
                 for(Map.Entry<String, List<TravelEvent>> entry : lineMap.entrySet()) {
                     List<TravelEvent> travelEventList = entry.getValue();
 
@@ -92,14 +79,18 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                     for(String recordId : listState.get()) {
                         recordStopStatusMap.put(recordId, true);
                     }
+                    // 严格一点，这里要对事件进行时间的排序, 因为乱序会导致车辆的状态不正确
+                    Collections.sort(travelEventList, Comparator.comparing(TravelEvent::getEventTime));
+
                     // 再处理本周期内发生的事件
                     for(int i = 0; i < travelEventList.size(); i ++) {
-                            // 严格一点，这里要对事件进行时间的排序，暂且不做
                         TravelEvent e = travelEventList.get(i);
                         if(e.getEventType() == EventType.STOP) {
                             recordStopStatusMap.put(e.getRecordId(), true);
+                            globalRecordStopStatusMap.put(e.getRecordId(), true);
                         } else if(e.getEventType() == EventType.STARTUP) {
                             recordStopStatusMap.put(e.getRecordId(), false);
+                            globalRecordStopStatusMap.put(e.getRecordId(), false);
                         }
                     }
 
@@ -112,24 +103,13 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                         }
                     }).collect(Collectors.toList());
 
-
-
-                        List<TravelEvent> lineArrivedAndInCrossList = travelEventList.stream().filter(record -> {
-                            return record.getEventType() == EventType.INCROSS && recordIdSet.contains(record.getRecordId());
-                        }).collect(Collectors.toList());
-
                     Set<String> arrivedRecordIdSet = new HashSet<>();
-                        Set<String> arrivedAndInCrossRecordIdSet = new HashSet<>();
 
                     for(TravelEvent travel:  lineArrivedList) {
                         arrivedRecordIdSet.add(travel.getRecordId());
                     }
-                        for(TravelEvent travel: lineArrivedAndInCrossList) {
-                            arrivedAndInCrossRecordIdSet.add(travel.getRecordId());
-                        }
 
                     int lineArrivedCount = arrivedRecordIdSet.size();
-                        int lineArrivedAndInCrossCount = arrivedAndInCrossRecordIdSet.size();
 
                     int stopFlagCount = 0;
                     int stopCount = 0;
@@ -138,7 +118,6 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                         stopCount ++;
                     }
 
-
                     DirectionEvalSecondMetric metric = new DirectionEvalSecondMetric();
                     metric.setCycleStart(cycleStart);
                     metric.setCycleEnd(cycleEnd);
@@ -147,17 +126,45 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
                     metric.setCurrentSecond(start);
                     metric.setCycleOrder(cycle.getCycleOrder());
                     metric.setAccArrivedFlow(lineArrivedCount);
-                        metric.setAccQueueLength(lineArrivedCount - lineArrivedAndInCrossCount);
+                    metric.setAccQueueLength(stopFlagCount);
                     metric.setStopFlagCount(stopFlagCount);
                     metric.setAccStopCount(stopCount);
+                    collector.collect(metric);
+                }
+            }
+        }
+
+        // 将本周期结束后还处于停车状态的车辆放到listState中
+        listState.clear();
+        for(Map.Entry<String, Boolean> en : globalRecordStopStatusMap.entrySet()) {
+            if(en.getValue() == true) {
+                System.out.println("stay stop status car record id:" + en.getKey());
+                listState.add(en.getKey());
+            }
+        }
 
-                        out.collect(metric);
+        if(cachedTravelEvent != null) {
+            List<TravelEvent> newOne = new ArrayList<>();
+            for(TravelEvent t : cachedTravelEvent) {
+                // 删除掉这个周期结束时间之间的数据
+                if(Long.parseLong(t.getEventTime()) >= Long.parseLong(cycle.getEndDateTime())) {
+                    newOne.add(t);
                 }
             }
+            cachedTravelEvent = newOne;
+            mapState.put(cycle.getCrossCode(), cachedTravelEvent);
+        }
     }
 
-            // 计算后要清空
-            cachedTravelEvent.clear();
+    @Override
+    public void processElement2(TravelEvent travelEvent, KeyedCoProcessFunction<String, Cycle, TravelEvent, DirectionEvalSecondMetric>.Context context, Collector<DirectionEvalSecondMetric> collector) throws Exception {
+        List<TravelEvent> cachedTravelEvent = mapState.get(travelEvent.getCrossId());
+        if(cachedTravelEvent == null) {
+            cachedTravelEvent = new ArrayList<>();
+            mapState.put(travelEvent.getCrossId(), cachedTravelEvent);
         }
+        if(travelEvent.getEventType() == EventType.ARRIVED || travelEvent.getEventType() == EventType.INCROSS
+                || travelEvent.getEventType() == EventType.STOP || travelEvent.getEventType() == EventType.STARTUP)
+            cachedTravelEvent.add(travelEvent);
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalLaneMetricProcessFunction2.java

-package com.zhht.irn.functions;
-
-import com.zhht.irn.entity.Cycle;
-import com.zhht.irn.entity.TravelEvent;
-import org.apache.flink.api.common.state.MapState;
-import org.apache.flink.api.common.state.MapStateDescriptor;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
-import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
-import org.apache.flink.util.Collector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * 计算车道级指标的函数
- *
- */
-public class CalLaneMetricProcessFunction2 extends KeyedCoProcessFunction<String, Cycle, TravelEvent, Object> {
-    private static final Logger logger = LoggerFactory.getLogger(CalLaneMetricProcessFunction2.class);
-    // state 管理当前周期中存储的旅行事件
-    private MapState<String, List<TravelEvent>> cycleTravel;
-
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        cycleTravel = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravel", String.class, Object.class));
-    }
-
-    @Override
-    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, Object>.Context ctx, Collector<Object> out) throws Exception {
-        logger.info("进来个周期数据了：" + cycle.toString());
-        // 在此进行指标计算
-    }
-
-    @Override
-    public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, Object>.Context ctx, Collector<Object> out) throws Exception {
-        logger.info("进来个事件数据了：" + event.toString());
-        if(cycleTravel.get(event.getCrossId()) == null) {
-            cycleTravel.put(event.getCrossId(), new ArrayList<>());
-        }
-        cycleTravel.get(event.getCrossId()).add(event);
-
-    }
-}

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

@@ -4,15 +4,20 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
+import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
 import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
 import com.zhht.irn.sink.DirectionEvalSecondSink;
+import com.zhht.irn.source.CycleMockSource;
+import com.zhht.irn.source.TravelEventMockSource;
 import com.zhht.irn.utils.FlinkUtils;
 import com.zhht.irn.utils.StringUtils;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.KeyedStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
 import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,7 +43,7 @@ public class DirectionEvalJob {
         DataStream<Cycle> cycleStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
-                    System.out.println("信控周期->" + record);
+                    System.out.println(record);
                     return JSON.parseObject(record, Cycle.class);
                 });
 
@@ -46,7 +51,7 @@ public class DirectionEvalJob {
         DataStream<TravelEvent> travelEventStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
-                    System.out.println("旅行事件->" + record);
+                    System.out.println(record);
                     return JSON.parseObject(record, TravelEvent.class);
                 }).filter(event -> {
                     return StringUtils.isNotEmpty(event.getEventLineId());
@@ -69,10 +74,9 @@ public class DirectionEvalJob {
         // 周期数据join 旅行事件
         cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
                 .keyBy(record -> record.getCrossCode())
-                .intervalJoin(travelEventKeyedStream)
-                .between(Time.minutes(-9), Time.minutes(1))
+                .connect(travelEventKeyedStream)
                 .process(new CalDirectionSecondMetricFunction())
-                .addSink(new DirectionEvalSecondSink());
+                .addSink(new DirectionEvalSecondSink()).setParallelism(4);
 
         // 旅行数据流
         env.execute("DirectionEvalJob");

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/DirectionEvalSecondSink.java

@@ -44,7 +44,6 @@ public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecon
 
     @Override
     public void invoke(DirectionEvalSecondMetric metric, Context context) throws Exception {
-        System.out.println("===== insert metric {" + metric + "}");
         insertPstmt.setString(1, metric.getCrossId());
         insertPstmt.setString(2, metric.getLineId());
         insertPstmt.setLong(3, metric.getCycleOrder());

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelMetricSink.java

@@ -38,6 +38,7 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     }
     @Override
     public void close() throws Exception {
+        System.out.println("执行close方法了");
         super.close();
         if(insertPstmt != null) insertPstmt.close();
         if(connection != null) connection.close();
@@ -45,7 +46,6 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
 
     @Override
     public void invoke(Travel travel, Context context) throws Exception {
-        System.out.println("=====insert travel ======" + travel);
         insertPstmt.setString(1, travel.getRecordId());
         insertPstmt.setString(2, travel.getCrossId());
         insertPstmt.setDate(3, new Date(Long.parseLong(travel.getRecordTime())));

realtime/hologram-streaming/src/main/resources/kafka.properties

 bootstrap.servers=139.9.157.176:9092
 kafka.input.topics=trips_info
 auto.offset.reset=latest
-group.id=wuyanfei
+group.id=TravelInfoJob
 checkpoint.path=file:///data/flink/state
-checkpoint.interval=5000
+checkpoint.interval=60000
 enable.auto.commit=false
\ No newline at end of file

realtime/hologram-streaming/src/main/resources/mysql.properties

 driver=com.mysql.jdbc.Driver
-host=127.0.0.1
-port=13305
+host=10.243.0.26
+port=3306
 username=root
 password=mima
 database=test