吃透Flink架构:一个新版Connector的实现

发布于:2022-11-09 ⋅ 阅读:(11) ⋅ 点赞:(0) ⋅ 评论:(0)

前言

Flink 可以说已经是流计算领域的事实标准,其开源社区发展迅速,提出了很多改进计划(Flink Improvement Proposals,简称 FLIP [1])并不断迭代,几乎每个新的版本在功能、性能和使用便捷性上都有所提高。Flink 提供了丰富的数据连接器(connecotr)来连接各种数据源,内置了 kafka [2]、jdbc [3]、hive [4]、hbase [5]、elasticsearch [6]、file system [7] 等常见的 connector,此外 Flink 还提供了灵活的机制方便开发者开发新的 connector。对于 source connector 的开发,有基于传统的 SourceFunction [8] 的方式和基于 Flink 改进计划 FLIP-27 [9] 的 Source [10] 新架构的方式。本文首先介绍基于 SourceFunction 方式的不足,接着介绍 Source 新架构以及其设计上的深层思考,然后基于 Flink 1.13 ,以从零开发一个简单的 FileSource connector 为例,介绍开发 source connector 的基本要素,尽量做到理论与实践相结合加深大家的理解。

流计算 Oceanus 是大数据产品生态体系的实时化分析利器,是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标,加速企业实时化数字化的建设进程。流计算 Oceanus 提供了便捷的控制台环境,方便用户编写 SQL 分析语句、ETL 作业或者上传运行自定义 JAR 包,支持作业运维管理。欢迎大家试用,目前还有新用户 1 元秒杀活动,机会难得,不容错过。

Source 旧架构

在 Flink 1.12 之前,开发一个 source connector 通过实现 SourceFunction [8] 接口来完成,官方给出的通用的实现模式如下。当 source 开始发送数据时,run 方法被调用,其参数 SourceContext 用于发送数据。run 方法是一个无限循环,通过一个标识 isRunning 来跳出循环结束 source。批模式和流模式通常需要不同的处理逻辑,例如示例的批模式通过一个计数器来结束批数据。此外,还需要通过 checkpoint 锁来保证状态更新和数据发送的原子性。值得一提的是,Flink 在 SourceFunction 之上抽象出了 InputFormatSourceFunction,开发者只需要实现 InputFormat,批模式 source connector(如 HBase)通常基于 InputFormat 实现,当然 InputFormat 也可以用于流模式,在一定程度上体现了批流融合的思想,但整体上来看至少在接口层面上流批并没有完全一致。

public class ExampleCountSource implements SourceFunction<Long>, CheckpointedFunction {
      private long count = 0L;
      private volatile boolean isRunning = true;

      private transient ListState<Long> checkpointedCount;

      public void run(SourceContext<T> ctx) {
          while (isRunning && count < 1000) {
              // this synchronized block ensures that state checkpointing,
              // internal state updates and emission of elements are an atomic operation
              synchronized (ctx.getCheckpointLock()) {
                  ctx.collect(count);
                  count++;
              }
          }
      }

      public void cancel() {
          isRunning = false;
      }

      public void initializeState(FunctionInitializationContext context) {
          this.checkpointedCount = context
              .getOperatorStateStore()
              .getListState(new ListStateDescriptor<>("count", Long.class));

          if (context.isRestored()) {
              for (Long count : this.checkpointedCount.get()) {
                  this.count += count;
              }
          }
      }

      public void snapshotState(FunctionSnapshotContext context) {
          this.checkpointedCount.clear();
          this.checkpointedCount.add(count);
      }
 }

在基于 SourceFunction 的开发模式下,以 Kafka Source 为例,见下图,FlinkKafkaConsumer 为 SourceFunction 的实现类,该类中集中了 kafka partition 发现逻辑(KafkaPartitionDiscoverer)、数据读取逻辑(KafkaFetcher)、基于阻塞队列实现的生产者消费者模型(KafkaConsumerThread -> Handover -> SourceContext)等等。

我们可以发现,这种开发模式存在如下不足:

  1. 首先对于批模式和流模式需要不同的处理逻辑,不符合批流融合的业界趋势。
  2. 数据分片(例如 kafka partition、file source 的文件 split)和实际数据读取逻辑混合在 SourceFunction 中,导致复杂的实现。
  3. 数据分片在接口中并不明确,这使得很难以独立于 source 的方式实现某些功能,例如事件时间对齐(event-time alignment)、分区 watermarks(per-partition watermarks)、动态数据分片分配、工作窃取(work stealing)。
  4. 没有更好的方式来优化 Checkpoint 锁,在锁争用下,一些线程(例如 checkpoint 线程)可能无法获得锁。
  5. 没有通用的构建模式,每个源都需要实现自行实现复杂的线程模型,这使得开发和测试一个新的 source 变得困难,也提高了开发者对现有 source 的作出贡献的门槛。

