Flink Source 详解

embedded/2024/11/14 4:31:10/

Flink Source 详解

原文

flip-27
FLIP-27 介绍了新版本Source 接口定义及架构

相比于SourceFunction,新版本的Source更具灵活性,原因是将“splits数据获取”与真“正数据获取”逻辑进行了分离
在这里插入图片描述

重要部件

Source 作为工厂类,会创建以下两个重要部件

  1. SplitEnumerator

    • 通过createEnumerator创建

    • SplitEnumerator 响应request split请求

      • handleSplitRequest
    • 工作在SourceCoordinator (官方描述如下),可以理解为在JobMaster上运行一个单线程的逻辑,所以需要跟在worker上的reader通过rpc通信

      Where to run the enumerator
      There was a long discussion about where to run the enumerator which we documented in the appendix. The final approach we took was very similar to option 3 with a few differences. The approach is following.Each SplitEnumerator will be encapsulated in one SourceCoordinator. If there are multiple sources, multiple SourceCoordinator will there be. The SourceCoordinators will run in the JobMaster, but not as part of the ExecutionGraph. In this FLIP, we propose to failover the entire execution graph when the SplitEnumerator fails. A finer grained enumerator failover will be proposed in a later FLIP.
      
  2. SourceReader

    • 通过createReader创建

    • 工作在worker

    • 由于单独实现SourceReader过于复杂,官方抽象了3种比较通用的模型供开发者使用,MySqlSourceReader就是继承了SingleThreadMultiplexSourceReaderBase

      1. Sequential Single Split (File, database query, most bounded splits)
      2. Multi-split multiplexed (Kafka, Pulsar, Pravega, …)
      3. Multi-split multi-threaded (Kinesis, …)
        在这里插入图片描述

      在这里插入图片描述

      在这里插入图片描述

    • 使用了抽象后的类,开发者的关注点集中在实现一个SplitReader

      public interface SplitReader<E, SplitT extends SourceSplit> {RecordsWithSplitIds<E> fetch() throws InterruptedException;void handleSplitsChanges(Queue<SplitsChange<SplitT>> splitsChanges);void wakeUp();
      }
      
      1. fetch 获取数据,这里是包含了split信息的record
      2. 响应split改变
      3. 唤醒
  3. RecordEmitter

    1. The RecordsWithSplitIds returned by the SplitReader will be passed to an RecordEmitter one by one.
    2. The RecordEmitter is responsible for the following:
      • Convert the raw record type into the eventual record type
      • Provide an event time timestamp for the record that it processes.
    3. 在 emitRecord 方法中实现

由于通信使用mail风格的rpc(单线程串行),所以响应函数需要保证非阻塞,所以后面可以看到无论enumerator还是reader的最终响应都是在异步线程池中

Non-blocking progress methods, to it supports running in an actor/mailbox/dispatcher style operator

MysqlSource 举例

flink cdc中的MysqlSource来举例分析

  1. MysqlSource
    • 通过 createEnumerator 创建 MySqlSourceEnumerator

      • 初始化调用start
        • 调用splitAssigner.open()
          • splitAssigner 是获取/分配split动作的真正实现
            • 创建异步线程,填充remainingSplits
      • handleSplitRequest 响应空闲worker的请求
        • assignSplits
          • splitAssigner.getNext()
            • 从 remainingSplits 拿一个可用的split
      • 调用 context.assignSplit 发送 AddSplitEvent
      • MySqlSourceEnumerator 中 splitAssigner 的实现说明
        • splitAssigner 默认实现是 MySqlHybridSplitAssigner
          • hybrid的含义,启动分为两个步骤 1. 读取全量数据 2. 全量数据读取完毕后读取增量数据。将两种模式混合在一起被称为hybird。所以MySqlSnapshotSplitAssigner可以创建两种split
            1. 通过MySqlSnapshotSplitAssigner创建存量数据的split
              • 在读取存量数据时通过chunkSplitter切分为多个split,之后分发给多个reader并行读取
                • chunkSplitter 通过 chunkKey 的范围将存量数据切分
                • 用户可以手动设置chunkKey,否则使用主key作为chunkKey,切分split
            2. 通过 createBinlogSplit 创建增量数据的split
              • 只assign一次binlog的split
              • 只能分发给一个reader,所以在进入增量模式后flink实际所有并行度上只有一个source有数据
                在这里插入图片描述
    • 通过 createReader 创建 MySqlSourceReader

      • 创建 SingleThreadFetcherManager 传入 elementQueue splitReaderSupplier
        • elementQueue: io线程和主线程公用队列,io线程写,主线程读
        • splitReaderSupplier: split reader的工厂
        • SingleThreadFetcherManager 启动后创建线程池
      • sourceOperator 收到 AddSplitEvent 调用 sourceReader.addSplits 这里 sourceReader 是 MySqlSourceReader
        • readerBase 中会调用 splitFetcherManager.addSplits(splits);
          • 由于使用的是 SingleThreadFetcherManager,所以addSplits中永远看到只同时存在一个fetcher
            • fetcher 初始化时加入默认任务 FetchTask 构造的时候传入 elementQueue 传入构造好的 splitReader
            • fetcher addSplits时加入任务 AddSplitsTask
            • fetcher 启动时调用 startFetcher
              • 调用 executors.submit(fetcher); 提交到线程池
              • 线程池中运行 runOnce
                • FetchTask 调用 splitReader.fetch() 获取records 写入 elementQueue
      • 主线程 SourceReaderBase 中的 pollNext 会被框架调用
        • 调用 getNextFetch
          • elementsQueue.poll() 取得 records
            在这里插入图片描述

