Apache Flink源码解析之stream-transformation

文章目录
  1. 1. StreamTransformation
    1. 1.1. 内置的StreamTransformation
      1. 1.1.1. SourceTransformation
      2. 1.1.2. SinkTransformation
      3. 1.1.3. OneInputTransformation
      4. 1.1.4. TwoInputTransformation
      5. 1.1.5. SplitTransformation
      6. 1.1.6. SelectTransformation
      7. 1.1.7. UnionTransformation
      8. 1.1.8. PartitionTransformation
      9. 1.1.9. FeedbackTransformation
      10. 1.1.10. CoFeedbackTransformation
  2. 2. 小结

之前我们聊了Flink程序的sourcesink就差transformation了。今天我们就来解读一下Flink的transformation。它们三者的关系如下图:

flink-stream-transformation_relationship

当然这还是从Flink编程API的角度来看的(编程视角)。所谓的transformation,用于转换一个或多个DataStream从而形成一个新的DataStream对象。Flink提供编程接口,允许你组合这些transformation从而形成非常灵活的拓扑结构。

StreamTransformation

StreamTransformation是所有transformation的抽象类,提供了实现transformation的基础功能。每一个DataStream都有一个与之对应的StreamTransformation

一些API操作,比如DataStream#map,将会在底层创建一个StreamTransformation树,而在程序的运行时,该拓扑结构会被翻译为StreamGraph

StreamTransformation无关运行时的执行,它只是逻辑上的概念。

属性如下:

  • name : 转换器的名称,这个主要用于可视化的目的
  • uid : 用户指定的uid,该uid的主要目的是用于在job重启时可以再次分配跟之前相同的uid,应该是用于持久保存状态的目的。
  • bufferTimeout :buffer超时时间
  • parallelism : 并行度
  • id : 跟属性uid无关,它的生成方式是基于一个静态累加器
  • outputType : 输出类型
  • slotSharingGroup : 给当前的transformation设置slot共享组。slot sharing group用于将并行执行的operator“归拢”到相同的TaskManager slot中(slot概念基于资源的划分,因此这里的目的是让不同的subtask共享slot资源)

其中,StreamTransformation构造器需要的参数是:

  • name
  • outputType
  • parallelism

核心的抽象方法:

  • setChainingStrategy : 设置chaining策略
  • getTransitivePredecessors :返回中间过渡阶段的前置StreamTransformation集合,该方法的可能的应用场景是用来决定在迭代中的feedback edge(反馈边)最终是有前置StreamTransformation

内置的StreamTransformation

因为就一层继承关系的树形结构,所以这里类之间的关系图就不再暂时了

绝大部分StreamTransformation都需要依赖上游StreamTransformation作为输入SourceTransformation等少数特例除外;

如果没有特别说明,getTransitivePredecessors的实现逻辑都是,由自身加input(上游StreamTransformation)组成的集合。

根据实现,我们可以将它们分成两类:

I :输入输出相关,需要自行定义name,都需要与之对应的operatorsetChainingStrategy的实现都返回operator#setChainingStrategy

属于该分类的有:

1
2
3
4
SourceTransformation
SinkTransformation
OneInputTransformation
TwoInputTransformation

II :内置函数name内部固定,无法更改,无需operatorsetChainingStrategy的实现都只是抛出UnsupportedOperationException异常

属于该分类的有:

1
除了上面那些,其他所有的transformation

SourceTransformation

它表示一个sorce,它并不真正做转换工作,因为它没有输入,但它是任何拓扑的根StreamTransformation

除了StreamTransformation构造器需要的那三个参数,SourceTransformation还需要StreamSource类型的参数,它是真正执行转换的operator

值得一提的是,其getTransitivePredecessors抽象方法的实现:

1
2
3
public Collection<StreamTransformation<?>> getTransitivePredecessors() {
return Collections.<StreamTransformation<?>>singleton(this);
}

因为其没有前置转换器,所以其返回只存储自身实例的集合对象。

SinkTransformation

它表示一个sink,创建的时候构造器需要operator 它是 StreamSink的实例,是最终做转换的operator

getTransitivePredecessors方法的实现是将自身以及input#getTransitivePredecessors的返回值(之前的StreamTransformation集合)集合

