陈胡：Apache SeaTunnel实现 非CDC数据抽取实践
- 2022 年 5 月 19 日
- 筆記
- DataFunTalk, 人工智能, 大数据, 算法
导读： 随着全球数据量的不断增长，越来越多的业务需要支撑高并发、高可用、可扩展、以及海量的数据存储，在这种情况下，适应各种场景的数据存储技术也不断的产生和发展。与此同时，各种数据库之间的同步与转化的需求也不断增多，数据集成成为大数据领域的热门方向，于是SeaTunnel应运而生。SeaTunnel是一个分布式、高性能、易扩展、易使用、用于海量数据（支持实时流式和离线批处理）同步和转化的数据集成平台，架构于Apache Spark和Apache Flink之上。本文主要介绍SeaTunnel 1.X在交管行业中的应用，以及其中如何实现从Oracle数据库把数据增量导入数仓这样一个具体的场景。
今天的介绍会围绕下面六点展开：
- SeaTunnel简介
- SeaTunnel应用场景
- 相关业务痛点
- 选择SeaTunnel的原因
- 具体实现方案
- 具体实现流程
—
01 SeaTunnel简介
下面对SeaTunnel从产品功能，技术特性、工作流程、环境依赖、用户使用等方面做一个总体的介绍。
1. Apache SeaTunnel整体介绍
互联网行业数据量非常大，对性能还有其他各方面的技术要求都非常高，在笔者所在的交管行业中，情况就不太一样，各方面的要求也没有互联网行业那么高，在具体的数据集成应用中，主要是使用SeaTunnel1.X版本。
上图所示内容引用了Apache SeaTunnel官网中的介绍。
Apache Spark对于分布式数据处理来说是一个伟大的进步，但是直接使用Spark框架还是有一定门槛的，SeaTunnel这个产品把业界使用Spark的优质经验固化到了其中，明显降低了学习成本，加快分布式数据处理能力在生产环境中落地。在SeaTunnel2.X版本中，除了Spark，也增加了对Flink的支持。
除此之外，SeaTunnel还可以较好的解决实际业务场景中碰到的下列问题：
- 数据丢失与重复
- 数据集成中任务堆积与延迟
- 数据同步较低的吞吐量
- Spark/Flink应用到生产环境周期较长、复杂度较高
- 缺少应用运行状态的监控
2. Apache SeaTunnel技术特性
SeaTunnel具备如上图所示的技术特性：
- 简单易用，开发配置简单、灵活，无需编码开发，支持通过SQL进行数据处理和聚合，使用成本低
- 分布式，高性能，经历大规模生产环境使用和海量数据检验，成熟稳定
- 模块化和插件化，内置丰富插件，并且可以开发定制个性化插件，支持热插拔，具备高扩展性
- 使用Spark/Flink作为底层数据同步引擎使其具备分布式执行能力
3. Apache SeaTunnel工作流程
SeaTunnel的架构和整个工作流程如下图所示，Input/Source [数据源输入] -> Filter/Transform [数据处理] -> Output/Sink [结果输出]，数据处理流水线由多个过滤器构成，以满足多种数据处理需求。如果用户习惯了SQL，也可以直接使用SQL构建数据处理管道，更加简单高效。目前，SeaTunnel支持的过滤器列表也在扩展中。
在插件方面，SeaTunnel已支持多种Input/Sink插件，同时也支持多种Filter/Transform处理插件，整体上基于系统非常易于扩展，用户还可以自行开发数据处理插件，具体如下：
- Input/Source 插件
Fake, File, Hive/Hdfs, Kafka, Jdbc, ClickHouse, TiDB, HBase, Kudu, S3, Socket, 自行开发的Input插件
- Filter/Transform 插件
Add, Checksum, Convert, Date, Drop, Grok, Json, Kv, Lowercase, Remove, Rename, Repartition, Replace, Sample, Split, Sql, Table, Truncate, Uppercase, Uuid, 自行开发的Filter/Transform插件
- Output/Sink 插件
Elasticsearch, File, Hdfs, Jdbc, Kafka, Mysql, ClickHouse, Stdout, 自行开发的Output 插件
4. Apache SeaTunnel环境依赖
SeaTunnel1.X支持Spark计算引擎，SeaTunnel2.X目前支持Spark/Flink两种计算引擎，在笔者的实际项目中使用的是SeaTunnel1.X版本。
5. Apache SeaTunnel用户使用情况
目前有很多公司都在使用SeaTunnel，其中不乏大型公司，例如：中国移动、腾讯云、今日头条、还有笔者所在的中电科。
—
02 SeaTunnel应用场景
SeaTunnel特别适合以下场景使用：
- 海量数据集成和ETL
- 海量数据聚合
- 多源数据处理
下面主要介绍SeaTunnel在交管行业中的应用。
1. 交管行业数据简介
在交管行业中，数据主要包括驾驶人、车辆相关的数据，平时在道路上发生的一些交通警情数据，交通违法数据，机动车登记信息，执勤执法的数据，交通事故以及其他一些互联网数据，这些数据的量不是很大，另外还有卡口过车、车辆GPS数据，这两种数据的数据量都比较大，例如一些省会城市，每秒钟至少有几千条过车数据，这些数据都是属于交管行业内的数据。
2. 交管行业数据特点
交管行业数据，跟互联网行业的数据还是有很大区别的，首先这些数据的体量大小不一，并且分布在内部的公安网以及智能专网，这两个网之间是物理隔离的，我们需要把这些数据在两个网络之间转移，在这个过程中，还要做一些数据处理。其次，在数据处理实时性方面的要求，并不是非常高，数据的更新频率也不是很高。然后，在数据安全方面，要求比较高，数据是不能丢的，同时对保密性要求也比较高，所以具体的数据也不能展示出来。
—
03 相关业务痛点
1. 数据抽取限制较多
在做业务的过程中，会有一些业务痛点，首先因为交管行业是政府行业，基本各个子平台的数据都是存储在Oracle数据库中的，我们需要把数据从Oracle数据库中抽取到我们的数仓里面，出于安全性的考虑，无法得到用户级别的权限，我们只能通过一些视图级别的用户权限去处理数据，对于数据源表结构的变更也无法及时知晓。其次，会话数是受到限制的，多线程抽取数据的话，如果会话数达到上限，连接就会受到影响，而且这个分配的用户也同时会用于其他用途。最后，我们在处理一些增量数据的时候，一般情况下需要一个增量列，用于保持一个增量更新，很多时候，是没办法确定哪些列可以作为增量列的。以上就是在做业务的过程中，经常会遇到的一些问题，下图也把这些问题列举了出来。
—
04 选择SeaTunnel的原因
最初的时候，做数据处理、数据抽取的时候，并没有使用SeaTunnel，而是使用Apache NiFi，这个工具功能比较强大而且全面，但是NiFi中用于数据处理的处理器比较多，而且数据处理链路中要做很多转换，所以需要对NiFi里面的各种组件要非常熟悉，对使用者的要求也比较高。
1. SeaTunnel的优势
我们一开始也用Spark程序做数据处理，对大数据相关人员的要求比较高，我们这边大数据人员比较少，有时处理一些新的需求的时候，会比较繁忙。如果不需要通过编码，而是直接使用工具，进行简单的配置就能实现的话，会带来较大的便利和效率的提高。
前面在SeaTunnel的介绍中，已经讲到SeaTunnel是比较易于使用的，安装部署方便，开箱即用，执行效率很高，因为它是分布式的，可以应用整个集群资源来做数据处理工作。
SeaTunnel无需编程，只要做简单的配置，并且它的Source和Sink都比较丰富，并且可以自己根据接口开发需要的插件，对数据源的权限要求也不高。
更加重要的是，SeaTunnel是首个进入Apache孵化的国人开源数据集成平台。
2. SeaTunnel的安装部署
如上图所示是SeaTunnel官方部署文档，只需要简单几步，就可以把SeaTunnel安装到我们的环境之中，然后就可以使用了。
3. SeaTunnel配置文件
下图所示是一个配置文件的示例，这个配置文件是SeaTunnel1.X版本的一个配置，一个完整的SeaTunnel配置包含spark, input, filter, output四部分，其中spark是spark相关的配置，例如，启动多少个executor，每个 executor使用多少核数的CPU，多少内存等，input可配置任意的input插件及其参数，具体参数随不同的input插件而变化，filter可配置任意的filter插件及其参数，具体参数随不同的filter插件而变化，filter中的多个插件按配置顺序形成了数据处理的pipeline, 上一个filter的输出是下一个filter的输入，通过input插件把数据取出，成为了spark里面的一个数据集，然后filter插件会对这个数据集做一些转换操作，output可配置任意的output插件及其参数，具体参数随不同的output插件而变化，filter处理完的数据，会发送给output中配置的每个插件
4. SeaTunnel插件支持
如下图所示，SeaTunnel支持的插件非常丰富，日常所能用到的基本都有。
这里面着重介绍一下filter插件中的sql插件，这个插件非常灵活，在用sql插件做转换操作时，只要是sparksql里面支持的函数等内容，都可以在这里使用，然后再output到目标数据存储，例如HDFS、Kafka、ES、Clickhouse等。
—
05 具体实现方案
接下来讲一下具体的实现方案，在我们具体的业务中，如何把这些行业数据从智能专网直接抽取到公安网中，这里会涉及到数据的增量更新。
1. 数据增量更新具体实现
当需要实现一个增量更新的时候，首先就是增量列的选择，之前提到原先是用NiFi来做增量更新，但是对增量列的支持不是特别好，尤其是对日期类型的支持不是很好。但是SeaTunnel对增量列的支持不受列的类型限制，可以比较灵活的进行选择。
2. 具体方法
实际业务当中，选取了记录的更新时间列作为增量列，每次数据抽取过来，会记录增量列的最大值，下次数据抽取时，可以从这个位置继续抽取数据，这个也是受以前写spark程序的启发，把checkpoint存储在HDFS里面。当然，增量列的选择，在实际应用中，除了更新时间，增量ID以外，还有其他业务字段可以做为增量列，增量列的选择一定是根据真正的业务需求，实时的程度和粒度来决定的。
—
06 具体实现流程
做数据增量更新，最重要的是实现的思路，接下来详细描述一下具体实现过程。
1. 确定运算资源
首先，如下图所示，先要确定计算资源，这里使用了spark，并且针对spark做了相关的配置。
2. 确定数据来源
选择一个增量列，对增量列每次产生的最大值（checkpoint），保存在HDFS一个具体的目录下。这里input插件选择HDFS，每次产生的那个增量数据，指向HDFS的一个具体路径下面，input插件有个通用参数叫做result_table_name，当指定result_table_name时，处理后的数据，会被注册为一个可供其他插件直接访问的数据集，或者被称为临时表。当增量列的最大值保存到HDFS之后，需要取出时，会保存在result_table_name指定的表中。接下来因为是从Oracle数据库中取数据，所以设置相应的Jdbc。当数据量比较大的时候，还可以指定分区列，这样的话，数据处理的效率会提高，详细配置，如下图所示。
3. 数据转换
下图所示是必要的数据转换，在实际业务中，需要做一个过滤操作，取出大于最大更新时间的数据，convert插件里面做的是中间的一些数据类型转换操作，最后使用了一个sql插件，用于记录本次取到的数据的一个最大值，用于下次取数的比较。
4. 数据输出
下图所示的是数据处理后的输出，也就是output插件对应的配置，具体是把数据抽取到Clickhouse里面。然后数据集里面，那个更新列的最大值，通过追加模式，写回到HDFS中，供下次使用。
5. 脚本和调度执行
整个过程是通过下图所示的shell脚本来做的，通过nohup后台执行的方式，利用Crontab进行调度执行，因为在我们实际的业务中，对定时调度的要求不是很高，所以可以采用Crontab或者开源的Dolphin Scheduler都是可以满足的。
下面的截图，是实际运行过程中，产生在HDFS上的增量文件，Crontab调度脚本，以及执行过程中产生的一些Yarn任务列表。
在上述整体数据处理过程中，由于实际情况的限制，尤其我们的数据源是高度受限的Oracle数据库。但是对于很多传统公司，如果老系统是以Oracle为主，并且掌控力度比较大的话，现在想做数据架构升级，需要迁移Oracle中的数据，那么可以采用CDC读取日志或者触发器的方式，把数据变化写入到消息队列里面，通过SeaTunnel就可以很容易的把数据实时写入到其他异构的数据库。
本文首发于微信公众号“DataFunTalk”。
