20171103周报
本周工作内容：
- 继续研究下FLink transfer latency，主要是学习相关的生产者消费者通信机制和数据传输方式，了解相关接口来编写测试topology。
- 按项目的要求重新部署了集群环境，期间浪费了不少时间，不熟练
- 了解项目功能模块，尝试编写测试样例
- 整理下掌握的FLink总体架构与知识点
下面对FLink进程间通信与数据传输所形成的执行图进行一个分析。
FLink的执行图，分为四层：StreamGraph -> JobGraph -> ExecutionGraph -> 物理执行图。
- StreamGraph：是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。
- JobGraph：StreamGraph经过优化后生成了 JobGraph，提交给 JobManager 的数据结构。主要的优化为，将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。
- ExecutionGraph：JobManager 根据 JobGraph 生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。
- 物理执行图：JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个TaskManager 上部署 Task 后形成的“图”，并不是一个具体的数据结构。用SocketTextStreamWordCount举例，四层执行图的演变过程如下图所示。
FLink执行图
最后涉及到底层代码实现的物理执行图，有两个核心的概念ResultPartition和InputGate，下面分别进行分析。
ResultPartition
在最后演变出来的物理图中可以看到，FLink的Runtime中，使用结果分区（ResultPartition）来表示任务的子任务实例所生产的数据，在执行图中等价于jobmanager的IntermediateResultPartition。IntermediateResultPartition主要用于JobManager组织作业图的一种逻辑数据结构，ResultPartition是运行时的一种逻辑概念，两者处于不同的层面。任何想消费ResultPartition的任务，最终都是请求ResultPartition的某个ResultSubPartition。而请求要么是同一TaskManager中的本地请求，要么是来自另外一个TaskManager中的消费子任务实例发起的远程请求。
其中，结果分区编号（ResultPartitionID）用来标识ResultPartition。ResultPartitionID关联着IntermediateResultPartitionID（也即调度时的分区编号）和ExecutionAttemptID（部署时的生产者子任务实例编号）。在任务失败时，单靠IntermediateResultPartitionID无法鉴别ResultPartition，必须结合ExecutionAttemptID一起鉴别。
结果分区编号（ResultPartitionID）用来标识ResultPartition。在任务失败时，单靠IntermediateResultPartitionID无法鉴别ResultPartition，必须结合ExecutionAttemptID一起鉴别。
InputGate
一个InputGate包含多个输入通道（InputChannel），输入通道用于请求ResultSubpartitionView，并从中消费数据。
所谓的ResultSubpartitionView是由ResultSubpartition所创建的用于供消费者任务消费数据的视图对象。
对于每个InputChannel，消费的生命周期会经历如下的方法调用过程：
requestSubpartition：请求ResultSubpartition；
getNextBuffer：获得下一个Buffer；
releaseAllResources：释放所有的相关资源；
以getNextBufferOrEvent方法为主线来分析SingleInputGate类。SingleInputGate是消费ResultPartition的实体.
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());
}
}
InputChannel根据ResultPartitionLocation提供了三种实现：
LocalInputChannel：用于请求同实例中生产者任务所生产的ResultSubpartitionView的输入通道；
RemoteInputChannel：用于请求远程生产者任务所生产的ResultSubpartitionView的输入通道；
UnknownInputChannel：一种用于占位目的的输入通道，需要占位通道是因为暂未确定相对于生产者任务位置，但最终要么被替换为RemoteInputChannel，要么被替换为LocalInputChannel。
小结：结果子分区中存有真正的数据，会根据结果分区来被消费端消费，不同位置的消费方式不一样，有LOCAL、REMOTE和UNKOWN三种。管理结果分区的是结果分区管理器ResultPartitionManager，一个NetworkEnvironment下只有一个。
