Union

定义

Transformation Description
DataStream* → DataStream Union of two or more data streams creating a new stream containing all the elements from all the streams.Creates a new DataStream by merging DataStream outputs of the same type with each other. The DataStreams merged using this operator will be transformed simultaneously.

说明

两个或以上的流合并成一个，这两个流的数据类型必须一样

样例

代码

  1. public class UnionDemo {
  3.     public static void main(String[] args) throws Exception{
  4.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  5.         env.setParallelism(1);
  7.         DataStream<Tuple2<String, Integer>> orangeStream = env.addSource(new DataSource("orangeStream"));
  8.         DataStream<Tuple2<String, Integer>> greenStream = env.addSource(new DataSource("greenStream"));
  10.         orangeStream.union(greenStream).print("union");
  11.         env.execute("Union Demo");
  12.     }
  14.     private static class DataSource extends RichParallelSourceFunction<Tuple2<String, Integer>> {
  15.         private volatile boolean running = true;
  16.         private volatile String name;
  18.         public DataSource(String name) {
  19.             this.name = name;
  20.         }
  22.         @Override
  23.         public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
  24.             Random random = new Random();
  25.             int bound = 100;
  26.             final long numElements = 6;
  27.             int i = 0;
  29.             while (running && i < numElements) {
  30.                 Thread.sleep(1500);
  31.                 Tuple2 data = new Tuple2<>("foo", random.nextInt(bound));
  32.                 ctx.collect(data);
  33.                 System.out.println(Thread.currentThread().getId() + "-" + this.name + "-sand data:" + data);
  34.                 i++;
  35.             }
  36.         }
  38.         @Override
  39.         public void cancel() {
  40.             running = false;
  41.         }
  42.     }
  43. }

输出结果

  1. 59-greenStream-sand data:(foo,6)
  2. 58-orangeStream-sand data:(foo,33)
  3. union> (foo,33)
  4. union> (foo,6)
  5. 59-greenStream-sand data:(foo,99)
  6. 58-orangeStream-sand data:(foo,99)
  7. union> (foo,99)
  8. union> (foo,99)
  9. 58-orangeStream-sand data:(foo,23)
  10. 59-greenStream-sand data:(foo,36)
  11. union> (foo,23)
  12. union> (foo,36)
  13. 59-greenStream-sand data:(foo,78)
  14. 58-orangeStream-sand data:(foo,7)
  15. union> (foo,7)
  16. union> (foo,78)
  17. 59-greenStream-sand data:(foo,7)
  18. 58-orangeStream-sand data:(foo,58)
  19. union> (foo,58)
  20. union> (foo,7)
  21. 59-greenStream-sand data:(foo,66)
  22. 58-orangeStream-sand data:(foo,80)
  23. union> (foo,80)
  24. union> (foo,66)

说明

DataSource类随机发送数据作为数据源，将两个流做union输出

Connect&CoFlatMap

定义

Transformation Description
DataStream,DataStream → ConnectedStreams “Connects” two data streams retaining their types. Connect allowing for shared state between the two streams.
ConnectedStreams → DataStream Similar to flatMap on a connected data stream

说明

Connect可以连接两个不同数据类型的流，这个是和union最主要的区别，其次union支持2个以上流的合并，而Connect只支持2个流；可以借助CoFlatMap将不同类型的流进行类型统一等操作。

样例

代码

  1. public class ConnectDemo {
  2.     public static void main(String[] args) throws Exception{
  3.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  4.         env.setParallelism(1);
  6.         DataStream<Tuple2<String, Integer>> orangeStream = env.addSource(new DataSource1());
  7.         DataStream<Tuple3<String, Integer, Integer>> greenStream = env.addSource(new DataSource2());
  9.         orangeStream.connect(greenStream).flatMap(new CoFlatMapFunction<Tuple2<String, Integer>, Tuple3<String, Integer, Integer>, Object>() {
  10.             @Override
  11.             public void flatMap1(Tuple2<String, Integer> value, Collector<Object> out) throws Exception {
  12.                 if (!value.f0.contains("@")){
  13.                     out.collect(new Tuple3<>(value.f0, value.f1, RandomUtils.nextInt(0, value.f1)));
  14.                 }
  15.             }
  17.             @Override
  18.             public void flatMap2(Tuple3<String, Integer, Integer> value, Collector<Object> out) throws Exception {
  19.                 for (String s : value.f0.split("@")) {
  20.                     out.collect(new Tuple3<>(s, value.f1, value.f2));
  21.                 }
  22.             }
  23.         }).print("Connect");
  25.         env.execute("Connect Demo");
  26.     }
  28.     private static class DataSource1 extends RichParallelSourceFunction<Tuple2<String, Integer>> {
  29.         private volatile boolean running = true;
  31.         @Override
  32.         public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
  33.             int bound = 50;
  34.             String[] keys = new String[]{"foo@xxyz", "bar", "baz"};
  36.             final long numElements = RandomUtils.nextLong(10, 20);
  37.             int i = 0;
  39.             while (running && i < numElements) {
  40.                 Thread.sleep(RandomUtils.nextLong(1, 5) * 1000L);
  41.                 Tuple2 data = new Tuple2<>(keys[RandomUtils.nextInt(0, 3)], RandomUtils.nextInt(0, bound));
  42.                 ctx.collect(data);
  43.                 System.out.println(Thread.currentThread().getId() + "-sand data:" + data);
  44.                 i++;
  45.             }
  46.         }
  48.         @Override
  49.         public void cancel() {
  50.             running = false;
  51.         }
  52.     }
  54.     private static class DataSource2 extends RichParallelSourceFunction<Tuple3<String, Integer, Integer>> {
  56.         private volatile boolean running = true;
  58.         @Override
  59.         public void run(SourceContext<Tuple3<String, Integer, Integer>> ctx) throws Exception {
  60.             int bound = 50;
  61.             String[] keys = new String[]{"foo@xxyz", "bar", "baz"};
  63.             final long numElements = RandomUtils.nextLong(10, 20);
  64.             int i = 0;
  65.             while (running && i < numElements) {
  66.                 Thread.sleep(RandomUtils.nextLong(1, 5) * 1000L);
  67.                 Tuple3 data = new Tuple3<>(keys[RandomUtils.nextInt(0, 3)], RandomUtils.nextInt(0, bound), RandomUtils.nextInt(0, bound));
  68.                 ctx.collect(data);
  69.                 System.out.println(Thread.currentThread().getId() + "-sand data:" + data);
  70.                 i++;
  71.             }
  72.         }
  74.         @Override
  75.         public void cancel() {
  76.             running = false;
  77.         }
  78.     }
  79. }

