Flink运行时之结果分区消费端

文章目录
  1. 1. 结果分区消费端
    1. 1.1. 输入网关
    2. 1.2. 输入通道

task-exchange-dataflow-for-consumer

结果分区消费端

在前一篇,我们讲解了生产者分区,生产者分区是生产者任务生产中间结果数据的过程。消费者任务在获得结果分区可用的通知之后,会发起对数据的请求。我们仍然以生产者分区的例子作为假设,其在消费端示意图如下:

consumer-end-at-runtime

可以看到在生产端和消费端存在对等的模型,具体ResultSubpartition中的数据如何被消费,我们将在本篇进行深入剖析。

输入网关

输入网关(InputGate)用于消费中间结果(IntermediateResult)在并行执行时由子任务生产的一个或多个结果分区(ResultPartition)。

可以认为生产端的ResultPartition跟消费端的InputGate是对等的。

Flink当前提供了两个输入网关的实现,分别是:

  • SingleInputGate:常规输入网关;
  • UnionInputGate:联合输入网关,它允许将多个输入网关联合起来;

我们主要分析SingleInputGate,因为它是消费ResultPartition的实体,而UnionInputGate主要充当InputGate容器的角色。

SingleInputGate主要的初始化逻辑被封装在其静态的create方法中,当一个Task被实例化时在其构造器中会调用该create方法初始化它对应的SingleInputGate实例。create方法根据传递进来的InputGateDeploymentDescriptor完成对其包含的所有InputChannel的实例化。因为InputChannel记录是按照跟生产端任务的位置来分类的(我们会在下面进行具体分析),所以其实例化也是按照InputGateDeploymentDescriptor中每个InputChannelDeploymentDescriptor包含的ResultPartitionLocation属性来初始化的。

不同的ResultPartitionLocation,消费端任务其请求结果子分区的方式也不同,这一点我们在讲解生产者分区是有所说明。

作为数据的消费者,InputGate最关键的方法自然是获取生产者所生产的缓冲区,提供该功能的方法为getNextBufferOrEvent,它返回的对象是我们之前谈到的统一的数据交换对象BufferOrEvent。

BufferOrEvent的直接消费对象是通信层API中的记录读取器,它会将Buffer中的数据反序列化为记录供上层任务使用。

我们以getNextBufferOrEvent方法为主线来分析SingleInputGate类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
public BufferOrEvent getNextBufferOrEvent() throws IOException, InterruptedException {
//如果已接收到所有EndOfPartitionEvent事件,则说明每个ResultSubpartition中的数据都被消费完成
if (hasReceivedAllEndOfPartitionEvents) {
return null;
}
//触发所有的输入通道向ResultSubpartition发起请求
requestPartitions();
InputChannel currentChannel = null;
//阻塞并循环等待有可获取数据的通道可用
while (currentChannel == null) {
if (isReleased) {
throw new IllegalStateException("Released");
}
//从阻塞队列中请求(并删除)队首的输入通道,阻塞两秒钟,如果没有获取到则不断请求,直到获取到一个输入通道位置
currentChannel = inputChannelsWithData.poll(2, TimeUnit.SECONDS);
}
//从输入通道中获得下一个Buffer
final Buffer buffer = currentChannel.getNextBuffer();
if (buffer == null) {
throw new IllegalStateException("Bug in input gate/channel logic: input gate got " +
"notified by channel about available data, but none was available.");
}
//如果该Buffer是用户数据,则构建BufferOrEvent对象并返回
if (buffer.isBuffer()) {
return new BufferOrEvent(buffer, currentChannel.getChannelIndex());
}
//否则把它当作事件来处理
else {
final AbstractEvent event = EventSerializer.fromBuffer(buffer, getClass().getClassLoader());
//如果获取到的是标识某ResultSubpartition已经生产完数据的事件
if (event.getClass() == EndOfPartitionEvent.class) {
//对获取该ResultSubpartition的通道进行标记
channelsWithEndOfPartitionEvents.set(currentChannel.getChannelIndex());
//如果所有信道都被标记了,置全部通道获取数据完成
if (channelsWithEndOfPartitionEvents.cardinality() == numberOfInputChannels) {
hasReceivedAllEndOfPartitionEvents = true;
}
//对外发出ResultSubpartition已被消费的通知同时释放资源
currentChannel.notifySubpartitionConsumed();
currentChannel.releaseAllResources();
}
//以事件来构建BufferOrEvent对象
return new BufferOrEvent(event, currentChannel.getChannelIndex());
}
}

