算子介绍(Operators)

Flink 的转换(Transformmations) 就是用户通过算子将一个 DataStream 转为新的 DataStream 的过程，主要分为三类：

  • 数据流转换(DataStream Transformations)：进行数据流相关转换操作；
  • 物理分区(Physical partitioning)：Flink 提供的底层 API ，用户在数据转换完成后对数据分区进行更细粒度的配置；
  • 算子链和资源组(Task chaining and resource groups)：允许用户进行算子链和资源组的细粒度的控制。

算子分类

Map

Transformation(数据流转换):[DataStream → DataStream]
描述:使用最多的算子，即输入一个元素，输出一个元素
代码示例:
给定1-9的数字，将其乘以2

  2. import org.apache.flink.api.common.functions.MapFunction;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. public class OperatorTest {
  7.     public static void main(String[] args) throws Exception {
  8.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  9.         DataStream<Integer> dataStream = env.fromElements(1,2,3,4,5,6,7,8,9,10);
  10.         DataStream<Integer> mapDataStream = dataStream.map(new MapFunction<Integer, Integer>() {
  11.             @Override
  12.             public Integer map(Integer integer) throws Exception {
  13.                 // 所有元素都乘以2
  14.                 return integer * 2;
  15.             }
  16.         });
  17.         mapDataStream.print();
  18.     env.execute("Opeartor Test");
  19.     }
  20. }
  22. // 输出: 2,4,6,8,10,12,14,16,18

FlatMap

Transformation(数据流转换):DataStream → DataStream
描述:与Map功能一致，但是 FlatMap 是输入一个元素，输出0个或多个元素
代码示例:

  3. import org.apache.flink.api.common.functions.FlatMapFunction;
  4. import org.apache.flink.api.common.typeinfo.Types;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.util.Collector;
  9. import java.util.Arrays;
  12. public class FlatMapTest {
  14.     public static void main(String[] args) throws Exception {
  15.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         String in = "Flink,Spark,Storm,Hive";
  17.         DataStream<String> dataStream = env.fromElements(in);
  18.         DataStream<String> flatMapStream = dataStream.flatMap(
  19.                 new FlatMapFunction<String, String>(){
  20.                     @Override
  21.                     public void flatMap(String in, Collector<String> out) throws Exception {
  22.                         for(String s: in.split(",")){
  23.                             out.collect(s);
  24.                         }
  25.                     }
  26.                 }
  27.         ).returns(Types.STRING);
  28.         // Lambda 写法
  29. //        DataStream<String> flatMapStream = dataStream.flatMap((String s,Collector<String> out) ->
  30. //                Arrays.stream(s.split(",")).forEach(x -> out.collect(x))
  31. //        ).returns(Types.STRING);
  32.         flatMapStream.print();
  33.         env.execute("FlatMap");
  34.     }
  35. }


KeyBy

