开发者社区> thinktothings> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

Flink1.7.2 Source、Window数据交互源码分析

简介: Source如何按partition发射数据到对应的Window Window 如何处理对应的Source发过来的对应的partition数据 理解Flink 数据从Source到Window,上下游数据交换的过程
+关注继续查看

Flink1.7.2 Source、Window数据交互源码分析

源码

概述

  • Source如何按partition发射数据到对应的Window
  • Window 如何处理对应的Source发过来的对应的partition数据
  • 理解Flink 数据从Source到Window,上下游数据交换的过程

StreamGraph 图

输入数据

1 2 3 4 5 6 7 8 9 10

WordCount程序

package com.opensourceteams.module.bigdata.flink.example.stream.worldcount.nc.parallelism

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * nc -lk 1234  输入数据
  */
object SocketWindowWordCountLocal {



  def main(args: Array[String]): Unit = {


    val port = 1234
    // get the execution environment
   // val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val configuration : Configuration = getConfiguration(true)

    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment(1,configuration)





    // get input data by connecting to the socket
    val dataStream = env.socketTextStream("localhost", port, '\n')



    import org.apache.flink.streaming.api.scala._
    val textResult = dataStream.flatMap( w => w.split("\\s") ).map( w => WordWithCount(w,1))
      .keyBy("word")
      /**
        * 每20秒刷新一次,相当于重新开始计数,
        * 好处,不需要一直拿所有的数据统计
        * 只需要在指定时间间隔内的增量数据,减少了数据规模
        */
      .timeWindow(Time.seconds(5))
      //.countWindow(3)
      //.countWindow(3,1)
      //.countWindowAll(3)


      .sum("count" )

    textResult
      .setParallelism(3)
      .print()




    if(args == null || args.size ==0){


      println("==================================以下为执行计划==================================")
      println("执行地址(firefox效果更好):https://flink.apache.org/visualizer")
      //执行计划
      println(env.getExecutionPlan)
      println("==================================以上为执行计划 JSON串==================================\n")
      //StreamGraph
     //println(env.getStreamGraph.getStreamingPlanAsJSON)



      //JsonPlanGenerator.generatePlan(jobGraph)

      env.execute("默认作业")

    }else{
      env.execute(args(0))
    }

    println("结束")

  }


  // Data type for words with count
  case class WordWithCount(word: String, count: Long){
    //override def toString: String = Thread.currentThread().getName + word + " : " + count
  }


  def getConfiguration(isDebug:Boolean = false):Configuration = {

    val configuration : Configuration = new Configuration()

    if(isDebug){
      val timeout = "100000 s"
      val timeoutHeartbeatPause = "1000000 s"
      configuration.setString("akka.ask.timeout",timeout)
      configuration.setString("akka.lookup.timeout",timeout)
      configuration.setString("akka.tcp.timeout",timeout)
      configuration.setString("akka.transport.heartbeat.interval",timeout)
      configuration.setString("akka.transport.heartbeat.pause",timeoutHeartbeatPause)
      configuration.setString("akka.watch.heartbeat.pause",timeout)
      configuration.setInteger("heartbeat.interval",10000000)
      configuration.setInteger("heartbeat.timeout",50000000)
    }


    configuration
  }


}

源码分析(Source)

RecordWriter.emit

  • record 数据为: WordWithCount(1,1)
  • 也就是Source.通过Socket.connect 得到流数据后,按行拆分,进行flatMap,map函数之后,得到的数据
  • numChannels为Source partitions,这个并行度是在DataStream.setParallelism(3)设置的
  • channelSelector.selectChannels(record, numChannels),这个是按key,进行hash,算得当前元素所分的partition,即为change
  • 续续调用 RecordWriter.emit()

    public void emit(T record) throws IOException, InterruptedException {
        emit(record, channelSelector.selectChannels(record, numChannels));
    }

