ApacheFlinkTask执行之数据流如何处理
这篇文章主要介绍Apache Flink Task执行之数据流如何处理，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！
站在用户的角度思考问题，与客户深入沟通，找到黄陵网站设计与黄陵网站推广的解决方案，凭借多年的经验，让设计与互联网技术结合，创造个性化、用户体验好的作品，建站类型包括：成都网站制作、成都做网站、外贸营销网站建设、企业官网、英文网站、手机端网站、网站推广、主机域名、网络空间、企业邮箱。业务覆盖黄陵地区。
获取流数据
用户提交的代码最终被封装成了org.apache.flink.runtime.taskmanager.Task,Task是一个Runnable因此核心代码就在run方法，run方法调用了doRun方法，在doRun中调用了invokable.invoke()，Task的整个处理流程其实就在这里面。org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable是一个抽象类，它的子类是不同类型的Task，这里我们主要关注流处理任务相关的org.apache.flink.streaming.runtime.tasks.StreamTask，StreamTask的invoke方法执行了runMailboxLoop()方法。
runMailboxLoop()方法就是执行org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor的runMailboxLoop方法。MailboxProcessor是一种线程模型，runMailboxLoop就是在while轮询中不断执行任务和默认动作，其中默认动作就是StreamTask的processInput方法，该方法调用了StreamInputProcessor的inputProcessor方法，在这个方法中获取并处理了流数据。StreamInputProcessor的子类StreamOneInputProcessor和StreamTwoInputProcessor分别用来处理有1个和2个入度的Task（StreamMultipleInputProcessor先不管）。StreamOneInputProcessor中有1个StreamTaskInput用来获取数据，1个DataOutput用来收集从StreamTaskInput获取的数据；同理，StreamTwoInputProcessor有2个StreamTaskInput和2个DataOutput。StreamTaskInput的子类StreamTaskNetworkInput用来从网络中获取流数据，通过调用他它的emitNext不仅处理流数据还处理了checkpoint barrier，本篇文章只关注数据流的处理流程。StreamTaskNetworkInput从反序列化器中获取到完整流数据后把数据交给DataOutput。DataOutput也有处理1个入度和2个入度的子类，它们都持有OperatorChain中第一个operator的引用，称为headOperator，DataOutput从StreamTaskInput那里获取到数据后会交给headOperator来处理。到此为止，流数据被获取并传入了OperatorChain。 这里总结一下：StreamTask的processInput方法在MailboxProcessor中被反复调用，在processInput方法中StreamTask使用StreamInputProcessor来获取并处理流数据。StreamInputProcessor中的StreamTaskInput用来获取数据，获取的数据交给DataOutput，DataOutput将数据传入OperatorChain的第一个operator。其中StreamTask，StreamInputProcessor和DataOutput都有处理1个入度和2个入度的子类。
数据流过OperatorChain
OperatorChain的第一个operator获取数据后，数据是怎样在OperatorChain中流动的呢？首先说说OperatorChain，StreamOperatorWrapper是chain的每个节点，每个节点都有指向下一个或上一个节点的引用，因此OperatorChain是一个双向链表。但是数据的流动并不依靠这个链式结构。上文我们提到DataOutput将数据交给了headOperator，OperatorChain的第一个节点都是StreamOperator的子类，我们编写的filer算子，map算子等最终都会被封装成StreamOperator，例如子类StreamFlatMap就是执行flatMap方法，StreamFilter就是执行fliter方法等。这些方法执行的时候用org.apache.flink.streaming.api.operators.Output对处理后的结果进行收集。例如StreamFilter当FilterFunction返回true时收集数据，而StreamFlatMap将Output传入flatMap方法中由用户代码进行收集数据。收集的数据是怎样向OperatorChain的下一个节点传递的呢？原来Output中持有OneInputStreamOperator变量指向了chain中下一个节点的算子，调用Output的collect方法会调用下一个算子的processElement，数据就这样在整个OperatorChain中传递了。
发向下游Task
当数据传到OperatorChain的最后一个算子时数据是怎样发向下个Task的呢？最后一个算子拥有的Output实现类是org.apache.flink.streaming.runtime.io.RecordWriterOutput。RecordWriterOutput的collect方法会调用的org.apache.flink.runtime.io.network.api.writer.RecordWriter#emit方法用来发送数据，该方法会将序列化器中的数据复制到BufferBuilder中。BufferBuilder维护了一个内存片段MemorySegment并且可以创建相应的消费者。RecordWriter有2个实现类ChannelSelectorRecordWriter和BroadcastRecordWriter。Task向下游节点的多个并行度发送数据，每个并行度都对应一个channel。ChannelSelectorRecordWriter为每个chanel都保存一个BufferBuilder并分别添加BufferConsumer：
BufferBuilder bufferBuilder = super.requestNewBufferBuilder(targetChannel);//按channel获取BufferBuilder addBufferConsumer(bufferBuilder.createBufferConsumer(), targetChannel);//按channel添加BufferConsumer bufferBuilders[targetChannel] = bufferBuilder;
BroadcastRecordWriter只有一个BufferBuilder，使用同一个BufferBuilder给所有的channel添加BufferConsumer：
try (BufferConsumer bufferConsumer = builder.createBufferConsumer()) {
for (int channel = 0; channel < numberOfChannels; channel++) {
addBufferConsumer(bufferConsumer.copy(), channel);//所有channel用同一个BufferBuilder达到广播的目的
}
}RecordWriter#requestNewBufferBuilder方法会获取BufferBuilder，如果获取失败会导致Task执行线程阻塞造成反压。
public BufferBuilder requestNewBufferBuilder(int targetChannel) throws IOException, InterruptedException {
BufferBuilder builder = targetPartition.tryGetBufferBuilder(targetChannel);//尝试获取，获取不到返回null
if (builder == null) {
long start = System.currentTimeMillis();
builder = targetPartition.getBufferBuilder(targetChannel);//阻塞获取，导致反压
idleTimeMsPerSecond.markEvent(System.currentTimeMillis() - start);
}
return builder;
}BufferBuilder最终来自LocalBufferPool，LocalBufferPool有几个重要的属性：
//taskmanager的网络缓存池，MemorySegment从这里获取 private final NetworkBufferPool networkBufferPool; //已经获取的MemorySegment被组织成一个队列 private final ArrayDequeavailableMemorySegments = new ArrayDeque (); //当前localBufferPool的大小 private int currentPoolSize; //已经获取的MemorySegment private int numberOfRequestedMemorySegments; //每个channel能同时获取的最大BufferBuilder数 private final int maxBuffersPerChannel; //subpartition就是channel，数组存储了每个channel同时使用的BufferBuilder数 private final int[] subpartitionBuffersCount;
BufferBuilder由requestMemorySegment方法和requestMemorySegmentBlocking方法获取，requestMemorySegmentBlocking方法也是调用requestMemorySegment方法并在没有获取到MemorySegment时通过AvailableFuture的get方法来阻塞直到获取成功为止，AvailableFuture是一个用CompletableFuture表示的状态位，这里用到了CompletableFuture的get方法会阻塞直到complete的特性，没有完成的future表示unavailable，完成了的表示available。requestMemorySegment方法中如果已经获取的MemorySegment（numberOfRequestedMemorySegments）大于了localBufferPool的大小（currentPoolSize）需要将多余的MemorySegment先归还给networkBufferPool。之后获取MemorySegment，如果获取不到就设置AvailableFuture为不可用，否则记录channel使用的MemorySegment数量，如果大于maxBuffersPerChannel，也设置AvailableFuture为不可用。
@Nullable
private MemorySegment requestMemorySegment(int targetChannel) throws IOException {
MemorySegment segment = null;
synchronized (availableMemorySegments) {
returnExcessMemorySegments();//将多余的segment归还给networkBufferPool
if (availableMemorySegments.isEmpty()) {
segment = requestMemorySegmentFromGlobal();//全局获取
}
// segment may have been released by buffer pool owner
if (segment == null) {
segment = availableMemorySegments.poll();//局部获取
}
if (segment == null) {
availabilityHelper.resetUnavailable();//获取不到设置为不可用
}
//记录channel正在使用segment数，如果超了设置为不可用
if (segment != null && targetChannel != UNKNOWN_CHANNEL) {
if (subpartitionBuffersCount[targetChannel]++ == maxBuffersPerChannel) {
unavailableSubpartitionsCount++;
availabilityHelper.resetUnavailable();
}
}
}
return segment;
}反压的采集
上面说的AvailableFuture设置为不可用其实和反压有关，Task的isBackPressured方法返回了该Task是否产生了反压。
public boolean isBackPressured() {
if (invokable == null || consumableNotifyingPartitionWriters.length == 0 || !isRunning()) {
return false;
}
//获取所有的AvailableFuture，如果有没完成了则有反压
final CompletableFuture[] outputFutures = new CompletableFuture[consumableNotifyingPartitionWriters.length];
for (int i = 0; i < outputFutures.length; ++i) {
outputFutures[i] = consumableNotifyingPartitionWriters[i].getAvailableFuture();
}
return !CompletableFuture.allOf(outputFutures).isDone();
}以上是“Apache Flink Task执行之数据流如何处理”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注创新互联行业资讯频道！
当前标题：ApacheFlinkTask执行之数据流如何处理
文章来源：http://shouzuofang.com/article/ipgehg.html