Transformation(数据流转换):DataStream -> KeyedStream
描述:KeyBy 将相同键划分刀捅一个分区，在逻辑上是基于 key 对流进行分区。在内部，它使用 hash 函数对流进行分区
代码示例:

  2. import org.apache.flink.api.java.functions.KeySelector;
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.datastream.KeyedStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. public class KeyByTest {
  9.     public static void main(String[] args) throws Exception {
  10.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  11.         DataStream<Tuple2<String,Integer>> dataStream = env.fromElements(
  12.                 new Tuple2<>("Flink",10),
  13.                 new Tuple2<>("Flink",20),
  14.                 new Tuple2<>("Spark",30),
  15.                 new Tuple2<>("Spark",15),
  16.                 new Tuple2<>("Storm",20),
  17.                 new Tuple2<>("Storm",5)
  18.         );
  20.         KeyedStream<Tuple2<String, Integer>,String> keyedStream = dataStream.keyBy(
  21.                 new KeySelector<Tuple2<String, Integer>, String>() {
  22.                     @Override
  23.                     public String getKey(Tuple2<String, Integer> in) throws Exception {
  24.                         return in.f0;
  25.                     }
  26.                 }
  27.         );
  28.         keyedStream.print();
  31.         env.execute("KeyBy Test");
  33.     }

reduce

Transformation(数据流转换):KeyedStream -> DataStream
描述:Reduce 返回单个的结果值，并且 reduce 操作每处理一个元素总是创建一个新值。常用的方法有 average, sum, min, max, count，使用 reduce 方法都可实现。
代码示例:

  2. import javassist.bytecode.analysis.Type;
  3. import org.apache.flink.api.common.functions.ReduceFunction;
  4. import org.apache.flink.api.common.typeinfo.Types;
  5. import org.apache.flink.api.java.functions.KeySelector;
  6. import org.apache.flink.api.java.tuple.Tuple2;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.datastream.KeyedStream;
  9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. public class reduceTest {
  13.     public static void main(String[] args) throws Exception {
  14.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         DataStream<Tuple2<String,Integer>> dataStream = env.fromElements(
  16.                 new Tuple2<>("Flink",10),
  17.                 new Tuple2<>("Flink",20),
  18.                 new Tuple2<>("Spark",30),
  19.                 new Tuple2<>("Spark",15),
  20.                 new Tuple2<>("Storm",20),
  21.                 new Tuple2<>("Storm",5)
  22.         );
  24.         KeyedStream<Tuple2<String, Integer>,String> keyedStream = dataStream.keyBy(
  25.                 new KeySelector<Tuple2<String, Integer>, String>() {
  26.                     @Override
  27.                     public String getKey(Tuple2<String, Integer> in) throws Exception {
  28.                         return in.f0;
  29.                     }
  30.                 }
  31.         );
  32.         DataStream<Tuple2<String,Integer>> reduceStream = keyedStream.reduce(
  33.                 new ReduceFunction<Tuple2<String, Integer>>() {
  34.                     @Override
  35.                     public Tuple2<String, Integer> reduce(Tuple2<String, Integer> stringIntegerTuple2, Tuple2<String, Integer> t1) throws Exception {
  36.                         return new Tuple2<>(stringIntegerTuple2.f0,stringIntegerTuple2.f1 + t1.f1);
  37.                     }
  38.                 }
  39.         );
  40.         // lambda写法
  41. //        SingleOutputStreamOperator<Tuple2<String,Integer>> reduceStream = dataStream.keyBy(0)
  42. //                .reduce((value1,value2) -> new Tuple2<>(value1.f0,value1.f1+value2.f1)).returns(Types.TUPLE(Types.STRING, Types.INT));
  43.         reduceStream.print();
  44.         env.execute("reduce Test");
  46.     }
  47. }


Filter

Transformation(数据流转换):DataStream -> DataStream
描述:过滤出符合条件的数据
代码示例:

  2. import org.apache.flink.api.common.functions.FilterFunction;
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. public class filterTest {
  8.     public static void main(String[] args) throws Exception {
  9.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  10.         DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  11.         DataStream<Integer> filterDataStream = dataStream.filter(
  12.                 new FilterFunction<Integer>() {
  13.                     @Override
  14.                     public boolean filter(Integer integer) throws Exception {
  15.                         return integer % 2 == 0;
  16.                     }
  17.                 }
  18.         );
  19.         // Lambda 写法
  20. //        DataStream<Integer> filterDataStream = dataStream.filter(x -> x % 2 == 0 );
  21.         filterDataStream.print();
  22.         env.execute("filter Test");
  24.     }
  25. }


Union

Transformation(数据流转换):DataStream → DataStream
描述:Union 函数将两个或多个数据流结合在一起。 这样就可以并行地组合数据流。 如果我们将一个流与自身组合，那么它会输出每个记录两次
*代码示例:

  2. import org.apache.flink.streaming.api.datastream.DataStream;
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. import scala.Tuple2;
  6. public class UnionTest {
  7.     public static void main(String[] args) throws Exception {
  8.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  10.         DataStream<Tuple2<String,Integer>> dataStream1 = env.fromElements(
  11.                 new Tuple2<>("Flink",10),
  12.                 new Tuple2<>("Flink",20)
  14.         );
  15.         DataStream<Tuple2<String,Integer>> dataStream2 = env.fromElements(
  16.                 new Tuple2<>("Spark",15),
  17.                 new Tuple2<>("Spark",25));
  19.         DataStream<Tuple2<String,Integer>> unionStream = dataStream1.union(dataStream2);
  20.         unionStream.print();
  22.                 env.execute("Union Test");
  23.     }
  24. }

Connection

Transformation(数据流转换):DataStream,DataStream → ConnectedStream
描述:将两个数据流合并到一起与Union 算子的区别：

  • ConnectedStreams 只能连接两个流，而 union 可以连接多于两个流。
  • ConnectedStreams 连接的两个流类型可以不一致，而 union 连接的流的类型必须一致。
  • ConnectedStreams 会对两个流的数据应用不同的处理方法，并且双流之间可以共享状态。这在第一个流的输入会影响第二个流时, 会非常有用。
  • Connect合并后，可用map中的CoMapFunction或flatMap中的CoFlatMapFunction来对合并流中的每个流进行处理

代码示例:

  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.api.java.tuple.Tuple3;
  5. import org.apache.flink.streaming.api.datastream.ConnectedStreams;
  6. import org.apache.flink.streaming.api.datastream.DataStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.co.CoMapFunction;
  10. public class ConnectTest {
  11.     public static void main(String[] args) throws Exception {
  12.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  13.         DataStream<Tuple2<String,Integer>> dataStream1 = env.fromElements(
  14.                 new Tuple2<>("Flink",10),
  15.                 new Tuple2<>("Spark",15)
  17.         );
  19.         DataStream<Tuple2<String,String>> dataStream2 = env.fromElements(
  20.                 new Tuple2<>("Flink","流计算"),
  21.                 new Tuple2<>("Spark","微批计算")
  22.         );
  24.         ConnectedStreams<Tuple2<String, Integer>, Tuple2<String, String>> connectedStreams = dataStream1.connect(dataStream2);
  25.         DataStream<Object> dataStream3 =  connectedStreams.keyBy(0,0)
  26.                 .map(new CoMapFunction<Tuple2<String, Integer>, Tuple2<String, String>, Object>() {
  27.             @Override
  28.             public String map1(Tuple2<String, Integer> value) throws Exception {
  29.                 return value.f0 + "-" + value.f1;
  30.             }
  32.             @Override
  33.             public Tuple3<String,String,String> map2(Tuple2<String, String> value) throws Exception {
  34.                 return new Tuple3<String,String,String>(value.f0,"-", value.f1);
  35.             }
  36.         });
  37.         dataStream3.print();
  38.         env.execute("Connection Test");
  39.     }
  40. }

Aggregations

Transformation(数据流转换):KeyedStream -> DataStream
描述:Aggregations 是官方提供的聚合算子，封装了常用的聚合操作
代码示例:

Tuple

访问 Tuple 的元素时可以用一个或多个位置键

  3. import org.apache.flink.api.java.tuple.Tuple;
  4. import org.apache.flink.api.java.tuple.Tuple6;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.datastream.KeyedStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. public class AggregationsTest {
  11.     public static void main(String[] args) throws Exception {
  12.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  13.         DataStream<Tuple6<String,String,String,Integer,Double,Integer>> dataStream = env.fromElements(
  14.                 new Tuple6<>("技术部","小明","男",25,5000.0,2),
  15.                 new Tuple6<>("技术部","李华","男",26,6000.0,3),
  16.                 new Tuple6<>("技术部","玛丽亚","女",24,4500.0,1),
  17.                 new Tuple6<>("技术部","Jane","男",28,8000.0,5),
  18.                 new Tuple6<>("技术部","康康","男",30,10000.0,10)
  19.         );
  20.         KeyedStream<Tuple6<String,String,String,Integer,Double,Integer>, Tuple> keyedStream = dataStream.keyBy(0);
  21. //        keyedStream.print();
  22.         // 滚动计算指定key的最小值，可以通过index或者fieldName来指定key
  23. //        keyedStream.min(0).print();
  24.         keyedStream.min(0).print();
  25. //        // 滚动计算指定key的最大值
  26.         keyedStream.max(0).print();
  27. //        // 滚动计算指定key的最小值，并返回其对应的元素
  28.         keyedStream.minBy(0).print();
  29. //        // 滚动计算指定key的最大值，并返回其对应的元素
  30.         keyedStream.maxBy(0).print();
  31.         env.execute("Aggregations Test");
  33.     }
  34. }

POJOs

使用 POJOs 时不能使用位置键，需要指定key
定义 PoJo

  1. import lombok.AllArgsConstructor;
  2. import lombok.Data;
  3. import lombok.EqualsAndHashCode;
  5. @Data
  6. @AllArgsConstructor
  7. @EqualsAndHashCode
  8. public class Company {
  9.     public String name;
  10.     public String department;
  11.     public String gender;
  12.     public int age;
  13.     public double salary;
  14.     public int workYear;
  16.     public Company(){};
  19. }
  2. import dao.Company;
  3. import org.apache.flink.api.java.tuple.Tuple;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.datastream.KeyedStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. public class AggregationsPojoTest {
  9.     public static void main(String[] args) throws Exception {
  10.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  11.         DataStream<Company> dataStream = env.fromElements(
  12.                 new Company("技术部","小明","男",25,5000.0,2),
  13.                 new Company("技术部","李华","男",26,6000.0,3),
  14.                 new Company("技术部","玛丽亚","女",24,4500.0,1),
  15.                 new Company("技术部","Jane","男",28,8000.0,5),
  16.                 new Company("技术部","康康","男",30,10000.0,10)
  17.         );
  19.         KeyedStream<Company, Tuple> keyedStream = dataStream.keyBy("gender");
  20. //        keyedStream.print();
  21.         // 滚动计算指定key的最小值，可以通过index或者fieldName来指定key
  22. //        keyedStream.min(0).print();
  23.         keyedStream.min("gender").print();
  24. //        // 滚动计算指定key的最大值
  25.         keyedStream.max("gender").print();
  26. //        // 滚动计算指定key的最小值，并返回其对应的元素
  27.         keyedStream.minBy("gender").print();
  28. //        // 滚动计算指定key的最大值，并返回其对应的元素
  29.         keyedStream.maxBy("gender").print();
  30.         env.execute("Aggregations Test");
  32.     }
  33. }


windows

Transformation(数据流转换):KeyedStream → WindowedStream
描述:Window 函数允许按时间或其他条件对现有 KeyedStream 进行分组。 以下是以 5 秒的时间窗口聚合：
代码示例:

  2. import org.apache.flink.api.common.typeinfo.Types;
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.datastream.WindowedStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  8. import org.apache.flink.streaming.api.windowing.time.Time;
  9. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  12. public class windowTest {
  13.     public static void main(String[] args) throws Exception {
  14.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         DataStream<Tuple2<String,Integer>> dataStream = env.fromElements(
  16.                 new Tuple2<>("Flink",10),
  17.                 new Tuple2<>("Flink",20),
  18.                 new Tuple2<>("Spark",30),
  19.                 new Tuple2<>("Spark",15),
  20.                 new Tuple2<>("Storm",20),
  21.                 new Tuple2<>("Storm",5)
  22.         );
  24.         WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream =  dataStream.keyBy(value->value.f0)
  25.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(5)));
  26. //            .window( TumblingEventTimeWindows.of(Time.seconds(10))).;
  27.         // 转为DataStream
  28.         windowedStream.reduce(
  29.                 (value1,value2) -> new Tuple2<>(value1.f0, value1.f1+value2.f1)).returns(Types.TUPLE(Types.STRING, Types.INT)
  30.         ).print();
  31.         env.execute("Windows Test");
  32.     }
  33. }

