connected 链接方式

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

@@ -3,7 +3,7 @@ package com.zhht.irn.flow;
 import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
-import com.zhht.irn.functions.CalLaneMetricProcessFunction;
+import com.zhht.irn.functions.CalLaneMetricProcessFunction2;
 import com.zhht.irn.source.CycleMockSource;
 import com.zhht.irn.source.TravelEventMockSource;
 import com.zhht.irn.trigger.CycleTimeTrigger;
@@ -15,9 +15,12 @@ import org.apache.flink.api.common.functions.CoGroupFunction;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
+import org.apache.flink.streaming.api.datastream.ConnectedStreams;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
 import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
+import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.triggers.Trigger;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.streaming.api.windowing.windows.Window;
@@ -79,23 +82,25 @@ public class LaneMetricJob {
                 .assignTimestampsAndWatermarks(cycleWatermarkStrategy);
 
         // 通过join操作，获取一个Tuple<List<TravelEvent>, Cycle> 类型的数据流
-        DataStream<Tuple2<Cycle, TravelEvent>> joinStream = cycleStream.coGroup(eventStream)
-                // 根据路口分组
-                .where(cycle  -> cycle.getCrossCode())
-                .equalTo(travelEvent -> travelEvent.getCrossId())
-                // 根据周期来进行窗口划分
-                .window(new CycleWindowAssigner())
-                .trigger(new CycleTimeTrigger())
-                //
-                .apply(new CoGroupFunction<Cycle, TravelEvent, Tuple2<Cycle, TravelEvent>>() {
-                    @Override
-                    public void coGroup(Iterable<Cycle> iterable, Iterable<TravelEvent> iterable1, Collector<Tuple2<Cycle, TravelEvent>> collector) throws Exception {
-                        System.out.println(1);
-                    }
-                });
 
-        joinStream.keyBy(value -> value.f0.getCrossCode())
-                .process(new CalLaneMetricProcessFunction())
+
+//        DataStream<Tuple2<Cycle, TravelEvent>> joinStream = cycleStream.coGroup(eventStream)
+//                // 根据路口分组
+//                .where(cycle  -> cycle.getCrossCode())
+//                .equalTo(travelEvent -> travelEvent.getCrossId())
+//                // 根据周期来进行窗口划分
+//                .window(new CycleWindowAssigner())
+//                .trigger(new CycleTimeTrigger())
+//                .allowedLateness(Time.minutes(10))
+//                //
+//                .apply(new CoGroupFunction<Cycle, TravelEvent, Tuple2<Cycle, TravelEvent>>() {
+//                    @Override
+//                    public void coGroup(Iterable<Cycle> iterable, Iterable<TravelEvent> iterable1, Collector<Tuple2<Cycle, TravelEvent>> collector) throws Exception {
+//                        collector.collect(Tuple2.of(iterable.iterator().next(), iterable1.iterator().next()));
+//                    }
+//                });
+        ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleStream.connect(eventStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
+        cycleTravelEventConnectedStreams.process(new CalLaneMetricProcessFunction2())
                 .print();
 
         env.execute();

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);
+
+    }
+}