前面我们谈论了Flink stream中的transformation。你可以将transformation看成编写Flink程序并构建流式处理程序的必要组成部分（静态表现形式）；而本篇我们将探讨transformation在Flink运行时对应的动态表现形式——operator。他们之间的映射关系见下图：
具体的探讨可以查看前文：Flink中的一些核心概念
StreamOperator
所有operator的最终基类，operator的分类方式，按照输入流个数不同分为：
- 无输入：StreamSource
- 单个流输入：OneInputStreamOperator
- 两个流输入：TwoInputStreamOperator
跟生命周期有关的核心抽象方法：
- setup ： 实例化
operator - open ：该方法会在任何元素被处理之前执行，它的实现通常包含了
operator的初始化逻辑 - close ：该方法在所有的元素都进入到
operator被处理之后调用 - dispose ：该方法在
operator生命周期的最后阶段执行，主要用于回收资源
StreamOperator及其实现类中还包含了一些状态恢复与保存相关的逻辑，但这些不是本文的主题，所有暂时不做探讨。
先来看一下整个package的类关系图：
我们整个剖析方式大致也按照以上operator的分类方式以及类的层次结构来。
StreamSource
作为一个流处理DAG的起点，source operator相比其他operator无疑是特别的（从类的继承关系图也可以看出来）。
它需要接受SourceFunction的实例。并且我们可以看到，它的chaining strategy是HEAD（它表示operator不能有前置operator，但可以作为其他operator的前置operator，下文会谈到）。
this.chainingStrategy = ChainingStrategy.HEAD;StreamSource的实现略显复杂，因为它涉及到我们前面文章谈SourceFunction时谈到的SourceFunction.SourceContext的实现。在这里提供了三个实现，分别对应我们之前谈到的Flink对事件时间的三个分类：
- NonTimestampContext:针对
ProcessingTime，该SourceContext将时间戳设置为-1，并且不发射watermark - AutomaticWatermarkContext:针对
IngestionTime，提供自动的watermark发射机制的SourceContext - ManualWatermarkContext:针对
EventTime的人工发射watermark的SourceContext
它们之间的对应关系也体现在其run方法的实现中：
switch (timeCharacteristic) {
case EventTime:
ctx = new ManualWatermarkContext<>(this, lockingObject, collector);
break;
case IngestionTime:
ctx = new AutomaticWatermarkContext<>(this, lockingObject, collector,
getRuntimeContext().getExecutionConfig().getAutoWatermarkInterval());
break;
case ProcessingTime:
ctx = new NonTimestampContext<>(this, lockingObject, collector);
break;
default:
throw new Exception(String.valueOf(timeCharacteristic));
}在run方法内部会调用SourceFunction的run方法：
try {
userFunction.run(ctx);
// if we get here, then the user function either exited after being done (finite source)
// or the function was canceled or stopped. For the finite source case, we should emit
// a final watermark that indicates that we reached the end of event-time
if (!isCanceledOrStopped()) {
ctx.emitWatermark(Watermark.MAX_WATERMARK);
}
} finally {
// make sure that the context is closed in any case
ctx.close();
}StreamSource通过一个属性：canceledOrStopped来控制sourceFunction的停止。
整个StreamSource的运行逻辑由run来表述，通过cancel来控制停止逻辑。
NonTimestampContext
NonTimestampContext会忽略时间戳，因此它的实现里稍微特别一点的地方在下面的这两个方法：
public void collectWithTimestamp(T element, long timestamp) {
// ignore the timestamp
collect(element);
}以及
public void emitWatermark(Watermark mark) {
owner.checkAsyncException();
// do nothing else
}第一个方法忽略了时间戳，第二个方法不发送watermark。
ManualWatermarkContext
无需特别说明
AutomaticWatermarkContext
该类是自动发送watermark的实现，在构造器中接收参数watermarkInterval来指定自动发送watermark的时间间隔。具体的实现机制是，新建一个独立的发射线程，以指定的时间间隔发射：
this.scheduleExecutor = Executors.newScheduledThreadPool(1);
this.watermarkTimer = scheduleExecutor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
final long currentTime = System.currentTimeMillis();
if (currentTime > nextWatermarkTime) {
// align the watermarks across all machines. this will ensure that we
// don't have watermarks that creep along at different intervals because
// the machine clocks are out of sync
final long watermarkTime = currentTime - (currentTime % watermarkInterval);
synchronized (lockingObjectParam) {
if (currentTime > nextWatermarkTime) {
outputParam.emitWatermark(new Watermark(watermarkTime));
nextWatermarkTime += watermarkInterval;
}
}
}
}
}, 0, watermarkInterval, TimeUnit.MILLISECONDS);除了这种基于时间的以固定频率发射watermark的机制，在collect方法被调用时，也会检查当前的时间戳，如果达到发送条件也会触发emit watermark。
而因为该类实现的是自动发送，在构造器中实现一个定时发送机制，所以emitWatermark方法也就不需要再实现发送逻辑（因为已不再需要用户程序调用emitWatermark方法了），而该方法在该类中的主要任务是负责停止自动发送。停止自动发送的触发条件是收到最后一个元素的信号（将最后一个元素的时间戳设置为Long.MAX_VALUE），emitWatermark收到该标识后，再将其往下游传递并关闭定时发送线程。
OneInputStreamOperator
单一输入流的operator接口，继承自StreamOperator。提供了两个接口方法：
- processElement：处理到达该
operator的一个元素 - processWatermark：处理一个
Watermark
TwoInputStreamOperator
支持两个流作为输入的operator，同样继承自StreamOperator。扩充了多个接口方法：
- processElement1 ： 处理来自第一个输入的某个元素
- processElement2 ： 处理来自第二个输入的某个元素
- processWatermark1 ： 处理来自第一个输入的一个
Watermark - processWatermark2 ： 处理来自第二个输入的一个
Watermark
辅助实现类
Output
Collector的扩展，增加了发射WaterMark的功能。该接口主要供operator用于发射元素或者WaterMark。
- emitWatermark ： 该发射
WaterMark将广播给下游的所有operator
TimeCharacteristic
Flink在涉及到时间相关的处理时，将时间划分为三类。而时间类型的定义在Flink中就是用该枚举来表示：
- ProcessingTime
- IngestionTime
- EventTime
这三种时间类型之前我们曾多次提及，这里不再啰嗦
TimestampedCollector
Output的包装器实现，它用于给元素设置时间戳
AbstractStreamOperator
该抽象类为实现一个具体的operator提供基本的支持，Flink内置提供的operator全部都直接或间接继承自AbstractStreamOperator。
它内部包含了三大类的属性：
- 配置属性
- 运行时属性
- 键值对状态属性
大都数方法都是辅助方法，值得一提的是setup方法。从这里我们可以看到所有operator标识符的生成方式：
String operatorIdentifier = getClass().getSimpleName() + "_" + config.getVertexID() + "_" + runtimeContext.getIndexOfThisSubtask();可以看到标识是由”_”间隔的三段拼接而成。三段分别是：类名，vertex id，以及当前subtask的索引。
然后基于此标识，创建了用于存储状态的stateBackend：
stateBackend = container.createStateBackend(operatorIdentifier, keySerializer);stateBackend 在 dispose方法中会被关闭。
AbstractStreamOperator并没有对open/close等生命周期方法提供具体的实现，这些方法的具体实现被后延至后面谈到的AbstractUdfStreamOperator中。
AbstractUdfStreamOperator
该类主要针对operator生命周期相关的方法(open/close/dispose)提供了模板实现。而这些实现都统一针对用户定义的Function的实例（简称udf）。
ChainingStrategy
该枚举定义了operator的chain strategy(链接策略)。当一个operator链接到其前置operator时，意味着它们将在同一个线程上执行。StreamOperator的默认值是HEAD，这意味着它将没有前置operator，不过它有可能成为其他operator的前置operator。大部分StreamOperator将该枚举以ALWAYS覆盖，表示它们将链接到一个前置operator。
它的三个枚举值：
- ALWAYS ：上面已经提到过，它允许将当前
operator链接到某前置operator，这是提升性能的良好实践，它能够提升operator的并行度 - NEVER ：该策略不支持
operator被链接到某前置operator也不支持被作为其他operator的前置operator。 - HEAD ：该策略表示
operator没有前置operator，不过可以作为其他operator的chain header
内置的Operator实现
StreamCounter
元素累加器，没有什么特别的
StreamProject
这里需要解释一下，此处的project，并非通常所指的项目的意思，而是投射、投影的意思。你可以将其类比于SQL中的SELECT子句。因此他允许你选择你需要的fields集合。这通过其构造器的一个字段索引数组来指定：
在processElement方法中，它依次遍历所有需要的字段索引，将元素中需要的字段提取出来，放入一个用于输出的outTuple，最后再将其发射出去：
public void processElement(StreamRecord<IN> element) throws Exception {
for (int i = 0; i < this.numFields; i++) {
outTuple.setField(((Tuple) element.getValue()).getField(fields[i]), i);
}
output.collect(element.replace(outTuple));
}StreamFilter
filter operator，处理逻辑很简单，根据自定义的FilterFunction方法，对每个元素进行过滤，如果满足过滤条件，则将该元素emit出去。
StreamMap
map operator，根据传入的MapFunction，对每个元素应用map操作后将其发射出去。
StreamFlatMap
flatmap operator接收FlatMapFunction函数，有一些特别之处：在其open方法中，它初始化了一个TimestampedCollector，作为传递给FlatMapFunction的collector，该collector是给那些特定的userFunction使用的，并且用于给他们操作的元素设置时间戳。
StreamGroupedFold
分组的fold operator，fold函数的执行依赖于一个初始化值initialValue。因此这里涉及到状态保存。并且状态是跟具体的分区关联的。因此，在open方法的实现中，需要获得跟分区关联的ValueState：
ValueStateDescriptor<OUT> stateId = new ValueStateDescriptor<>(STATE_NAME, outTypeSerializer, null);
values = getPartitionedState(stateId);processElement方法的实现，涉及到一系列的操作：从ValueState中获取数据，作为“新”的初始值跟当前元素一起进行fold函数运算，获得结果后更新ValueState，然后将获得的结果emit出去。
StreamGroupedReduce
按分组进行reduce操作的operator.
基于特定的状态名称：
private static final String STATE_NAME = "_op_state";构建状态id
ValueStateDescriptor<IN> stateId = new ValueStateDescriptor<>(STATE_NAME, serializer, null);然后再获取状态值：
values = getPartitionedState(stateId);以上两个动作在open方法中实现
在processElement方法中，分为两种情况：
- 如果之前已存在状态值，那么拿当前值跟之前的状态值做
reduce并获得结果，将结果再次更新到最新状态并emit出去 - 如果之前不存在状态值，那么直接将当前值更新到状态中，并将当前值
emit出去
StreamSink
sink operator，通常是流处理的最后一个operator。它接收SinkFunction的实例。在processElement中依次调用其invoke方法。
小结
本文主要探讨了stream transformation的运行时形式operator的大致实现.
原文发布时间为：2016-05-21
本文作者：vinoYang
本文来自云栖社区合作伙伴CSDN博客，了解相关信息可以关注CSDN博客。
