流处理开源项目flinkStreamSQL FlinkX简介

墨蓝 2022-10-14 00:52 238阅读 0赞

## flinkStreamSQL ##

源码下载地址：

https://github.com/DTStack/flinkStreamSQL

熟悉flink的应该都了解，flink支持流表之间的join，但到1.6为止都不支持流表和维表的join。浪尖最近，也在开发流平台，需要到flink流表和维表的join。那么针对这个大家第一印象，可以写个算子去实现，比如map等。但是浪尖这里开发的流平台不是说自己写api，而是用户通过sql去实现创建source，sink，udf，sql等，这个时候要进行维表join，大家可能是想到了udf。是的对于只有一个维表的情况下使用udf比较方便，但是多个维表，相对就麻烦很多了。

而基于flink开发的flinkStreamSQL主要是实现了flink 流表和维表的join，其主要功能如下：

*  自定义create table 语法（包括源表,输出表,维表）
 *  自定义create function 语法
 *  实现了流与维表的join

源码思路很清晰，主要是两个步骤：

用flink api实现维表的功能: 要实现维表功能就要用到 flink Aysnc I/O 这个功能，是由阿里巴巴贡献给apache flink的。关于异步IO的介绍，可以参考：https://yq.aliyun.com/articles/457385

解析流与维表join的sql语法转化成底层的flinkAPI

为了方便大家阅读，把维表转化的过程主要函数贴出来吧：

主函数
    Main#main
    
    SQL解析
    SqlTree sqlTree = SqlParser.parseSql(sql)
    拆读
    SqlParser#parseSql
    TableInfoParserFactory#parseWithTableType
    
    
    注册表
    registerTable
    
    存在维表的话，维表转换与逻辑sql执行
    SideSqlExec#exec
    也即是
    sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);
    
    不存在维表的话
    tableEnv.sqlUpdate(result.getExecSql());

## FlinkX ##

FlinkX开源地址：[https://github.com/DTStack/flinkx][https_github.com_DTStack_flinkx]

FlinkX是一款基于Flink的分布式离线/实时数据同步插件，可实现多种异构数据源高效的数据同步，其由袋鼠云于2016年初步研发完成，目前有稳定的研发团队持续维护，已在Github上开源（开源地址详见文章末尾）。并于今年6年份，完成批流统一，离线计算与流计算的数据同步任务都可基于FlinkX实现。

FlinkX将不同的数据源库抽象成不同的Reader插件，目标库抽象成不同的Writer插件，具有以下特点：

1.  基于Flink开发，支持分布式运行；
2.  双向读写，某数据库既可以作为源库，也可以作为目标库；
3.  支持多种异构数据源，可实现MySQL、Oracle、SQLServer、Hive、Hbase等近20种数据源的双向采集。
4.  高扩展性，强灵活性，新扩展的数据源可与现有数据源可即时互通。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE2NjM2NDE_size_16_color_FFFFFF_t_70][]

使用安装部署，可以参考：