有鉴于此,Flink 社区提出了 FLIP-27 [9] 的改进计划,并在 Flink 1.12 实现了基础框架,在 Flink 1.13 中 kafka、hive 和 file source 已移植到新架构,开源社区的 Flink CDC connector 2.0 [11] 也基于新架构实现。

Source 新架构

基于 FLIP-27 的 Source 新架构如上图所示,由两个主要部件组成:SplitEnumerator 和 SourceReader。SplitEnumerator 负责数据分片和分配,SourceReader 则负责具体分片数据的读取。当一个新的分片被 SplitEnumerator 添加到 SourceReader,首先初始化分片状态并放入状态哈希表中,然后分片被分配给 SplitReader 读取数据。读取的数据以小批量模式封装于 RecordsWithSplitIds 并放置于中间队列 Queue,这种批量数据模式可以提高性能。SourceReader 从 Queue 中获取一批数据,遍历每一条数据,并查找数据相应的分片状态,数据和分片状态一并传递给 RecordEmitter,RecordEmitter 先把数据传递给 SourceOutput,然后更新分片状态。状态哈希表中的状态在 checkpoint 时持久化到状态存储。

Source 新架构具有以下特点。

数据分片与数据读取分离。例如在 FileSource 中,SplitEnumerator 负责列出所有的文件,并有可能把文件按块或者范围进行切分,SourceReader 则负责具体的文件/块的数据读取。又例如在 KafkaSource 中,SplitEnumerator 负责发现需要读取的 kafka partition,SourceReader 则负责具体 partition 数据的读取。

批流融合。基于新架构开发的 Source 既可以工作于批模式也可以工作于流模式,批仅仅是有界的流。大多数情况下,只有 SplitEnumerator 需要感知数据源是否有界。例如对于 FileSource,批模式下 SplitEnumerator 只需要一次性的列出目录下的所有文件,流模式下则需要周期性的列出所有文件,并为新增的文件生成数据分片。对于 KafkaSource,批模式下 SplitEnumerator 列出处有的 partition,并把每个 partition 的当前最新的数据偏移作为数据分片的结束点,流模式下 SplitEnumerator 则把无穷大作为 partition 数据分片的结束点,即会持续的读取每个 partition 的新增数据,流模式下还可以周期性的监测 partition 的变化并为新增的 partitition 生成数据分片。

双向通信。SplitEnumerator 运行在 JobManager,SourceReader 运行在 TaskManager,SplitEnumerator 和 SourceReader 之间可以双向通信,SourceReader 可以主动向 SplitEnumerator 请求数据分片实现 pull 模式的数据分片分配(例如 FileSource),SplitEnumerator 也可以把数据分片直接分配给 SourceReader 实现 push 模式的分配(例如 KafkaSource)。此外,根据需要还可以定制化一些消息实现 SplitEnumerator 和 SourceReader 之间的交互需求。基于双向通信的能力,比较容易实现事件时间对齐(event-time alignment)的功能,实现数据分片之间事件时间的均衡推进。

资料领取直通车:大厂面试题锦集+视频教程icon-default.png?t=M85Bhttps://docs.qq.com/doc/DTlhVekRrZUdDUEpy

Linux服务器学习网站:C/C++Linux服务器开发/后台架构师icon-default.png?t=M85Bhttps://ke.qq.com/course/417774?flowToken=1028592

通用线程模型。考虑到外部数据源系统的客户端 API 调用方式的差异(阻塞、非阻塞、异步),SourceReader 在设计上支持单分片串行读取、多分片多路复用、多分片多线程三种模式。Flink 1.13 内核的 SingleThreadMultiplexSourceReaderBase/SingleThreadFetcherManager 抽象出的框架支持前两种线程模型,开发者基于此开发 source connector 变得容易。例如 FileSource 采用了单分片串行读取模式,在一个数据分片读取后,再向 SplitEnumerator 请求新的数据分片。KafkaSource 采用了多分片多路复用模式,SplitEnumerator 把启动时读取的 partition 列表和定期监测时发现的新的 partition 列表批量分配给 SourceReader,SourceReader 使用 KafkaConsumer API 读取所有分配到的 partition 的数据。

