文章目录
StreamTask节点中的中间结果数据元素最终通过RecordWriterOutput实现了网络输出，RecordWriterOutput底层依赖RecordWriter组件完成数据输出操作，接下来我们深入了解RecordWriter的设计和实现。
一.创建RecordWriter实例都做了啥
在StreamTask构造器方法中会直接创建RecordWriter实例，用于输出当前任务产生的Intermediate Result数据。
1. 根据recordWrites数量创建不同的代理类
其中createRecordWriterDelegate根据recordWrites的数量创建对应的RecordWriterDelegate代理类。
protected StreamTask(
Environment environment,
@Nullable TimerService timerService,
Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
StreamTaskActionExecutor actionExecutor,
TaskMailbox mailbox)
throws Exception {
...
this.recordWriter = createRecordWriterDelegate(configuration, environment);
...
}
//1. 如果recordWrites数量等于1，则创建SingleRecordWriter代理类；
//2. 如果recordWrites数量等于0，则创建NonRecordWriter代理类；
//3. 其他情况则创建MultipleRecordWriters代理类。
@VisibleForTesting
public static <OUT>
RecordWriterDelegate<SerializationDelegate<StreamRecord<OUT>>>
createRecordWriterDelegate(
StreamConfig configuration, Environment environment) {
List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWrites =
createRecordWriters(configuration, environment);
if (recordWrites.size() == 1) {
return new SingleRecordWriter<>(recordWrites.get(0));
} else if (recordWrites.size() == 0) {
return new NonRecordWriter<>();
} else {
return new MultipleRecordWriters<>(recordWrites);
}
}
2. 创建RecordWriters
获取StreamTask的所有输出边放到RecordWriters中，返回创建的RecordWriter集合。
private static <OUT> List<RecordWriter<SerializationDelegate<StreamRecord<O
UT>>>> createRecordWriters(
StreamConfig configuration,
Environment environment) {
// 创建RecordWriter集合
List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWriters =
new ArrayList<>();
// 获取输出的StreamEdge(所有的输出边)
List<StreamEdge> outEdgesInOrder =
configuration.getOutEdgesInOrder(environment.getUserClassLoader());
// 获取chainedConfigs参数
Map<Integer, StreamConfig> chainedConfigs = configuration.
getTransitiveChainedTaskConfigsWithSelf(environment.getUserClassLoader());
// 遍历输出节点，分别创建RecordWriter实例
for (int i = 0; i < outEdgesInOrder.size(); i++) {
StreamEdge edge = outEdgesInOrder.get(i);
recordWriters.add(
createRecordWriter(
edge,
i,
environment,
environment.getTaskInfo().getTaskName(),
chainedConfigs.get(edge.getSourceId()).getBufferTimeout()));
}
return recordWriters;
}
3. 单个RecordWriter的创建细节
看单个RecordWriter的创建过程，包括分区策略、缓存结果的ResultPartition、最后通过RecordWriterBuilder创建RecordWriter、以及设定MetricGroup监控RecordWriter指标并输出。
private static <OUT> RecordWriter<SerializationDelegate<StreamRecord<OUT>>>
createRecordWriter(
StreamEdge edge,
int outputIndex,
Environment environment,
String taskName,
long bufferTimeout) {
@SuppressWarnings("unchecked")
// 获取边上的StreamPartitioner
StreamPartitioner<OUT> outputPartitioner =
(StreamPartitioner<OUT>) edge.getPartitioner();
LOG.debug("Using partitioner {} for output {} of task {}",
outputPartitioner, outputIndex, taskName);
// 获取ResultPartitionWriter
ResultPartitionWriter bufferWriter = environment.getWriter(outputIndex);
// 初始化Partitioner
if (outputPartitioner instanceof ConfigurableStreamPartitioner) {
int numKeyGroups = bufferWriter.getNumTargetKeyGroups();
if (0 < numKeyGroups) {
((ConfigurableStreamPartitioner) outputPartitioner).configure(numKeyGroups);
}
}
// 创建RecordWriter
RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output =
new RecordWriterBuilder<SerializationDelegate<StreamRecord<OUT>>>()
.setChannelSelector(outputPartitioner)
.setTimeout(bufferTimeout)
.setTaskName(taskName)
.build(bufferWriter);
// 设定MetricGroup监控
output.setMetricGroup(environment.getMetricGroup().getIOMetricGroup());
return output;
}
二. RecordWriter包含的主要组件
RecordWriter内部主要包含RecordSerializer和ResultPartitionWriter两个组件。
1. RecordWriter两种实现类分别实现分发策略和广播
RecordWriter主要有两种实现类：ChannelSelectorRecordWriter和BroadcastRecordWriter。
如代码：通过RecordWriterBuilder创建RecordWriter，此时会根据selector.isBroadcast()条件选择创建ChannelSelectorRecordWriter还是BroadcastRecordWriter实例。
public RecordWriter<T> build(ResultPartitionWriter writer) {
if (selector.isBroadcast()) {
return new BroadcastRecordWriter<>(writer, timeout, taskName);
} else {
return new ChannelSelectorRecordWriter<>(writer, selector, timeout, taskName);
}
}
2. ChannelSelectorRecordWriter的发送策略
ChannelSelectorRecordWriter控制数据元素发送到下游的哪些InputChannel中。如代码，调用channelSelector.selectChannel(record)选择下游的InputChannel。
对于非广播类型的分区器，最终都会创建ChannelSelectorRecordWriter实现StreamRecord数据的下发操作。
public void emit(T record) throws IOException, InterruptedException {
emit(record, channelSelector.selectChannel(record));
}
2.1. ChannelSelector根据实现类实现不同发送策略
ChannelSelector的实现类主要有StreamPartitioner、RoundRobinChannelSelector和OutputEmitter三种。
2. BroadcastRecordWriter广播所有元素到下游InputChannel
BroadcastRecordWriter的实现就比较简单了，在BroadcastRecordWriter中不需要ChannelSelector组件选择数据元素分区，直接将所有的数据元素广播发送到下游所有InputChannel中即可。