windowsAll

Transformation(数据流转换): DataStream -> WindowedStream
描述:flink 也可以可以在常规 DataStreams 上定义 WindowedStream,Windows 根据某些特征对所有流事件进行分组,通常，这是非并行数据转换，因为它在非分区数据流上运行
代码示例:

  2. import org.apache.flink.api.common.typeinfo.Types;
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.streaming.api.datastream.AllWindowedStream;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.datastream.WindowedStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  9. import org.apache.flink.streaming.api.windowing.time.Time;
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  12. public class windowAllTest {
  13.     public static void main(String[] args) throws Exception {
  14.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15. //        env.setStreamTimeCharacteristic();
  16.         DataStream<Tuple2<String,Integer>> dataStream = env.fromElements(
  17.                 new Tuple2<>("Flink",10),
  18.                 new Tuple2<>("Flink",20),
  19.                 new Tuple2<>("Spark",30),
  20.                 new Tuple2<>("Spark",15),
  21.                 new Tuple2<>("Storm",20),
  22.                 new Tuple2<>("Storm",5)
  23.         );
  25.         AllWindowedStream<Tuple2<String, Integer>, TimeWindow> windowedStream =  dataStream
  26.                 .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)));
  27. //            .window( TumblingEventTimeWindows.of(Time.seconds(10))).;
  29.         windowedStream.reduce(
  30.                 (value1,value2) -> new Tuple2<>(value1.f0, value1.f1+value2.f1)).returns(Types.TUPLE(Types.STRING, Types.INT)
  31.         ).print();
  32.         env.execute("WindowsAll Test");
  33.     }
  34. }