容错。SplitEnumerator 和 SourceReader 通过 Flink 的分布式快照机制持久化状态,发生异常时从状态恢复。通常 SplitEnumerator 状态保存了未分配的数据分片,SourceReader 状态保存了分配的数据分片以及分片读取状态(例如 kafka offset,文件 offset)。例如流模式下 FileSource 的 SplitEnumerator 状态保存了未分配的分片以及处理过的文件列表,并定期监测文件列表的变化,为新增文件生成数据分片;SourceReader 状态保存了当前读取的分片信息和文件读取 offset。

FileSource 开发实践

下面我们进入实际操作阶段,基于新架构开一个简单的 FileSource connector,该 connector 工作于流模式,读取指定目录下的文件,并定期监测新增文件。

初始化项目

  1. 我们使用 IntelliJ IDEA 作为开发工具,并按照 Flink 编码规范配置 IntelliJ IDEA [12],在菜单栏选择 "File -> New -> Project...",填写必要的信息,按照提示操作完成项目创建。

  1. 在 pom.xml [13] 文件添加必要的 Flink 依赖。
  2. Flink 基于 Java SPI 机制 l 发现和加载自定义 connector,我们在 resources 目录下创建目录 META-INF/services,并在该目录下创建文件 org.apache.flink.table.factories.Factory,文件内容为:
com.tencent.cloud.oceanus.connector.file.table.FileDynamicTableFactory
  1. 创建 Java 类 com.tencent.cloud.oceanus.connector.file.table.FileDynamicTableFactory 实现 DynamicTableSourceFactory
  2. 现在,我们项目初始化已经完成,可以在 IntelliJ IDEA 项目右侧选择 "Maven -> flink-connector-files -> LifeCyle -> package" 构建项目,能够在 target 目录下正确构建出名为 flink-connector-files-1.0.0.jar 的二进制包。

Connector 开发

我们按照 Flink 官方的自定义 connector 开发文档 [14] 来一步步完成 FileSource connector 的开发。

Metadata 层

简单起见,我们的 connector 只支持按行读取指定目录的文件,在 SQL 语句中按如下方式使用 connector。

CREATE TABLE test (
  `line` STRING
) WITH (
  'connector' = 'file',
  'path' = 'file:///path/to/files'
);

Planning 层

  1. 创建类 FileDynamicTableFactory [15],添加自定义 connector 标识 file和参数 path。校验参数并创建 FileDynamicSource [16]。
  2. FileDynamicSource [16] 创建 Runtime 层的 FileSource [17]。

Runtime 层

FileSource [17]

实现 Source [10] 接口,需要三个类型参数:第一类型参数为 Source 输出数据类型,由于我们的 connector 用于 SQL 作业场景,这里设置为 RowData 类型。第二个类型参数为数据分片类型 SourceSplit [18]。第三个类型参数为 SplitEnumerator checkpoint 数据类型。

FileSource 是一个工厂类,用于创建 SplitEnumerator、SourceReader、数据分片序列化器、SplitEnumerator checkpoint 序列化器。

FileSourceSplit [19]

实现 SourceSplit [18]。该类保存了数据分片 id、文件路径、数据分片起始位置的文件偏移(我们这里整个文件作为一个数据分片,不再细分,因此偏移始终为 0)、文件长度、文件读取进度(恢复时从该位置继续数据读取)。

FileSourceSplitSerializer [20]

数据分片序列化器,对 FileSourceSplit [19] 序列化和反序列化。数据分片在从 SplitEnumerator 传输到 SourceReader,以及被 SourceReader checkpoint 持久化时都需要序列化。

PendingSplitsCheckpoint [21]

SplitEnumerator checkpoint 数据,保存了未分配的分片以及处理过的文件列表。

PendingSplitsCheckpointSerializer [22]

SplitEnumerator checkpoint 序列化器,对 PendingSplitsCheckpoint [21] 序列化和反序列化。

FileSourceEnumerator [23]

定期监测目录下的文件,生成数据分片,并分配给 SourceReader。Flink 内核提供了定时回调接口

SplitEnumeratorContext#callAsync [24] 方便我们使用。这里我们采用 pull 模式的数据分片分配策略。

FileSourceReader [25]

