StreamTask数据流：StreamTask能力概述、Flink处理网络数据逻辑

文章目录一. StreamTask核心组件与能力二. OneInputStreamTask接入网络数据并处理三. 处理数据1. StreamElement类别2. 业务数据处理逻辑四. 小结先来看数据是如何经过网络写入下游Task节点并通过算子进行处理的这里以OneInputStreamTask为例进行说明。一. StreamTask核心组件与能力如代码OneInputStreamTask.init()方法包含了初始化StreamTask主要核心组件的逻辑。OneInputStreamTaskpublicvoidinit()throwsException{StreamConfigconfigurationgetConfiguration();intnumberOfInputsconfiguration.getNumberOfInputs();if(numberOfInputs0){// 创建CheckpointedInputGateCheckpointedInputGateinputGatecreateCheckpointedInputGate();TaskIOMetricGrouptaskIOMetricGroupgetEnvironment().getMetricGroup().getIOMetricGroup();taskIOMetricGroup.gauge(checkpointAlignmentTime,inputGate::getAlignmentDurationNanos);// 创建DataOutput组件DataOutputINoutputcreateDataOutput();StreamTaskInputINinputcreateTaskInput(inputGate,output);// 创建StreamOneInputProcessorinputProcessornewStreamOneInputProcessor(input,output,getCheckpointLock(),operatorChain);}headOperator.getMetricGroup().gauge(MetricNames.IO_CURRENT_INPUT_WATERMARK,this.inputWatermarkGauge);getEnvironment().getMetricGroup().gauge(MetricNames.IO_CURRENT_INPUT_WATERMARK,this.inputWatermarkGauge::getValue);创建CheckpointedInputGateCheckpointedInputGate是对InputGate进行封装实现对CheckpointBarrier对齐的功能。此组件可以接入上游Task实例写入指定InputChannel中的Buffer数据。创建DataOutput组件在StreamTaskInput中会将接入的数据 通过DataOutput组件输出到算子链的HeaderOperator中。创建StreamTaskInput组件用于接收数据将InputGate和DataOutput作为内部成员完成对数据的接入和输出。创建StreamOneInputProcessor数据处理器此组件会被Task线程模型调度并执行实现周期性地从StreamTaskInput组件中读取数据元素并处理。小结OneInputStreamTask初始化过程中包括创建StreamTaskInput和DataOutput组件。接下来了解StreamTask如何利用StreamTaskInput和DataOutput完成数据元素的接收并发送到算子链中进行处理。二. OneInputStreamTask接入网络数据并处理StreamTask.processInput()方法定义了处理数据的主要流程。数据最终会通过MailboxProcessor调度与执行调用StreamOneInputProcessor.processInput()方法完成数据元素的获取和处理调度StreamOneInputProcessor组件串联并运行StreamTaskInput组件、DataOutput组件和OperatorChain组件最终完成数据元素的处理操作。StreamTask.processInput()protectedvoidprocessInput(MailboxDefaultAction.Controllercontroller)throwsException{InputStatusstatusinputProcessor.processInput();// 上游如果还有数据则继续等待执行if(statusInputStatus.MORE_AVAILABLErecordWriter.isAvailable()){return;}// 上游如果没有数据则发送控制消息到控制器if(statusInputStatus.END_OF_INPUT){controller.allActionsCompleted();return;}CompletableFuture?jointFuturegetInputOutputJointFuture(status);MailboxDefaultAction.SuspensionsuspendedDefaultActioncontroller.suspendDefaultAction();jointFuture.thenRun(suspendedDefaultAction::resume);}接下来详细看StreamOneInputProcessor.processInput()emitNext()通过StreamTaskNetworkInput接收数据元素并返回InputStatus判断数据元素是否全部消费完毕。emitNext()会将DataOutput作为参数传递到方法内部用于将数据元素输出到算子链中。publicInputStatusprocessInput()throwsException{InputStatusstatusinput.emitNext(output);if(statusInputStatus.END_OF_INPUT){synchronized(lock){operatorChain.endHeadOperatorInput(1);}}returnstatus;}StreamTaskNetworkInput.emitNext()处理数据逻辑。//BufferOrEvent代表数据元素可以是Buffer类型也可以是事件类型//比如CheckpointBarrier、TaskEvent等事件。publicInputStatusemitNext(DataOutputToutput)throwsException{while(true){// 从Deserializer中获取数据元素if(currentRecordDeserializer!null){DeserializationResultresultcurrentRecordDeserializer.getNextRecord(deserializationDelegate);// 如果DeserializationResult对应的Buffer数据已经被消费则回收Bufferif(result.isBufferConsumed()){currentRecordDeserializer.getCurrentBuffer().recycleBuffer();currentRecordDeserializernull;}// 如果result是完整的数据元素则调用processElement()方法进行处理if(result.isFullRecord()){processElement(deserializationDelegate.getInstance(),output);returnInputStatus.MORE_AVAILABLE;}}// 从checkpointedInputGate中拉取数据//如果bufferOrEvent为空则判断checkpointedInputGate是否已经关闭如果已经关闭了则直接返回END_OF_INPUT状态否则返回NOTHING_AVAILABLE状态。OptionalBufferOrEventbufferOrEventcheckpointedInputGate.pollNext();// 如果有数据则调用processBufferOrEvent()方法进行处理if(bufferOrEvent.isPresent()){processBufferOrEvent(bufferOrEvent.get());}else{// 如果checkpointedInputGate已关闭则返回END_OF_INPUTif(checkpointedInputGate.isFinished()){checkState(checkpointedInputGate.getAvailableFuture().isDone(),Finished BarrierHandler should be available);if(!checkpointedInputGate.isEmpty()){thrownewIllegalStateException(Trailing data in checkpoint barrier handler.);}returnInputStatus.END_OF_INPUT;}returnInputStatus.NOTHING_AVAILABLE;}}}三. 处理数据1. StreamElement类别StreamElement具体类别有StreamRecord、StreamStatus以及Watermark其中StreamRecord就是需要处理的业务数据Watermark则是上游传递下来的Watermark事件。//StreamTaskNetworkInput.processElement()privatevoidprocessElement(StreamElementrecordOrMark,DataOutputToutput)throwsException{// StreamRecord类型if(recordOrMark.isRecord()){output.emitRecord(recordOrMark.asRecord());// Watermark类型}elseif(recordOrMark.isWatermark()){statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(),lastChannel);// LatencyMarker类型}elseif(recordOrMark.isLatencyMarker()){output.emitLatencyMarker(recordOrMark.asLatencyMarker());// StreamStatus类型}elseif(recordOrMark.isStreamStatus()){statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(),lastChannel);}else{thrownewUnsupportedOperationException(Unknown type of StreamElement);}}2. 业务数据处理逻辑对于业务数据调用output.emitRecord(recordOrMark.asRecord())方法进行数据元素的输出操作然后通过DataOutput输出到算子链中进行处理。如下方法调用operator处理实际就是在创建StreamTaskNetworkOutput时指定的算子链HeaderOperator。OneInputStreamTask.StreamTaskNetworkOutput.emitRecord()publicvoidemitRecord(StreamRecordINrecord)throwsException{synchronized(lock){//累加器计算消费数量numRecordsIn.inc();//通过算子链处理operator.setKeyContextElement1(record);operator.processElement(record);}}四. 小结Flink从InputGate中拉取数据元素并进行反序列化操作转换成StreamElement类型后再调用StreamTaskNetworkOutput.emitRecord()方法将数据元素推送到OperatorChain的HeaderOperator中进行处理。《Flink设计与实现核心原理与源码解析》 – 张利兵