Flink 算子：数据处理的魔法师

目录

导语

用户通过算子能将一个或多个 DataStream 转换成新的 DataStream，在应用程序中可以将多个数据转换算子合并成一个复杂的数据流拓扑。

这部分内容将描述 Flink DataStream API 中基本的数据转换 API，数据转换后各种数据分区方式，以及算子的链接策略。

数据流转换

1.Map：输入一个元素同时输出一个元素。

java：

DataStream<Integer> dataStream =//...

    dataStream.map(newMapFunction<Integer,Integer>(){@OverridepublicIntegermap(Integer value)throwsException{return2* value;}});

scala：

dataStream.map { x => x *2}

2.FlatMap：输入一个元素同时产生零个、一个或多个元素。

java：

dataStream.flatMap(newFlatMapFunction<String,String>(){@OverridepublicvoidflatMap(String value,Collector<String> out)throwsException{for(String word: value.split(" ")){

            out.collect(word);}}});

scala:

 dataStream.flatMap { str => str.split(" ")}

3.Filter：为每个元素执行一个布尔 function，并保留那些 function 输出值为 true 的元素

java：

true 的元素。</font>
    dataStream.filter(newFilterFunction<Integer>(){@Overridepublicbooleanfilter(Integer value)throwsException{return value !=0;}});

sclala：

    dataStream.filter { _ !=0}

4.KeyBy：在逻辑上将流划分为不相交的分区。具有相同 key 的记录都分配到同一个分区。在内部， keyBy() 是通过哈希分区实现的。有多种指定 key的方式。
java:

    dataStream.keyBy(value -> value.getSomeKey());
    dataStream.keyBy(value -> value.f0);

scala:

    dataStream.keyBy(_.someKey)
    dataStream.keyBy(_._1)

以下情况，一个类不能作为 key：

1.它是一种 POJO 类，但没有重写 hashCode() 方法而是依赖于 Object.hashCode() 实现。
2. 它是任意类的数组。

5.Reduce:在相同 key 的数据流上“滚动”执行 reduce。将当前元素与最后一次 reduce 得到的值组合然后输出新值。
java:

    keyedStream.reduce(newReduceFunction<Integer>(){@OverridepublicIntegerreduce(Integer value1,Integer value2)throwsException{return value1 + value2;}});

scala:

    keyedStream.reduce { _ + _ }

6.Window:可以在已经分区的 KeyedStreams 上定义 Window。Window 根据某些特征（例如，最近 5 秒内到达的数据）对每个 key Stream 中的数据进行分组。
java:

dataStream
  .keyBy(value -> value.f0).window(TumblingEventTimeWindows.of(Time.seconds(5)));

scala:

dataStream
  .keyBy(_._1).window(TumblingEventTimeWindows.of(Time.seconds(5)))

7.WindowAll:可以在普通 DataStream 上定义 Window。 Window 根据某些特征（例如，最近 5 秒内到达的数据）对所有流事件进行分组。

java:

    dataStream
      .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

scala:

dataStream
  .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))

8.Window Apply:将通用 function 应用于整个窗口

如果你使用 windowAll 转换，则需要改用 AllWindowFunction。

java:

windowedStream.apply(newWindowFunction<Tuple2<String,Integer>,Integer,Tuple,Window>(){publicvoid apply (Tuple tuple,Window window,Iterable<Tuple2<String,Integer>> values,Collector<Integer> out)throwsException{int sum =0;for(value t: values){
            sum += t.f1;}
        out.collect (newInteger(sum));}});// 在 non-keyed 窗口流上应用 AllWindowFunction
allWindowedStream.apply (newAllWindowFunction<Tuple2<String,Integer>,Integer,Window>(){publicvoid apply (Window window,Iterable<Tuple2<String,Integer>> values,Collector<Integer> out)throwsException{int sum =0;for(value t: values){
            sum += t.f1;}
        out.collect (newInteger(sum));}});

scala:

    windowedStream.apply { WindowFunction }// 在 non-keyed 窗口流上应用 AllWindowFunction
    allWindowedStream.apply { AllWindowFunction }

9.WindowReduce:对窗口应用 reduce function 并返回 reduce 后的值

java:

windowedStream.reduce (newReduceFunction<Tuple2<String,Integer>>(){publicTuple2<String,Integer>reduce(Tuple2<String,Integer> value1,Tuple2<String,Integer> value2)throwsException{returnnewTuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);}});

scala:

    windowedStream.reduce { _ + _ }

10.Union:将两个或多个数据流联合来创建一个包含所有流中数据的新流。注意：如果一个数据流和自身进行联合，这个流中的每个数据将在合并后的流中出现两次。