以上代码段中,第一个关键调用是requestPartitions方法。它会触发所有InputChannel发起对requestSubpartition方法的调用以请求生产端的ResultSubpartition。

有一种占位目的的UnknowInputChannel不响应该方法,因为它最终会被确定为是LocalInputChannel还是RemoteInputChannel,确定的时机通常是JobManager通知器其可消费,TaskManager调用当前SingleInputGate的updateInputChannel方法,确定UnknowInputChannel会转变的具体的通道类型后再调用requestSubpartition方法。

由于requestPartitions只是起到触发其内部的InputChannel去请求的作用,这个调用可能并不会阻塞等待远程数据被返回。因为不同的InputChannel其请求的机制并不相同,RemoteChannel就是利用Netty异步请求的。所以SingleInputGate采用阻塞等待以及事件回调的方式来等待InputChannel上的数据可用。具体而言,它在while代码块中循环阻塞等待有可获取数据的InputChannel。而可用的InputChannel则由它们自己通过回调SingleInputGate的onAvailableBuffer添加到阻塞队列inputChannelsWithData中来。当有可获取数据的InputChannel之后,即可获取到Buffer。

Flink除了提供了SingleInputGate这种常规的输入网关之外,还提供了UnionInputGate,它更像一个包含SingleInputGate的容器,同时可以这些SingleInputGate拥有的InputChannel联合起来。并且多数InputGate约定的接口方法的实现,都被委托给了每个SingleInputGate。

那么它在实现getNextBufferOrEvent方法的时候,到底从哪个InputGate来获得缓冲区呢。它采用的是事件通知机制,所有加入UnionInputGate的InputGate都会将自己注册到InputGateListener。当某个InputGate上有数据可获取,该InputGate将会被加入一个阻塞队列。接着我们再来看getNextBufferOrEvent方法的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public BufferOrEvent getNextBufferOrEvent() throws IOException, InterruptedException {
if (inputGatesWithRemainingData.isEmpty()) {
return null;
}
//遍历每个InputGate,依次调用其requestPartitions方法
requestPartitions();
//阻塞等待输入网关队列中有可获取数据的输入网关
final InputGate inputGate = inputGateListener.getNextInputGateToReadFrom();
//从输入网关中获得数据
final BufferOrEvent bufferOrEvent = inputGate.getNextBufferOrEvent();
//如果获取到的是事件且该事件为EndOfPartitionEvent且输入网关已完成
if (bufferOrEvent.isEvent()
&& bufferOrEvent.getEvent().getClass() == EndOfPartitionEvent.class
&& inputGate.isFinished()) {
//尝试将该输入网关从仍然可消费数据的输入网关集合中删除
if (!inputGatesWithRemainingData.remove(inputGate)) {
throw new IllegalStateException("Couldn't find input gate in set of remaining " +
"input gates.");
}
}
//获得通道索引偏移
final int channelIndexOffset = inputGateToIndexOffsetMap.get(inputGate);
//计算真实通道索引
bufferOrEvent.setChannelIndex(channelIndexOffset + bufferOrEvent.getChannelIndex());
return bufferOrEvent;
}

基本上这个机制跟SingleInputGate是一致的,只不过在UnionInputGate中它是从InputGate中而非从InputChannel中罢了。

输入通道

一个InputGate包含多个输入通道(InputChannel),输入通道用于请求ResultSubpartitionView,并从中消费数据。

