1、场景描述

例如订单库进行了分库分表，其实例如下图所示：

现在的需求是希望创建一个任务就将数据同步到MQ集群，而不是为每一个数据库实例单独创建一个任务，将其数据导入到MQ集群，因为同步任务除了库不同之外，表的结构、数据映射规则都是一致的。

 2、flinkx 的解决方案详解

 2.1 fink Stream API 开发基本流程

使用 Flink Stream API 编程的通用步骤如下图所示：

温馨提示：有关 Stream API 的详细内容将在后续的文章中展开，本文主要是关注 InputFormatSourceFunction，重点关注数据源的拆分。

 2.2 flinkx Reader(数据源)核心类图

在 flinkx 中将不同的数据源封装成一个个 Reader，其基类为 BaseDataReader，上图中主要罗列了如下几个关键的类体系：

• InputFormat
  flink 核心API，主要是对输入源进行数据切分、读取数据的抽象，其核心接口说明如下：1）void configure(Configuration parameters)对输入源进行额外的配置，该方法在 Input 的生命周期中只需调用一次。2）BaseStatistics getStatistics(BaseStatistics cachedStatistics)返回 input 的统计数据，如果不需要统计，在实现的时候可以直接返回 null。3）T[] createInputSplits(int minNumSplits)对输入数据进行数据切片，使之支持并行处理，数据切片相关类体系见：InputSplit。4）InputSplitAssigner getInputSplitAssigner(T[] inputSplits)获取 InputSplit 分配器，主要是在具体执行任务时如何获取下一个 InputSplit，其声明如下图所示：

5）void open(T split)
根据指定的数据分片 (InputSplit) 打开数据通道。为了加深对该方法的理解，下面看一下 Flink
x 关于 jdbc、es 的写入示例：

6）boolean reachedEnd()
数据是否已结束，在 Flink 中通常 InputFormat 的数据源通常表示有界数据 (DataSet)。

7）OT nextRecord(OT reuse)
从通道中获取下一条记录。

8）void close()
关闭。

• InputSplit
  数据分片根接口，只定义了如下方法：1） int getSplitNumber()获取当前分片所在所有分片中的序号。本文先简单介绍一下其通用实现子类：GenericInputSplit。1）int partitionNumber当前 split 所在的序号2）int totalNumberOfPartitions总分片数为了方便理解我们可以思考一下如下场景，对于一个数据量超过千万级别的表，在进行数据切分时可以考虑使用10个线程，即切割成 10分，那每一个数据线程查询数据时可以 id % totalNumberOfPartitions = partitionNumber，进行数据读取。
• SourceFunction
  Flink 源的抽象定义。
• RichFunction
  富函数，定义了生命周期、可获取运行时环境上下文。
• ParallelSourceFunction
  支持并行的 source function。
• RichParallelSourceFunction
  并行的富函数
• InputFormatSourceFunction
  Flink 默认提供的 RichParallelSourceFunction 实现类，可以当成是RichParallelSourceFunction 的通用写法，其内部的数据读取逻辑由 InputFormat 实现。
• BaseDataReader
  flinkx 数据读取基类，在 flinkx 中将所有的数据读取源封装成 Reader 。

 2.3 flinkx Reader构建 DataStream 流程

经过了上面类图的梳理，大家应该 flink 中提到的上述类的含义有了一个大概的理解，但如何运用呢？接下来将通过查阅 flinkx 的 DistributedJdbcDataReader(BaseDataReader的子类)的 readData 调用流程，体会一下其使用方法。

基本遵循创建 InputFormat、从而创建对应的 SourceFunction，然后通过 StreamExecutionEnvironment 的 addSource 方法将 SourceFunction 创建对应的 DataStreamSource。

 2.4 flinkx 针对数据库分库分表任务拆分解决方案

正如本文开头部分的场景描述那样，某订单系统被设计成4库8表，每一个库(Schema)中包含2个表，如何提高数据导出的性能呢，如何提高数据的抽取性能呢？通常的解决方案如下：

1. 首先按库按表进行拆分，即4库8表，可以进行切分8份，每一个数据分配处理一个实例中的1个表。
2. 单个表的数据抽取再进行拆分，例如按ID进行取模进一步分解。

flinkx 就是采取上面的策略，我们来看一下其具体做法。

Step1：首先先根据数据库实例、表进行拆分，按表维度组织成一个 DataSource 列表，后续将基于这个原始数据执行拆分算法。

接下来具体的任务拆分在 InputFormat 中实现，本实例在 DistributedJdbcInputFormat 的 createInputSplitsInternal 中。

DistributedJdbcInputFormat#createInputSplitsInternal

Step2：根据分区创建 inputSplit 数组，这里分区的概念就相当于上文提到方案中的第一条。

DistributedJdbcInputFormat#createInputSplitsInternal

Step3：如果指定了 splitKey 的任务拆分算法，首先 DistributedJdbcInputSplit 继承自 GenericInputSplit，总分区数为 numPartitions，然后生成数据库的参数，这里主要是生成 SQL Where 语句中的 splitKey mod totalNumberOfPartitions = partitionNumber，其中 splitKey 为分片键，例如 id，而 totalNumberOfPartitions 表示分区总数，partitionNumber 表示当前分片的序号，通过 SQL 取模函数进行数据拆分。

DistributedJdbcInputFormat#createInputSplitsInternal

Step4：如果未指定表级别的数据拆分键，则拆分策略是对 sourceList 进行拆分，即一些分区处理其中几个表。

关于 flinkx 中关于任务切分的介绍就到这里了。

 3、总结

本文主要是基于 flinkx 介绍 MySQL 分库分表情况下如何基于 flink 进行任务切分，简单介绍了 Flink 中关于基本的编程范式、InputFormat、SourceFunction 的基本类体系。

温馨提示：本文并没有太详细对 Flink API 进行深入研究，后续会单独对 Flink 内容进行逐一剖析，但 Flink 系列的文章组织，其文章的组织并不具备顺序性，笔者会在不断实践 Flink 的过程中对 FLink 进行剖析。