RecordWriter.emit

  • 经调试,按key,hash % 并行度,分配的数据如下

    key:2     partition:0
    key:3     partition:0
    key:4     partition:0
    key:6     partition:0
    
    key:1     partition:1
    key:5     partition:1
    key:8     partition:1
    key:10    partition:1
    
    key:7     partition:2
    key:9     partition:2
  • record进行序列化,数据长度写进ByteBuffer lengthBuffer,数据写进ByteBuffer dataBuffer;

    serializer.serializeRecord(record);
  • 调用RecordWriter.copyFromSerializerToTargetChannel(channel)往通道中写数据
    private void emit(T record, int[] targetChannels) throws IOException, InterruptedException {
        serializer.serializeRecord(record);

        boolean pruneAfterCopying = false;
        for (int channel : targetChannels) {
            if (copyFromSerializerToTargetChannel(channel)) {
                pruneAfterCopying = true;
            }
        }

        // Make sure we don't hold onto the large intermediate serialization buffer for too long
        if (pruneAfterCopying) {
            serializer.prune();
        }
    }

RecordWriter.copyFromSerializerToTargetChannel

  • getBufferBuilder(targetChannel)通过channel,得到BufferBuilder,就是得到当前的partition写入数据对象BufferBuilder,其实就是操作ResultPartition.subPartitions

    BufferBuilder bufferBuilder = getBufferBuilder(targetChannel);
  • 每个partition写入的数据,对应到各自的window,这样就实现了,在source端将数据分区,对应的window处理source对应的分区数据

    subpartitions = {ResultSubpartition[3]@5749} 
    0 = {PipelinedSubpartition@5771} "PipelinedSubpartition#0 [number of buffers: 0 (0 bytes), number of buffers in backlog: 0, finished? false, read view? true]"
    1 = {PipelinedSubpartition@5772} "PipelinedSubpartition#1 [number of buffers: 0 (0 bytes), number of buffers in backlog: 0, finished? false, read view? true]"
    2 = {PipelinedSubpartition@5773} "PipelinedSubpartition#2 [number of buffers: 0 (0 bytes), number of buffers in backlog: 0, finished? false, read view? true]"
  • BufferConsumer是消费BufferBuilder中的数据,BufferBuilder.append 是产生数据
/**
     * @param targetChannel
     * @return <tt>true</tt> if the intermediate serialization buffer should be pruned
     */
    private boolean copyFromSerializerToTargetChannel(int targetChannel) throws IOException, InterruptedException {
        // We should reset the initial position of the intermediate serialization buffer before
        // copying, so the serialization results can be copied to multiple target buffers.
        serializer.reset();

        boolean pruneTriggered = false;
        BufferBuilder bufferBuilder = getBufferBuilder(targetChannel);
        SerializationResult result = serializer.copyToBufferBuilder(bufferBuilder);
        while (result.isFullBuffer()) {
            numBytesOut.inc(bufferBuilder.finish());
            numBuffersOut.inc();

            // If this was a full record, we are done. Not breaking out of the loop at this point
            // will lead to another buffer request before breaking out (that would not be a
            // problem per se, but it can lead to stalls in the pipeline).
            if (result.isFullRecord()) {
                pruneTriggered = true;
                bufferBuilders[targetChannel] = Optional.empty();
                break;
            }

            bufferBuilder = requestNewBufferBuilder(targetChannel);
            result = serializer.copyToBufferBuilder(bufferBuilder);
        }
        checkState(!serializer.hasSerializedData(), "All data should be written at once");

        if (flushAlways) {
            targetPartition.flush(targetChannel);
        }
        return pruneTriggered;
    }

rdWrecoiter.flushAll

  • 把bufferBuilders存储着BufferBuilder 数组,数组个数,对应着并行度的个数,每个BufferBuilder对应着ResultSubpartition.subpartitions的partition的PipelinedSubpartition,会对这些PipelinedSubpartition进行Flush,并给对应的Window进行处理
  • 调用PipelinedSubpartition.flushAll()
    public void flushAll() {
        targetPartition.flushAll();
    }

PipelinedSubpartition.flushAll()

  • 调用PipelinedSubpartition.notifyDataAvailable
  • 调用PipelinedSubpartitionView.notifyDataAvailable