输出结果

  1. 58-sand data:(foo,xxyz,44)
  2. 59-sand data:(foo,xxyz,47,41)
  3. Connect> (foo,47,41)
  4. Connect> (xxyz,47,41)
  5. 58-sand data:(foo,xxyz,0)
  6. 59-sand data:(baz,12,12)
  7. Connect> (baz,12,12)
  8. 58-sand data:(foo,xxyz,39)
  9. 59-sand data:(baz,23,27)
  10. Connect> (baz,23,27)

说明

  1. DataSource1模拟流的数据类型为Tuple2，DataSource2模拟流的数据类型为Tuple3
  2. 通过CoFlatMap将数据做一个类型格式统一

Iterate

定义

Transformation Description
DataStream → IterativeStream → DataStream Creates a “feedback” loop in the flow, by redirecting the output of one operator to some previous operator. This is especially useful for defining algorithms that continuously update a model. The following code starts with a stream and applies the iteration body continuously. Elements that are greater than 0 are sent back to the feedback channel, and the rest of the elements are forwarded downstream.

说明

Iterate提供了一种流计算里面的类似递归方法

样例

代码

  1. public class IterateDemo {
  2.     public static void main(String[] args) throws Exception{
  3.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  4.         env.setParallelism(1);
  6.         DataStream<Tuple2<String, Integer>> orangeStream = env.addSource(new DataSource());
  8.         IterativeStream<Tuple2<String, Integer>> iteration = orangeStream.iterate(5000);
  9.         DataStream<Tuple2<String, Integer>> iterationBody = iteration.map(new MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
  10.             @Override
  11.             public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
  12.                 return new Tuple2<>(value.f0, value.f1-5);
  13.             }
  14.         });
  16.         DataStream<Tuple2<String, Integer>> feedback = iterationBody.filter(new FilterFunction<Tuple2<String, Integer>>() {
  17.             @Override
  18.             public boolean filter(Tuple2<String, Integer> value) throws Exception {
  19.                 return value.f1 > 25;
  20.             }
  21.         });
  23.         iteration.closeWith(feedback);
  25.         DataStream<Tuple2<String, Integer>> output = iterationBody.filter(new FilterFunction<Tuple2<String, Integer>>() {
  26.             @Override
  27.             public boolean filter(Tuple2<String, Integer> value) throws Exception {
  28.                 return value.f1 <= 25;
  29.             }
  30.         });
  32.         feedback.print("Iterate feedback");
  33.         output.print("Iterate output");
  34.         env.execute("Iterate Demo");
  35.     }
  37.     private static class DataSource extends RichParallelSourceFunction<Tuple2<String, Integer>> {
  38.         private volatile boolean running = true;
  40.         @Override
  41.         public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
  42.             int bound = 20;
  43.             String[] keys = new String[]{"foo", "bar", "baz"};
  45.             final long numElements = RandomUtils.nextLong(10, 20);
  46.             int i = 0;
  48.             while (running && i < numElements) {
  49.                 Thread.sleep(RandomUtils.nextLong(1, 5) * 1000L);
  50.                 if (i == 0){
  51.                     ctx.collect(new Tuple2<>(keys[RandomUtils.nextInt(0, 3)], 36));
  52.                 }else {
  53.                     Tuple2 data = new Tuple2<>(keys[RandomUtils.nextInt(0, 3)], RandomUtils.nextInt(10, bound));
  54.                     ctx.collect(data);
  55.                     System.out.println(Thread.currentThread().getId() + "-sand data:" + data);
  56.                 }
  57.                 i++;
  58.             }
  59.         }
  61.         @Override
  62.         public void cancel() {
  63.             running = false;
  64.         }
  65.     }
  66. }

输出结果

  1. Iterate feedback> (bar,31)
  2. Iterate feedback> (bar,26)
  3. Iterate output> (bar,21)
  4. 59-sand data:(baz,11)
  5. Iterate output> (baz,6)
  6. 59-sand data:(foo,14)

说明

发送一条值是36的数据，通过map方法之后仍然大于25，进入到递归，继续执行map里面的方法，直到不满足递归条件后输出结果