java:

dataStream.union(otherStream1, otherStream2,...);

scala:

dataStream.union(otherStream1, otherStream2,...)

11.Window Join：根据指定的 key 和窗口 join 两个数据流。

java：

dataStream.join(otherStream).where(<key selector>).equalTo(<key selector>).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply (newJoinFunction(){...});

scala：

dataStream.join(otherStream).where(<key selector>).equalTo(<key selector>).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply {...}

12.Interval Join：根据 key 相等并且满足指定的时间范围内的条件将分别属于两个 keyed stream 的元素 e1 和 e2 Join 在一起。

java：

// this will join the two streams so that// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2

keyedStream.intervalJoin(otherKeyedStream).between(Time.milliseconds(-2),Time.milliseconds(2))// lower and upper bound.upperBoundExclusive(true)// optional.lowerBoundExclusive(true)// optional.process(newIntervalJoinFunction(){...});

scala:

// this will join the two streams so that// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2

keyedStream.intervalJoin(otherKeyedStream).between(Time.milliseconds(-2), Time.milliseconds(2))// lower and upper bound.upperBoundExclusive(true)// optional.lowerBoundExclusive(true)// optional.process(new IntervalJoinFunction(){...})

13.Window CoGroup:根据指定的 key 和窗口将两个数据流组合在一起。

java:

dataStream.coGroup(otherStream).where(0).equalTo(1).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply (newCoGroupFunction(){...});

scala:

dataStream.coGroup(otherStream).where(0).equalTo(1).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply {}

14.Connect:“连接” 两个数据流并保留各自的类型。connect 允许在两个流的处理逻辑之间共享状态。

java:

DataStream<Integer> someStream =//...DataStream<String> otherStream =//...ConnectedStreams<Integer,String> connectedStreams = someStream.connect(otherStream);

scala:

someStream : DataStream[Int]=...
otherStream : DataStream[String]=...val connectedStreams = someStream.connect(otherStream)

15.CoMap, CoFlatMap:类似于在连接的数据流上进行 map 和 flatMap。

java:

connectedStreams.map(newCoMapFunction<Integer,String,Boolean>(){@OverridepublicBooleanmap1(Integer value){returntrue;}@OverridepublicBooleanmap2(String value){returnfalse;}});
connectedStreams.flatMap(newCoFlatMapFunction<Integer,String,String>(){@OverridepublicvoidflatMap1(Integer value,Collector<String> out){
       out.collect(value.toString());}@OverridepublicvoidflatMap2(String value,Collector<String> out){for(String word: value.split(" ")){
         out.collect(word);}}});

scala:

connectedStreams.map((_ :Int)=>true,(_ :String)=>false)
connectedStreams.flatMap((_ :Int)=>true,(_ :String)=>false)

16.Iterate:通过将一个算子的输出重定向到某个之前的算子来在流中创建“反馈”循环。这对于定义持续更新模型的算法特别有用。

//下面的代码从一个流开始，并不断地应用迭代自身。大于 0 的元素被发送回反馈通道，其余元素被转发到下游。
java:IterativeStream<Long> iteration = initialStream.iterate();DataStream<Long> iterationBody = iteration.map (/*do something*/);DataStream<Long> feedback = iterationBody.filter(newFilterFunction<Long>(){@Overridepublicbooleanfilter(Long value)throwsException{return value >0;}});
iteration.closeWith(feedback);DataStream<Long> output = iterationBody.filter(newFilterFunction<Long>(){@Overridepublicbooleanfilter(Long value)throwsException{return value <=0;}});

scala

//下面的代码从一个流开始，并不断地应用迭代自身。大于 0 的元素被发送回反馈通道，其余元素被转发到下游。
initialStream.iterate {
  iteration =>{
    val iterationBody = iteration.map {/*do something*/}(iterationBody.filter(_ >0), iterationBody.filter(_ <=0))}}

物理分区

1.自定义分区：使用用户定义的 Partitioner 为每个元素选择目标任务。

Java

dataStream.partitionCustom(partitioner,"someKey");
dataStream.partitionCustom(partitioner,0);

Scala

dataStream.partitionCustom(partitioner,"someKey")
dataStream.partitionCustom(partitioner,0)

2.随机分区:将元素随机地均匀划分到分区。

Java

dataStream.shuffle();

Scala

dataStream.shuffle()

2.Rescaling