public void flush() {
        final boolean notifyDataAvailable;
        synchronized (buffers) {
            if (buffers.isEmpty()) {
                return;
            }
            // if there is more then 1 buffer, we already notified the reader
            // (at the latest when adding the second buffer)
            notifyDataAvailable = !flushRequested && buffers.size() == 1;
            flushRequested = true;
        }
        if (notifyDataAvailable) {
            notifyDataAvailable();
        }
    }

PipelinedSubpartitionView.notifyDataAvailable

  • 调用LocalInputChannel.notifyDataAvailable
    public void notifyDataAvailable() {
        availabilityListener.notifyDataAvailable();
    }

LocalInputChannel.notifyDataAvailable

  • 调用InputChannel.notifyChannelNonEmpty()
  • 调用SingleInputGate.notifyChannelNonEmpty
    public void notifyDataAvailable() {
        notifyChannelNonEmpty();
    }

SingleInputGate.queueChannel

  • 通知Window,有数据产生了,可以开始消费了(处理数据)

      inputChannelsWithData.notifyAll();
private void queueChannel(InputChannel channel) {
        int availableChannels;

        synchronized (inputChannelsWithData) {
            if (enqueuedInputChannelsWithData.get(channel.getChannelIndex())) {
                return;
            }
            availableChannels = inputChannelsWithData.size();

            inputChannelsWithData.add(channel);
            enqueuedInputChannelsWithData.set(channel.getChannelIndex());

            if (availableChannels == 0) {
                inputChannelsWithData.notifyAll();
            }
        }

        if (availableChannels == 0) {
            InputGateListener listener = inputGateListener;
            if (listener != null) {
                listener.notifyInputGateNonEmpty(this);
            }
        }
    }

源码分析(Window)

OneInputStreamTask.run()

  • 调用StreamInputProcessor.processInput()进行处理,这个函数,会进行阻塞,如果没有读到数据的话
  • 注意,这是Window的处理线程,有多少个并行度,就会开多少个Window,就有多少个这个方法在一直处理Source不断发过来的数据
    protected void run() throws Exception {
        // cache processor reference on the stack, to make the code more JIT friendly
        final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;

        while (running && inputProcessor.processInput()) {
            // all the work happens in the "processInput" method
        }
    }

StreamInputProcessor.processInput()

  • 这是一个阻塞的方法,读取Source中对应的partition中的数据,调用BarrierTracker.getNextNonBlocked()

    final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();
public boolean processInput() throws Exception {
        if (isFinished) {
            return false;
        }
        if (numRecordsIn == null) {
            try {
                numRecordsIn = ((OperatorMetricGroup) streamOperator.getMetricGroup()).getIOMetricGroup().getNumRecordsInCounter();
            } catch (Exception e) {
                LOG.warn("An exception occurred during the metrics setup.", e);
                numRecordsIn = new SimpleCounter();
            }
        }

        while (true) {
            if (currentRecordDeserializer != null) {
                DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);

                if (result.isBufferConsumed()) {
                    currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
                    currentRecordDeserializer = null;
                }

                if (result.isFullRecord()) {
                    StreamElement recordOrMark = deserializationDelegate.getInstance();

                    if (recordOrMark.isWatermark()) {
                        // handle watermark
                        statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
                        continue;
                    } else if (recordOrMark.isStreamStatus()) {
                        // handle stream status
                        statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);
                        continue;
                    } else if (recordOrMark.isLatencyMarker()) {
                        // handle latency marker
                        synchronized (lock) {
                            streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());
                        }
                        continue;
                    } else {
                        // now we can do the actual processing
                        StreamRecord<IN> record = recordOrMark.asRecord();
                        synchronized (lock) {
                            numRecordsIn.inc();
                            streamOperator.setKeyContextElement1(record);
                            streamOperator.processElement(record);
                        }
                        return true;
                    }
                }
            }

            final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();
            if (bufferOrEvent != null) {
                if (bufferOrEvent.isBuffer()) {
                    currentChannel = bufferOrEvent.getChannelIndex();
                    currentRecordDeserializer = recordDeserializers[currentChannel];
                    currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
                }
                else {
                    // Event received
                    final AbstractEvent event = bufferOrEvent.getEvent();
                    if (event.getClass() != EndOfPartitionEvent.class) {
                        throw new IOException("Unexpected event: " + event);
                    }
                }
            }
            else {
                isFinished = true;
                if (!barrierHandler.isEmpty()) {
                    throw new IllegalStateException("Trailing data in checkpoint barrier handler.");
                }
                return false;
            }
        }
    }

