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://scyanting.com/article/ipgehg.html