将元素以 Round-robin轮询的方式分发到下游算子。
如果你想实现数据管道，这将很有用，例如，想将数据源多个并发实例的数据分发到多个下游 map来实现负载分配，但又不想像 rebalance() 那样引起完全重新平衡。该算子将只会到本地数据传输而不是网络数据传输，这取决于其它配置值，例如 TaskManager 的 slot 数量。

上游算子将元素发往哪些下游的算子实例集合同时取决于上游和下游算子的并行度。例如，如果上游算子并行度为 2，下游算子的并发度为 6，那么上游算子的其中一个并行实例将数据分发到下游算子的三个并行实例，另外一个上游算子的并行实例则将数据分发到下游算子的另外三个并行实例中。再如，当下游算子的并行度为2，而上游算子的并行度为 6 的时候，那么上游算子中的三个并行实例将会分发数据至下游算子的其中一个并行实例，而另外三个上游算子的并行实例则将数据分发至另下游算子的另外一个并行实例。

当算子的并行度不是彼此的倍数时，一个或多个下游算子将从上游算子获取到不同数量的输入。

请参阅下图来可视化地感知上述示例中的连接模式：

Java

dataStream.rescale();

Scala

dataStream.rescale()

广播:将元素广播到每个分区 。

Java

dataStream.broadcast();

Scala

dataStream.broadcast()

算子链和资源组

将两个算子链接在一起能使得它们在同一个线程中执行，从而提升性能。Flink 默认会将能链接的算子尽可能地进行链接(例如， 两个 map 转换操作)。此外， Flink 还提供了对链接更细粒度控制的 API 以满足更多需求：

如果想对整个作业禁用算子链，可以调用

StreamExecutionEnvironment.disableOperatorChaining()。

下列方法还提供了更细粒度的控制。需要注意的是，这些方法只能在 DataStream 转换操作后才能被调用，因为它们只对前一次数据转换生效。例如，可以

someStream.map(...).startNewChain()

这样调用，而不能

someStream.startNewChain()

这样。

一个资源组对应着 Flink 中的一个 slot 槽，更多细节请看 slots 。 你可以根据需要手动地将各个算子隔离到不同的 slot 中。

创建新链

基于当前算子创建一个新的算子链。

后面两个 map 将被链接起来，而 filter 和第一个 map 不会链接在一起。

Java

someStream.filter(...).map(...).startNewChain().map(...);

Scala

someStream.filter(...).map(...).startNewChain().map(...)

禁止链接

禁止和 map 算子链接在一起。

Java

 someStream.map(...).disableChaining();

Scala

someStream.map(...).disableChaining()

配置 Slot 共享组

为某个算子设置 slot 共享组。Flink 会将同一个 slot 共享组的算子放在同一个 slot 中，而将不在同一 slot 共享组的算子保留在其它 slot 中。这可用于隔离 slot 。如果所有输入算子都属于同一个 slot 共享组，那么 slot 共享组从将继承输入算子所在的 slot。slot 共享组的默认名称是 “default”，可以调用 slotSharingGroup(“default”) 来显式地将算子放入该组。

Java

someStream.filter(...).slotSharingGroup("name");

Scala

someStream.filter(...).slotSharingGroup("name")

名字和描述

Flink里的算子和作业节点会有一个名字和一个描述。名字和描述。名字和描述都是用来介绍一个算子或者节点是在做什么操作，但是他们会被用在不同地方。

名字会用在用户界面、线程名、日志、指标等场景。节点的名字会根据节点中算子的名字来构建。 名字需要尽可能的简洁，避免对外部系统产生大的压力。

描述主要用在执行计划展示，以及用户界面展示。节点的描述同样是根据节点中算子的描述来构建。 描述可以包括详细的算子行为的信息，以便我们在运行时进行debug分析。

Java

someStream.filter(...).setName("filter").setDescription("x in (1, 2, 3, 4) and y > 1");

Scala

someStream.filter(...).setName("filter").setDescription("x in (1, 2, 3, 4) and y > 1")

节点的描述默认是按照一个多行的树形结构来构建的，用户可以通过把

pipeline.vertex-description-mode设为CASCADING

, 实现将描述改为老版本的单行递归模式。

Flink SQL框架生成的算子默认会有一个由算子的类型以及id构成的名字，以及一个带有详细信息的描述。 用户可以通过将

table.optimizer.simplify-operator-name-enabled

设为

false

，将名字改为和以前的版本一样的详细描述。

当一个作业的拓扑很复杂时，用户可以把

pipeline.vertex-name-include-index-prefix

设为

true

，在节点的名字前增加一个拓扑序的前缀，这样就可以很容易根据指标以及日志的信息快速找到拓扑图中对应节点。

Flink 1.15