1.OperatorChain的设计与实现

OperatorChain的大致逻辑

在JobGraph对象的创建过程中，将链化可以连在一起的算子，常见的有StreamMap、StreamFilter等类型的算子。OperatorChain中的所有算子都会被运行在同一个Task实例中。StreamTaskNetworkOutput会将接入的数据元素写入算子链的HeadOperator中，从而开启整个OperatorChain的数据处理。

 
OperatorChain的Output组件：将数据发送到下游

如图所示，在OperatorChain中通过Output组件将上下游算子相连，当上游算子数据处理完毕后，会通过Output组件发送到下游的算子中继续处理。

 
OperatorChain的collect()：收集处理完的数据

如图所示，OperatorChain内部定义了WatermarkGaugeExposingOutput接口，且该接口分别继承了Output和Collector接口。Collector接口提供了collect()方法，用于收集处理完的数据。

 
OperatorChain的Output接口：也能输出Watermark和LatencyMarker等事件

Output接口提供了emitWatermark()、emitLatencyMarker()等方法，用于对Collector接口进行拓展，使得Output接口实现类可以输出Watermark和LatencyMarker等事件。WatermarkGaugeExposingOutput接口则提供了获取WatermarkGauge的方法，用于监控最新的Watermark。

 
OperatorChain内部定义了不同的WatermarkGaugeExposingOutput接口实现类。

1. RecordWriterOutput：用于输出OperatorChain中尾端算子处理完成的数据，借助RecordWriter组件将数据元素写入网络。
2. ChainingOutput/CopyingChainingOutput：适用于上下游算子连接在一起且上游算子属于单输出类型的情况。
3. BroadcastingOutputCollector/CopyingBroadcastingOutputCollector：上游算子是多输出类型但上下游算子之间的Selector为空时，创建广播类型的BroadcastingOutputCollector。
4. DirectedOutput/CopyingDirectedOutput：上游算子是多输出类型且Selector不为空时，创建DirectedOutput或CopyingDirectedOutput连接上下游算子。

例子：收集数据并通过Output发数据数据到下游

例如在WordCount的程序中定义flatMap()方法时，会调用Collector.collect()方法收集数据元素，每个算子在定义的函数或使用Output接口的实现类中，完成了上游算子向下游算子发送数据元素的操作。

 

2.OperatorChain的创建和初始化

接下来我们看OperatorChain的初始化过程，如下代码，OperatorChain的构造器包含如下逻辑。

1. 创建StreamOperator（即算子）实例，这里StreamOperator会封装为StreamOperatorFactory并存储在StreamGraph结构中。
2. 获取算子之间的链接配置。chainedConfigs的配置决定了算子之间Output接口的具体实现。
3. 遍历当前作业所有节点的输出边，并构建RecordWriterOutput组件，最终通过RecordWriterOutput组件将数据元素输出到网络中。
4. 创建OperatorChain内部算子之间的上下游连接，完成OperatorChain内部上下游算子之间的数据传输。
5. 单独创建headOperator。headOperator是OperatorChain的头部节点，创建完成后将headOperator暴露到StreamTask实例，供DataOutput接口实现类调用。
6. 如果OperatorChain构建失败，则关闭实例，防止出现内存泄漏。