BarrierTracker.getNextNonBlocked()

  • 调用inputGate.getNextBufferOrEvent();调到的是SingleInputGate.getNextBufferOrEvent()去读取Source发过来到当前Window中的数据
    public BufferOrEvent getNextNonBlocked() throws Exception {
        while (true) {
            Optional<BufferOrEvent> next = inputGate.getNextBufferOrEvent();
            if (!next.isPresent()) {
                // buffer or input exhausted
                return null;
            }

            BufferOrEvent bufferOrEvent = next.get();
            if (bufferOrEvent.isBuffer()) {
                return bufferOrEvent;
            }
            else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
                processBarrier((CheckpointBarrier) bufferOrEvent.getEvent(), bufferOrEvent.getChannelIndex());
            }
            else if (bufferOrEvent.getEvent().getClass() == CancelCheckpointMarker.class) {
                processCheckpointAbortBarrier((CancelCheckpointMarker) bufferOrEvent.getEvent(), bufferOrEvent.getChannelIndex());
            }
            else {
                // some other event
                return bufferOrEvent;
            }
        }
    }

SingleInputGate.getNextBufferOrEvent

  • 这个才真正读Source发过来对应的partition,对应当前的Window中的数据,currentChannel.getNextBuffer();
  • inputChannelsWithData 这个对象进行线程阻塞,通知,就是一个开关,Source发过来的数据处理完了就关了,Source有新的数据发过来,就打开了,开关打开时,就可以进行Window数据处理了
private Optional<BufferOrEvent> getNextBufferOrEvent(boolean blocking) throws IOException, InterruptedException {
        if (hasReceivedAllEndOfPartitionEvents) {
            return Optional.empty();
        }

        if (isReleased) {
            throw new IllegalStateException("Released");
        }

        requestPartitions();

        InputChannel currentChannel;
        boolean moreAvailable;
        Optional<BufferAndAvailability> result = Optional.empty();

        do {
            synchronized (inputChannelsWithData) {
                while (inputChannelsWithData.size() == 0) {
                    if (isReleased) {
                        throw new IllegalStateException("Released");
                    }

                    if (blocking) {
                        inputChannelsWithData.wait();
                    }
                    else {
                        return Optional.empty();
                    }
                }

                currentChannel = inputChannelsWithData.remove();
                enqueuedInputChannelsWithData.clear(currentChannel.getChannelIndex());
                moreAvailable = !inputChannelsWithData.isEmpty();
            }

            result = currentChannel.getNextBuffer();
        } while (!result.isPresent());

        // this channel was now removed from the non-empty channels queue
        // we re-add it in case it has more data, because in that case no "non-empty" notification
        // will come for that channel
        if (result.get().moreAvailable()) {
            queueChannel(currentChannel);
            moreAvailable = true;
        }

        final Buffer buffer = result.get().buffer();
        if (buffer.isBuffer()) {
            return Optional.of(new BufferOrEvent(buffer, currentChannel.getChannelIndex(), moreAvailable));
        }
        else {
            final AbstractEvent event = EventSerializer.fromBuffer(buffer, getClass().getClassLoader());

            if (event.getClass() == EndOfPartitionEvent.class) {
                channelsWithEndOfPartitionEvents.set(currentChannel.getChannelIndex());

                if (channelsWithEndOfPartitionEvents.cardinality() == numberOfInputChannels) {
                    // Because of race condition between:
                    // 1. releasing inputChannelsWithData lock in this method and reaching this place
                    // 2. empty data notification that re-enqueues a channel
                    // we can end up with moreAvailable flag set to true, while we expect no more data.
                    checkState(!moreAvailable || !pollNextBufferOrEvent().isPresent());
                    moreAvailable = false;
                    hasReceivedAllEndOfPartitionEvents = true;
                }

                currentChannel.notifySubpartitionConsumed();

                currentChannel.releaseAllResources();
            }

            return Optional.of(new BufferOrEvent(event, currentChannel.getChannelIndex(), moreAvailable));
        }
    }

