Flink,union双流，union实例，java版本

傷城~ 2021-09-24 10:50 282阅读 0赞

union双流后会生成一个新的流，没有去重

package union;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Author you guess
     * @Date 2020/6/15 18:00
     * @Version 1.0
     * @Desc
     */
    public class UnionDataStreamTest {
    
        private static final Logger LOG = LoggerFactory.getLogger(UnionDataStreamTest.class);
        private static final String[] TYPE1 = {"a","b"};
        private static final String[] TYPE2 = {"a","b"};
    
        public static void main(String[] args) throws Exception {
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
            DataStreamSource<Tuple2<String, Integer>> orderSource1 = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
                private volatile boolean isRunning = true;
                private final Random random = new Random();
    
                @Override
                public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                    while (isRunning) {
                        TimeUnit.SECONDS.sleep(1);
                        Tuple2<String, Integer> tuple2 = Tuple2.of(TYPE1[random.nextInt(TYPE1.length)], random.nextInt(10));
                        System.out.println(new Date() + ",orderSource1提交元素:" + tuple2);
                        ctx.collect(tuple2);                }
                }
    
                @Override
                public void cancel() {
                    isRunning = false;
                }
    
            }, "orderSource1");
    
            DataStreamSource<Tuple2<String, Integer>> orderSource2 = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
                private volatile boolean isRunning = true;
                private final Random random = new Random();
    
                @Override
                public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                    while (isRunning) {
                        TimeUnit.SECONDS.sleep(1);
                        Tuple2<String, Integer> tuple2 = Tuple2.of(TYPE2[random.nextInt(TYPE2.length)], random.nextInt(10));
                        System.out.println(new Date() + ",orderSource2提交元素:" + tuple2);
                        ctx.collect(tuple2);                }
                }
    
                @Override
                public void cancel() {
                    isRunning = false;
                }
    
            }, "orderSource2");
    
    
            /**
             * Creates a new {@link DataStream} by merging {@link DataStream} outputs of
             * the same type with each other. The DataStreams merged using this operator
             * will be transformed simultaneously.
             *
             * @param streams
             *            The DataStreams to union output with.
             * @return The {@link DataStream}.
             */
            orderSource1.union(orderSource2)
                    .timeWindowAll(Time.seconds(5))//return windowAll(TumblingProcessingTimeWindows.of(size));
                    .apply(new
                                   AllWindowFunction<Tuple2<String, Integer>, Map<String, Integer>, TimeWindow>() {
                                       @Override
                                       public void apply(TimeWindow window, Iterable<Tuple2<String, Integer>> values, Collector<Map<String, Integer>> out) throws Exception {
    //                                       System.out.println("-------左闭右开-------");
    //                                       System.out.println(new Date(window.getStart()) + "," + new Date(window.getEnd()));
                                           Map<String, Integer> accumulator = new HashMap<>();
                                           for (Tuple2<String, Integer> element : values) {
                                               accumulator.put(element.f0, element.f1 + accumulator.getOrDefault(element.f0, 0));
                                           }//for
                                           out.collect(accumulator);
                                       }
                                   }).print();
    
    //        Wed Jun 17 12:22:10 CST 2020,orderSource2提交元素:(a,6)
    //        Wed Jun 17 12:22:10 CST 2020,orderSource1提交元素:(b,5)
    //        Wed Jun 17 12:22:11 CST 2020,orderSource2提交元素:(a,4)
    //        Wed Jun 17 12:22:11 CST 2020,orderSource1提交元素:(a,6)
    //        Wed Jun 17 12:22:12 CST 2020,orderSource2提交元素:(b,0)
    //        Wed Jun 17 12:22:12 CST 2020,orderSource1提交元素:(b,9)
    //        Wed Jun 17 12:22:13 CST 2020,orderSource2提交元素:(b,6)
    //        Wed Jun 17 12:22:13 CST 2020,orderSource1提交元素:(b,2)
    //        Wed Jun 17 12:22:14 CST 2020,orderSource2提交元素:(b,1)
    //        Wed Jun 17 12:22:14 CST 2020,orderSource1提交元素:(b,9)
    //        12> {a=16, b=32}
    
    
    
    
    //        orderSource1.union(orderSource2).print();
    //        Wed Jun 17 12:20:10 CST 2020,orderSource2提交元素:(b,9)
    //        Wed Jun 17 12:20:10 CST 2020,orderSource1提交元素:(a,3)
    //        6> (a,3)
    //        10> (b,9)
    //        Wed Jun 17 12:20:11 CST 2020,orderSource1提交元素:(b,1)
    //        Wed Jun 17 12:20:11 CST 2020,orderSource2提交元素:(a,3)
    //        7> (b,1)
    //        11> (a,3)
    
            env.execute("Flink UnionDataStreamTest by Java ");
        }
    }

修改：

source1每1s发1个消息，source2每2s发1个消息，union后直接print，不计算也不使用window：

只要有1个消息来了就输出：

Wed Jun 17 12:25:13 CST 2020,orderSource1提交元素:(a,2)  
11> (a,2)  
Wed Jun 17 12:25:14 CST 2020,orderSource2提交元素:(a,4)  
Wed Jun 17 12:25:14 CST 2020,orderSource1提交元素:(a,7)  
12> (a,7)  
10> (a,4)  
Wed Jun 17 12:25:15 CST 2020,orderSource1提交元素:(a,9)  
1> (a,9)  
Wed Jun 17 12:25:16 CST 2020,orderSource2提交元素:(b,9)  
Wed Jun 17 12:25:16 CST 2020,orderSource1提交元素:(b,0)  
2> (b,0)  
11> (b,9)

输出实例3:没有去重

Wed Jun 17 12:38:02 CST 2020,orderSource1提交元素:(a,1)  
Wed Jun 17 12:38:02 CST 2020,orderSource2提交元素:(a,1)  
10> (a,1)  
11> (a,1)  
Wed Jun 17 12:38:03 CST 2020,orderSource2提交元素:(a,1)  
Wed Jun 17 12:38:03 CST 2020,orderSource1提交元素:(a,1)  
11> (a,1)  
12> (a,1)

flink版本1.9.2,java版本1.8