该类有两个特别的属性:

  • stateKeySelector
  • stateKeyType

这两个属性的目的是因为sink的状态也可能是基于key分区的。

OneInputTransformation

接受一种输入的StreamTransformation(换句话说,只接收一个输入流)。跟上面的SinkTransformation构造器类似,需要inputoperator两个参数(只不过这里的operator类型是对应的OneInputStreamOperator)。

TwoInputTransformation

表示接收两种输入的StreamTransformation(接收两种流作为输入)。其他的实现同OneInputTransformation

SplitTransformation

可将其看作分流转换器,该转换用于将一个流拆分成多个流(通过OutputSelector来达到这个目的),当然这个操作只是逻辑上的拆分(它只影响上游的流如何跟下游的流连接)。

构造该转换器,同样也是依赖于其输入转换器(input)以及一个输出选择器(outputSelector),但在实例化其父类(StreamTransformation,没有提供自定义的名称,而是固定的常量值Split

SelectTransformation

该选择转换器用于从上游流中筛选出特定的元素。它在使用时,必须跟随在SplitTransformation之后(SplitTransformation通过指定的名称将元素分配到多个逻辑流中)。

构造SelectTransformation需要前一个转换器作为输入,以及上游用于分流的SplitTransformation所使用的名称。跟SplitTransformation类似,这里也无需提供自定义的转换器名称,而是固定的常量值Select

UnionTransformation

合并转换器,该转换器用于将多个输入StreamTransformation进行合并。因此该转换器接收StreamTransformation的集合。其名称也在内部被固定为Union

PartitionTransformation

该转换器用于改变输入元素的分区,其名称为:Partition。因此,工作时除了提供一个StreamTransformation作为输入,还需要提供一个StreamPartitioner的实例来进行分区。

FeedbackTransformation

该转换器用于表示Flink DAG中的一个反馈点feedback point)。所谓反馈点,可用于连接一个或者多个StreamTransformation,这些StreamTransformation被称为反馈边(feedback edges)。处于反馈点下游的operation将可以从反馈点和反馈边获得元素输入。

反馈转换器的固定名称为Feedback,它的实例化需要两个参数:

  • input : 上游输入StreamTransformation
  • waitTime : feedback operator的等待时间,一旦超过该等待时间,将关闭并不再接收任何反馈元素。

实例化FeedbackTransformation时,会自动创建一个用于存储反馈边的集合feedbackEdges。那么反馈边如何收集呢?FeedbackTransformation通过定义一个实例方法:addFeedbackEdge来进行收集,而这里所谓的“收集”就是将下游StreamTransformation的实例加入feedbackEdges集合中(这里可以理解为将两个点建立连接关系,也就形成了边)。不过,这里加入的StreamTransformation的实例有一个要求:也就是当前FeedbackTransformation的实例跟待加入StreamTransformation实例的并行度一致

某种程度上,你可以将其类比于pub-sub机制

CoFeedbackTransformation

某种程度上跟FeedbackTransformation类似。feedback元素的类型不需要跟上游的StreamTransformation元素的类型一致,因为CoFeedbackTransformation之后只允许跟TwoInputTransformations。上游的StreamTransformation将会连接到TwoInputTransformations第一个输入,而feedback edge将会连接到其第二个输入。因此上游的StreamTransformation其实是跟CoFeedbackTransformation无关的,它跟TwoInputTransformation有关。

上游的StreamTransformationCoFeedbackTransformation无关,从CoFeedbackTransformation构造器需要的参数就可以看出来。通常,其他的StreamTransformation的实现都需要传入上游的StreamTransformation作为其输入。但CoFeedbackTransformation却没有,它只需要上游的并行度:parallelism。另外一个需要的参数是feedbackType

它绝大部分实现跟FeedbackTransformation区别在于getTransitivePredecessors方法的实现。我们之前谈及getTransitivePredecessors主要的应用场景就是用于feedback,而它又不像FeedbackTransformation跟其上游输入有关,所以它只返回了只有当前实例的单元素集合。

小结

本文剖析了Flink中的StreamTransformation实现。当然还没有谈到这些transformation之间是如何串联起来,实现非常灵活的拓扑。这是我们后面会谈论的内容。


微信扫码关注公众号:Apache_Flink

apache_flink_weichat