之前我们聊了Flink程序的source
、sink
就差transformation
了。今天我们就来解读一下Flink的transformation
。它们三者的关系如下图:
当然这还是从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
,都需要与之对应的operator
,setChainingStrategy
的实现都返回operator#setChainingStrategy
属于该分类的有:
SourceTransformation
SinkTransformation
OneInputTransformation
TwoInputTransformation
II :内置函数,name
内部固定,无法更改,无需operator
,setChainingStrategy
的实现都只是抛出UnsupportedOperationException
异常
属于该分类的有:
除了上面那些,其他所有的transformation
SourceTransformation
它表示一个sorce,它并不真正做转换工作,因为它没有输入,但它是任何拓扑的根StreamTransformation
。
除了StreamTransformation
构造器需要的那三个参数,SourceTransformation
还需要StreamSource
类型的参数,它是真正执行转换的operator
。
值得一提的是,其getTransitivePredecessors
抽象方法的实现:
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
构造器类似,需要input
和operator
两个参数(只不过这里的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
有关。
上游的StreamTransformation
跟CoFeedbackTransformation
无关,从CoFeedbackTransformation
构造器需要的参数就可以看出来。通常,其他的StreamTransformation
的实现都需要传入上游的StreamTransformation
作为其输入。但CoFeedbackTransformation
却没有,它只需要上游的并行度:parallelism
。另外一个需要的参数是feedbackType
。
它绝大部分实现跟FeedbackTransformation
区别在于getTransitivePredecessors
方法的实现。我们之前谈及getTransitivePredecessors
主要的应用场景就是用于feedback
,而它又不像FeedbackTransformation
跟其上游输入有关,所以它只返回了只有当前实例的单元素集合。
小结
本文剖析了Flink中的StreamTransformation
实现。当然还没有谈到这些transformation
之间是如何串联起来,实现非常灵活的拓扑。这是我们后面会谈论的内容。
微信扫码关注公众号:Apache_Flink