public OperatorChain(StreamTask<OUT, OP> containingTask,RecordWriterDelegate<SerializationDelegate<StreamRecord<OUT>>> recordWriterDelegate) {// 获取当前StreamTask的userCodeClassloaderfinal ClassLoader userCodeClassloader = containingTask.getUserCodeClassLoader();// 获取StreamConfigfinal StreamConfig configuration = containingTask.getConfiguration();// 获取StreamOperatorFactoryStreamOperatorFactory<OUT> operatorFactory = configuration.getStreamOperatorFactory(userCodeClassloader);// 读取chainedConfigsMap<Integer, StreamConfig> chainedConfigs = 
configuration.getTransitiveChainedTaskConfigsWithSelf(userCodeClassloader);// 根据StreamEdge创建RecordWriterOutput组件List<StreamEdge> outEdgesInOrder = configuration.getOutEdgesInOrder(userCodeClassloader);Map<StreamEdge, RecordWriterOutput<?>> streamOutputMap = new HashMap<>(outEdgesInOrder.size());this.streamOutputs = new RecordWriterOutput<?>[outEdgesInOrder.size()];boolean success = false;try {for (int i = 0; i < outEdgesInOrder.size(); i++) {StreamEdge outEdge = outEdgesInOrder.get(i);// 为每个输出边创建RecordWriterOutputRecordWriterOutput<?> streamOutput = createStreamOutput(recordWriterDelegate.getRecordWriter(i),outEdge,chainedConfigs.get(outEdge.getSourceId()),containingTask.getEnvironment());this.streamOutputs[i] = streamOutput;streamOutputMap.put(outEdge, streamOutput);}// 创建OperatorChain内部算子之间的连接List<StreamOperator<?>> allOps = new ArrayList<>(chainedConfigs.size());this.chainEntryPoint = createOutputCollector(containingTask,configuration,chainedConfigs,userCodeClassloader,streamOutputMap,allOps,containingTask.getMailboxExecutorFactory());if (operatorFactory != null) {WatermarkGaugeExposingOutput<StreamRecord<OUT>> output = getChainEntryPoint();// 创建headOperatorheadOperator = StreamOperatorFactoryUtil.createOperator(operatorFactory,containingTask,configuration,output);headOperator.getMetricGroup().gauge(MetricNames.IO_CURRENT_OUTPUT_WATERMARK,output.getWatermarkGauge());} else {headOperator = null;}allOps.add(headOperator);this.allOperators = allOps.toArray(new StreamOperator<?>[allOps.size()]);success = true;}finally {// 如果创建不成功，则关闭StreamOutputs中的RecordWriterOutput// 这里防止内存泄漏if (!success) {for (RecordWriterOutput<?> output : this.streamOutputs) {if (output != null) {output.close();}}}}
}

OperatorChain作用小结

当OperatorChain创建完成后，就能正常接收StreamTaskInput中的数据元素了。在OperatorChain内部算子之间进行数据传递和处理，最终通过RecordWriterOutput组件将处理完成的数据元素发送到网络中，供下游的Task实例使用。

对于OperatorChain内部Output接口的实现，这里暂不展开。

 

3.创建RecordWriterOutput

RecordWriterOutput用于将数据输出到网络指定位置。

OperatorChain.createStreamOutput()逻辑如下：

1. 获取输出边的OutputTag标签，判断当前Stream节点输出边是否为旁路输出，即在DataStream API中是否使用了旁路输出的相关方法。
2. 返回RecordWriterOutput。RecordWriterOutput中包含RecordWriter组件，最终会通过RecordWriter将算子链处理完成的数据写入网络。

private RecordWriterOutput<OUT> createStreamOutput(RecordWriter<SerializationDelegate<StreamRecord<OUT>>> recordWriter,StreamEdge edge,StreamConfig upStreamConfig,Environment taskEnvironment) {// 获取OutputTagOutputTag sideOutputTag = edge.getOutputTag(); // 获取数据序列化器TypeSerializerTypeSerializer outSerializer = null;// 如果StreamEdge指定了OutputTagif (edge.getOutputTag() != null) {// 则进行边路输出outSerializer = upStreamConfig.getTypeSerializerSideOut(edge.getOutputTag(), taskEnvironment.getUserClassLoader());} else {// 正常输出outSerializer = upStreamConfig.getTypeSerializerOut(taskEnvironment.getUserClassLoader());}// 返回创建的RecordWriterOutput实例return new RecordWriterOutput<>(recordWriter, outSerializer, sideOutputTag, this);
}

 

StreamRecord将数据输出的逻辑

在RecordWriterOutput.collect()方法中定义了StreamRecord数据的输出逻辑，实际上是调用pushToRecordWriter()方法将数据写入RecordWriter，最终通过RecordWriter组件进行数据元素的网络输出。

public void collect(StreamRecord<OUT> record) {if (this.outputTag != null) {return;}pushToRecordWriter(record);
}

 

pushToRecordWriter发送数据

1. 调用serializationDelegate.setInstance()方法，对接入的数据元素进行序列化操作，将数据元素转换成二进制格式。
2. 调用recordWriter.emit()方法通过RecordWriter组件将serializationDelegate中序列化后的二进制数据输出到下游网络中。

//RecordWriterOutput.pushToRecordWriter()
private <X> void pushToRecordWriter(StreamRecord<X> record) {serializationDelegate.setInstance(record);try {recordWriter.emit(serializationDelegate);}catch (Exception e) {throw new RuntimeException(e.getMessage(), e);}
}