Window Join

Transformation(数据流转换):
描述:窗口连接连接两个共享公共密钥并位于同一窗口中的流的元素。可以使用窗口分配器定义这些窗口，并对来自两个流的元素进行评估。然后将来自双方的元素传递给用户定义的，JoinFunction或者FlatJoinFunction用户可以发出满足连接条件的结果。
代码示例:

  2. import org.apache.flink.api.common.functions.JoinFunction;
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.api.java.tuple.Tuple3;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  8. import org.apache.flink.streaming.api.windowing.time.Time;
  10. public class WindowJoinTest {
  11.     public static void main(String[] args) throws Exception {
  12.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  13.         DataStream<Tuple2<String,Integer>> dataStream1 = env.fromElements(
  14.                 new Tuple2<>("Flink",10),
  15.                 new Tuple2<>("Spark",15)
  17.         );
  19.         DataStream<Tuple2<String,String>> dataStream2 = env.fromElements(
  20.                 new Tuple2<>("Flink","流计算"),
  21.                 new Tuple2<>("Spark","微批计算")
  22.         );
  23.          dataStream1.join(dataStream2).
  24.                 where(value -> value.f0)
  25.                 .equalTo(value -> value.f0)
  26.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
  27.                 .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, String>, Object>() {
  28.                     @Override
  29.                     public Object join(Tuple2<String, Integer> value1, Tuple2<String, String> value2) throws Exception {
  30.                         return new Tuple3<String,Integer,String>(value1.f0,value1.f1,value2.f1);
  31.                     }
  32.                 }).print()
  33.         ;
  34.          env.execute("Window Join Test");
  35.     }
  36. }

windows Reduce

Transformation(数据流转换):
描述:
代码示例: