Merge remote-tracking branch 'origin/master'

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

@@ -11,24 +11,32 @@ import com.zhht.irn.source.CarSource;
 import com.zhht.irn.source.CirlceSource;
 import com.zhht.irn.utils.DateUtils;
 import com.zhht.irn.utils.MySQLUtils;
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.CoGroupFunction;
 import org.apache.flink.api.common.functions.FilterFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 
+import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+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;
 import org.apache.flink.util.Collector;
 
 import java.sql.Connection;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
 /**
@@ -74,7 +82,7 @@ public class LaneNormJob {
                 keyBy(new KeySelector<Car,String>() {
             @Override
             public String getKey(Car car) throws Exception {
-                return car.getCrossId();
+                return car.toString();
             }
         }).filter(new FilterFunction<Car>() {
             @Override
@@ -87,6 +95,43 @@ public class LaneNormJob {
             }
         });
 
+        cycleDataStreamSource.join(carStream)
+                .where(new KeySelector<Cycle, String>() {
+                    @Override
+                    public String getKey(Cycle cycle) throws Exception {
+                        return cycle.getCrossCode();
+                    }
+                }).equalTo(new KeySelector<Car, String>() {
+                    @Override
+                    public String getKey(Car car) throws Exception {
+                        return car.getCrossId();
+                    }
+                }).window(new WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window>() {
+                    @Override
+                    public Collection<Window> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, Car> cycleCarTaggedUnion, long l, WindowAssignerContext windowAssignerContext) {
+                        // 这里以周期流的开始和结束时间作为一个Window
+                        Cycle currentCycle = cycleCarTaggedUnion.getOne();
+                        Car car = cycleCarTaggedUnion.getTwo();
+                        TimeWindow tw = new TimeWindow(DateUtils.stringToDate(currentCycle.getBeginDateTime(), "yyyy-MM-dd HH:mm:ss").getTime(), DateUtils.stringToDate(currentCycle.getEndDateTime(), "yyyy-MM-dd HH:mm:ss").getTime());
+                        return null;
+                    }
+
+                    @Override
+                    public Trigger<CoGroupedStreams.TaggedUnion<Cycle, Car>, Window> getDefaultTrigger(StreamExecutionEnvironment streamExecutionEnvironment) {
+                        return null;
+                    }
+
+                    @Override
+                    public TypeSerializer<Window> getWindowSerializer(ExecutionConfig executionConfig) {
+                        return null;
+                    }
+
+                    @Override
+                    public boolean isEventTime() {
+                        return false;
+                    }
+                });
+
         DataStream<String> apply = cycleDataStreamSource.coGroup(carStream).where(new KeySelector<Cycle, String>() {
             @Override
             public String getKey(Cycle value) throws Exception {