输出数据

  • 注意,打印输出的数据,是有规则的,按partition打印的,而且,parition打印的顺序,是在window时按key去重,此时没有排序,然后发送给sink进行打印输出
--------------------------
partition:0
WordWithCount(2,1)
WordWithCount(6,1)
WordWithCount(4,1)
WordWithCount(3,1)

partition:1
WordWithCount(1,1)
WordWithCount(10,1)
WordWithCount(8,1)
WordWithCount(5,1)

partition:2
WordWithCount(7,1)
WordWithCount(9,1)


``

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
Apache Flink Time & Window 深度解析
作者:邱从贤 1、 Window & Time 介绍 Apache Flink(以下简称 Flink) 是一个天然支持无限流数据处理的分布式计算框架,在 Flink 中 Window 可以将无限流切分成有限流,是处理有限流的核心组件,现在 Flink 中 Window 可以是时间驱动的(Time Window),也可以是数据驱动的(Count Window)。
1597 0
Flink 流批一体在小米的实践
基于流批一体的改造,无论是实时还是离线都只需要维护一套计算框架,为业务开发人员、平台提供方和计算引擎的支持方节省了大量人力资源。
1468 0
Flink 1.11 与 Hive 批流一体数仓实践
Flink 从 1.9.0 开始提供与 Hive 集成的功能,随着几个版本的迭代,在最新的 Flink 1.11 中,与 Hive 集成的功能进一步深化,并且开始尝试将流计算场景与Hive 进行整合。
22093 0
网易云音乐基于 Flink + Kafka 的实时数仓建设实践
本文由网易云音乐实时计算平台研发工程师岳猛分享,主要从以下四个部分将为大家介绍 Flink + Kafka 在网易云音乐的应用实战: 背景、Flink + Kafka 平台化设计、Kafka 在实时数仓中的应用、问题 & 改进。
20614 0
网易:Flink + Iceberg 数据湖探索与实践
今天主要和大家交流的是网易在数据湖 Iceberg 的一些思考与实践。从网易在数据仓库建设中遇到的痛点出发,介绍对数据湖 Iceberg 的探索以及实践之路。
17067 0
从 Storm 迁移到 Flink,美团外卖实时数仓建设实践
本文主要介绍一种通用的实时数仓构建的方法与实践。实时数仓以端到端低延迟、SQL 标准化、快速响应变化、数据统一为目标。
14502 0
Flink SQL 1.11 on Zeppelin 平台化实践
鉴于有很多企业都无法配备专门的团队来解决 Flink SQL 平台化的问题,那么到底有没有一个开源的、开箱即用的、功能相对完善的组件呢?答案就是本文的主角——Apache Zeppelin。
12868 0
字节跳动 Flink 单点恢复功能实践
在 Flink 现有的架构设计中,多流 Join 拓扑下单个 Task 失败会导致所有 Task 重新部署,耗时可能会持续几分钟,导致作业的输出断流,这对于线上业务来说是不可接受的。针对这一痛点,字节提出单点恢复的方案。
12707 0
Flink RocksDB 状态后端参数调优实践
RocksDB 的配置也是极为复杂的,可调整的参数多达百个,没有放之四海而皆准的优化方案。如果仅考虑 Flink 状态存储这一方面,我们仍然可以总结出一些相对普适的优化思路。本文先介绍一些基础知识,再列举方法。
7427 0
腾讯看点基于 Flink 的实时数仓及多维实时数据分析实践
当业务发展到一定规模,实时数据仓库是一个必要的基础服务。从数据驱动方面考虑,多维实时数据分析系统的重要性也不言而喻。但是当数据量巨大的情况下,拿腾讯看点来说,一天上报的数据量达到万亿级的规模,要实现极低延迟的实时计算和亚秒级的多维实时查询是有技术挑战的。
6249 0
+关注
thinktothings
主流大数据技术分享(Spark,Flink)
78
文章
4
问答
文章排行榜
最热
最新
相关电子书
更多
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载
冬季实战营第三期:MySQL数据库进阶实战
立即下载