其他

在Flink CDC 3.0 中

Flink Composer 中使用 WatermarkStrategy.noWatermarks()

 return env.fromSource(sourceProvider.getSource(),WatermarkStrategy.noWatermarks(),sourceDef.getName().orElse(generateDefaultSourceName(sourceDef)),new EventTypeInfo()).setParallelism(sourceParallelism);

很合理,因为pipeline的定义中不会出现聚合函数 window函数


http://www.ppmy.cn/embedded/137410.html

相关文章

Django博客网站上线前准备事项

Django博客网站上线前准备事项 1. 功能完善与测试 确保博客网站具备以下基础功能&#xff0c;并且经过充分测试&#xff1a; 用户认证&#xff1a;注册、登录、登出、密码重置。文章管理&#xff1a;文章的创建、编辑、发布、删除。分类与标签&#xff1a;文章分类和标签的管…

【css】overflow: hidden效果

1. 不添加overflow: hidden 1.1 效果 上面无圆角 1.2 代码 <template><view class"parent"><view class"child1">child1</view><view class"child2">child2</view></view></template><…

YOLOv11实战宠物狗分类

本文采用YOLOv11作为核心算法框架&#xff0c;结合PyQt5构建用户界面&#xff0c;使用Python3进行开发。YOLOv11以其高效的特征提取能力&#xff0c;在多个图像分类任务中展现出卓越性能。本研究针对5种宠物狗数据集进行训练和优化&#xff0c;该数据集包含丰富的宠物狗图像样本…

【Linux:IO多路复用(select函数)

什么是IO多路复用&#xff1f; 一种网络通信的手段&#xff0c;IO多路复用可以同时监测多个文件描述符&#xff0c;且这个过程是阻塞的&#xff0c;当检测有文件描述符就绪&#xff0c;程序的阻塞就会解除&#xff0c;就可以通过这些就绪的文件描述符进行通信。通过这种方式在…

使用YOLOv9进行图像与视频检测

大家好&#xff0c;YOLOv9 与其前身v8一样&#xff0c;专注于识别和精确定位图像和视频中的对象。本文将介绍如何使用YOLOv9进行图像与视频检测&#xff0c;自动驾驶汽车、安全系统和高级图像搜索等应用在很大程度上依赖于此功能&#xff0c;YOLOv9 引入了比 YOLOv8 更令人印象…

hbase集成phoenix

1.环境 环境准备 三台节点zookeeper三节点hadoop三节点hbase三节点 2.pheonix集成 官网下载地址&#xff0c;需挂梯子&#xff0c;使用官网推荐的对应hbase版本即可 https://phoenix.apache.org/download.html下载及解压 wget https://dlcdn.apache.org/phoenix/phoenix-…

npm list @types/node 命令用于列出当前项目中 @types/node 包及其依赖关系

文章目录 作用示例常用选项示例命令注意事项 1、实战举例**解决方法**1. **锁定唯一的 types/node 版本**2. **清理依赖并重新安装**3. **设置 tsconfig.json 的 types**4. **验证 Promise 类型支持** **总结** npm list types/node 命令用于列出当前项目中 types/node 包及其…

Typescript类型运算符、关键字以及内置高级类型

目录 类型运算符 |运算符 &运算符 类型关键字 Infer 示例 extends typeof keyof in 注意 内置高级类型 类型 PropertyKey ArrayBufferLike 对类型集合的操作 Exclude,> Extract,> 对类型的操作 NonNullable Readonly Required Partial Pick…