所谓的ResultSubpartitionView是由ResultSubpartition所创建的用于供消费者任务消费数据的视图对象。

对于每个InputChannel,消费的生命周期会经历如下的方法调用过程:

  1. requestSubpartition:请求ResultSubpartition;
  2. getNextBuffer:获得下一个Buffer;
  3. releaseAllResources:释放所有的相关资源;

InputChannel根据ResultPartitionLocation提供了三种实现:

  • LocalInputChannel:用于请求同实例中生产者任务所生产的ResultSubpartitionView的输入通道;
  • RemoteInputChannel:用于请求远程生产者任务所生产的ResultSubpartitionView的输入通道;
  • UnknownInputChannel:一种用于占位目的的输入通道,需要占位通道是因为暂未确定相对于生产者任务位置,但最终要么被替换为RemoteInputChannel,要么被替换为LocalInputChannel。

LocalInputChannel会从相同的JVM实例中消费生产者任务所生产的Buffer。因此,这种模式是直接借助于方法调用和对象共享的机制完成消费,无需跨节点网络通信。具体而言,它是通过ResultPartitionManager来直接创建对应的ResultSubpartitionView的实例,这种通道相对简单。

RemoteInputChannel是我们重点关注的输入通道,因为它涉及到远程请求结果子分区。远程数据交换的通信机制建立在Netty框架的基础之上,因此会有一个主交互对象PartitionRequestClient来衔接通信层跟输入通道。

我们以请求子分区的requestSubpartition为入口来进行分析。首先,通过一个ConnectionManager根据连接编号(对应着目的主机)来创建PartitionRequestClient实例。接着具体的请求工作被委托给PartitionRequestClient的实例:

1
partitionRequestClient.requestSubpartition(partitionId, subpartitionIndex, this, 0);

因为Netty以异步的方式处理请求。因此,上面的代码段中会看到将代表当前RemoteChannel实例的this对象作为参数注入到Netty的特定的ChannelHandler中去,在处理时根据特定的处理逻辑会触发RemoteChannel中相应的回调方法。

在RemoteChannel中定义了多个“onXXX”回调方法来衔接Netty的事件回调。其中,较为关键的自然是接收到数据的onBuffer方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public void onBuffer(Buffer buffer, int sequenceNumber) {
boolean success = false;
try {
synchronized (receivedBuffers) {
if (!isReleased.get()) {
//如果实际的序列号跟所期待的序列号相等
if (expectedSequenceNumber == sequenceNumber) {
//将数据加入接收队列同时将预期序列号计数器加一
receivedBuffers.add(buffer);
expectedSequenceNumber++;
//发出有可用Buffer的通知,该通知随后会被传递给其所归属的SingleInputGate,
//以通知其订阅者,有可用数据
notifyAvailableBuffer();
success = true;
}
else {
//如果实际序列号跟所期待的序列号不一致,则会触发onError回调,并相应以一个特定的异常对象
//该方法调用在成功设置完错误原因后,同样会触发notifyAvailableBuffer方法调用
onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber));
}
}
}
}
finally {
//如果不成功,则该Buffer会被回收
if (!success) {
buffer.recycle();
}
}
}

从代码段可以看出,消费时首先会进行序列号比对,这可以看作是一种“校验”机制。服务端每响应客户端一个Buffer都会将序列号加一并随响应数据一起发回给客户端,而客户端则会在消费时也同时累加本地的序列号计数器。在消费的过程中,两个序列号必须一致才能保证消费的顺利进行,否则InputChannel将会抛出IOException异常。

onBuffer方法的执行处于Netty的I/O线程上,但RemoteInputChannel中getNextBuffer却不会在Netty的I/O线程上被调用,所以必须有一个数据共享的容器,这个容器就是receivedBuffers队列。getNextBuffer就是直接从receivedBuffers队列中出队一条数据然后返回。


微信扫码关注公众号:Apache_Flink

apache_flink_weichat


QQ扫码关注QQ群:Apache Flink学习交流群(123414680)

qrcode_for_apache_flink_qq_group