[https://github.com/DTStack/flinkx/blob/1.10\_release/docs/quickstart.md][https_github.com_DTStack_flinkx_blob_1.10_release_docs_quickstart.md]

# **FlinkX应用场景** #

FllikX数据同步插件主要应用于大数据开发平台的数据同步/数据集成模块，通常采用将底层高效的同步插件和界面化的配置方式相结合的方式，使大数据开发人员可简洁、快速的完成数据同步任务开发。实现将业务数据库的数据同步至大数据存储平台，从而进行数据建模开发，以及数据开发完成后，将大数据处理好的结果数据同步至业务的应用数据库，供企业数据业务使用。

# **FlinkX工作原理详解** #

FlinkX基于Flink实现，其选型及优势详见[https://mp.weixin.qq.com/s/uQbGLY3\_cj0h2H\_PZZFRGw][https_mp.weixin.qq.com_s_uQbGLY3_cj0h2H_PZZFRGw]。

不同的数据源头被抽象成不同的Reader插件，不同的数据目标被抽象成不同的Writer插件。理论上，FlinkX框架可以支持任意数据源类型的数据同步工作。作为一套生态系统，每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。

在底层实现上，FlinkX依赖Flink，数据同步任务会被翻译成StreamGraph在Flink上执行

FlinkX数据同步任务的本质是一个Flink程序，读出写入的数据同步任务会被翻译成StreamGraph在Flink执行，FlinkX开发者只需要关注InputFormat和OutputFormat接口实现即可。工作原理如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE2NjM2NDE_size_16_color_FFFFFF_t_70 1][]

Engine是袋鼠云封装的任务调度引擎，WEB端配置好的数据同步任务首先会提交至任务调度引擎，Template模块根据同步任务的配置信息加载源数据库和目标数据库对应的Reader和 Writer插件，Reader插件实现InputFormat接口，从数据库获取DataStream对象，Writer插件实现OutFormat接口，将目标数据库与DataStream对象相关联，从而通过DataStream对象将读出写入串接在一起，组装成一个Flink任务提交至Flink集群上进行运行。

之前基于Flink的分片、累加器特性，解决了数据同步过程中的增量同步、多通道控制、脏数据管理与错误管理等场景。最近半年基于Flink的checkpoint机制，实现了断点续传、流数据续跑等功能，来了解一下它的新特性吧。

（1）断点续传

数据同步过程中，假如一个任务要同步500G的数据到目标库，已经跑了15min，但到400G的时候由于集群资源不够、网络等因素数据同步失败了，若需要重头跑此任务，想必该同学要抓狂了。FlinkX基于checkpoin机制可支持断点续传，当同步任务由于上述原因失败时，不需要重跑任务，只需从断点继续同步，节省重跑时间和集群资源。

Flink的Checkpoint功能是其实现容错的核心功能，它能够根据配置周期性地对任务中的Operator/task的状态生成快照，将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常。

并且断点续传可和任务失败重试机制配合，即当任务执行失败，系统会自动进行重试，若重试成功则系统会接着断点位置继续同步，从而减少人为运维。

（2）实时采集与续跑

今年6月份，袋鼠云数栈研发团队基于FlinkX实现批流数据采集统一，可对MySQL Binlog、Filebeats、Kafka等数据源进行实时采集，并可写入Kafka、Hive、HDFS、Greenplum等数据源，采集任务也支持作业并发数与作业速率的限制，以及脏数据管理。并基于checkpint机制，可实现实时采集任务的续跑。当产生业务数据或Flink程序引起的采集进程中断时，可基于Flink定期存储的快照，对流数据的读取节点进行保存，从而在进行故障修复时，可选择历史保存的数据断点进行续跑操作，保证数据的完整性。此功能在袋鼠云的StreamWorks产品中实现，欢迎大家了解。

（3）流数据的脏数据管理

之前在BatchWorks离线计算产品中，已实现离线数据同步的脏数据管理，并基于Flink的累加器实现脏数据的错误管理，当错误量达到配置时，置任务失败。目前流数据实时采集也支持了此功能，即在将源库数据写入目标库的过程中，将错误记录进行存储，以便后续分析数据同步过程中的脏数据，并进行处理。但由于是流数据采集，任务具有不间断性，没有进行错误数记录达到阈值的触发任务停止操作，待后续用户自行对脏数据分析，进行处理。

（4）数据写入至Greenplum、Oceanbase数据源

Greenplum是基于PostgreSQL的MPP数据库，支持海量数据的存储与管理，目前在市场上也被很多企业采用。于最近，数栈基于FlinkX实现多类型数据源写入Greenplum，除全量同步外，也支持部分数据库增量同步写入。OceanBase是阿里研发的一款可扩展的金融领域关系型数据库，其用法与MySQL基本一致，实现OceanBase的数据读入写出也是基于jdbc的连接方式，进行数据表与字段的同步与写入，也支持对OceanBase进行增量写入，以及作业同步通道、并发的控制。

写入Greenplum等关系数据库时，默认是不使用事务的，因为数据量特别大的情况下，一旦任务失败，就会对业务数据库产生巨大的影响。但是在开启断点续传的时候必须开启事务，如果数据库不支持事务，则无法实现断点续传的功能。开启断点续传时，会在Flink生成快照的时候提交事务，把当前的数据写入数据库，如果两次快照期间任务失败了，则这次事务里的数据不会写入数据库，任务恢复时从上一次快照记录的位置继续同步数据，这样就可以做到任务多次失败续跑的情况下准确的同步数据。

若需了解具体技术实现，详见[https://mp.weixin.qq.com/s/VknlH8L2kpnlcJ3990ZkUw][https_mp.weixin.qq.com_s_VknlH8L2kpnlcJ3990ZkUw]

**写在后面：**

FlinkX经过袋鼠云内部使用以及在大量的数据中台项目中实践，目前支持以下数据源。且在FlinkX的高扩展特性下，将持续支撑更多的数据源，满足不同企业的需要，目前正在研发Clickhouse、Hana等数据源。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE2NjM2NDE_size_16_color_FFFFFF_t_70 2][]

[https_github.com_DTStack_flinkx]: https://github.com/DTStack/flinkx
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE2NjM2NDE_size_16_color_FFFFFF_t_70]: /images/20221014/bf49c922156c4e679f031f0f35ec0e41.png
[https_github.com_DTStack_flinkx_blob_1.10_release_docs_quickstart.md]: https://github.com/DTStack/flinkx/blob/1.10_release/docs/quickstart.md
[https_mp.weixin.qq.com_s_uQbGLY3_cj0h2H_PZZFRGw]: https://mp.weixin.qq.com/s/uQbGLY3_cj0h2H_PZZFRGw
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE2NjM2NDE_size_16_color_FFFFFF_t_70 1]: /images/20221014/2999ad168a30458dba6f0be41a2dcbc6.png
[https_mp.weixin.qq.com_s_VknlH8L2kpnlcJ3990ZkUw]: https://mp.weixin.qq.com/s/VknlH8L2kpnlcJ3990ZkUw
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE2NjM2NDE_size_16_color_FFFFFF_t_70 2]: /images/20221014/bbb5c77a3ae749af92e70d7f39a43feb.png