继承自 SingleThreadMultiplexSourceReaderBase,在读取完一个数据分片(文件)后再向 FileSourceEnumerator [23] 请求下一个分片。我们需要实现数据分片状态初始化接口 initializedState [26],当新的数据分片加入时会调用该接口。实现接口 toSplitType [27],把可变的数据分片状态 FileSourceSplitState [28] 转换为不可变的数据分片 FileSourceSplit [19],checkpoint 时会调用该接口得到最新状态的 FileSourceSplit 并持久化。 FileSourceRecordEmitter [29] 发送数据到下游,并更新 FileSourceSplitState 的分片读取进度。具体分片数据读取逻辑在 FileSourceSplitReader [30] 实现,这里我们简单的每次读取一行数据。

Connector 测试

基本功能

  1. 从 Flink 官网下载已经编译好的二进制包 Apache Flink 1.13.3 for Scala 2.11 [31] 并解压,进入解压后的目录。拷贝我们开发的 connector 二进制包 flink-connector-files-1.0.0.jar [32] 到 lib 目录。
tar -zxvf flink-1.13.3-bin-scala_2.11.tgz
cd flink-1.13.3
cp flink-connector-files-1.0.0.jar lib/ -avi
  1. 启动本地集群。可在本地浏览器里打开 http://localhost:8081 进入 Flink UI 验证集群是否启动成功。
./bin/start-cluster.sh
  1. 创建测试数据目录,我们的 connector 从该目录下读取文件。然后进入 sql client 命令行。
mkdir -p /tmp/file-connector
./bin/sql-client.sh
  1. 在 sql client 命令行输入。
create table `source` (
    `line` STRING
) with (
    'connector' = 'file',
    'path' = '/tmp/file-connector'
);

select * from `source`;
  1. 我们往目录 /tmp/file-connector 写入几个文件测试一下,可见我们的 connector 能正常的读取文件数据。
cd /tmp/file-connector
echo "Hello World" > 1.txt
echo "tencent" > 2.txt
echo "oceanus" > 3.txt
echo "我爱我的祖国" > 4.txt

状态和容错

  1. 在 Flink 配置 conf/flink-conf.yaml 添加状态存储配置,设置 checkpoint 和 savepoint 目录,checkpoint 时间间隔,以及 Flink 重启策略。
state.backend: filesystem
state.checkpoints.dir: file:///tmp/flink-checkpoints
state.savepoints.dir: file:///tmp/flink-savepoints
execution.checkpointing.interval: 30s
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 30
restart-strategy.fixed-delay.delay: 5s
  1. 重启集群,并重新进入 sql client 命令行。
./bin/stop-cluster.sh
./bin/start-cluster.sh
./bin/sql-client.sh
  1. 重新提交作业。
create table `source` (
    `line` STRING
) with (
    'connector' = 'file',
    'path' = '/tmp/file-connector'
);

select * from `source`;
  1. 这次我们准备一个大的数据文件 bigfile.txt,例如有上千万行,拷贝文件到目录 /tmp/file-connector 。
  2. 在浏览器输入 http://localhost:8081 进入 Flink UI 界面,点击 'Running Jobs',可看到我们正在运行的作业,点击作业进入作业详情,选择 Checkpoints 标签页,然后选择 History 的子标签页,点击右侧的 Refresh 按钮。

  1. 在观察到一次成功的 checkpoint 之后,我们重启 taskmanager。
./bin/taskmanager.sh stop
./bin/taskmanager.sh start
  1. 稍等一会儿,我们可在 Flink UI 里观察到作业从 checkpoint 恢复的信息。

  1. 同时我们在 taskmanager 日志里可观察到作业恢复时的数据分片信息包含 checkpoint 时保存的文件读取 offset 信息。

FileSource 开发实践总结

作为流计算领域的事实标准,Flink 有着优秀的架构设计,其强大的可扩展能力让我们开发一个自定义 connector 变得简单。Flink 社区的文档也非常丰富和详细,这里我们按照 Flink 自定义 connector 开发文档,基于 FLIP-27 的 Source 新架构开发了一个简单 FileSource connector,并演示了其基本功能和错误恢复功能。我们在开发新的 connector 时可以多参考社区已有 connector 的设计和开发模式,甚至可以对现有 connector 进行功能增强来满足需求。我们这里的 FileSource connector 就是参考内核的 filesystem connector [7],进行了简化以方便大家理解。内核的 filesystem connector 不支持同一个文件的增量数据读取,如果我们实际业务场景一定需要这种能力,我们可以给 filesystem connector 添加该功能并贡献到社区。

总结

本文首先介绍了 Flink Source Connector 开发时基于传统的 SourceFunction 方式的不足,接着介绍了 FLIP-27 的 Source 新架构特点及其优势,然后基于新架构从零开发了一个简单的 FileSource connector,介绍开发 source connector 的基本要素。