流行的开源高性能数据同步工具 - Apache SeaTunnel 整体架构运行原理

概述

背景

数据集成在现代企业的数据治理和决策支持中扮演着至关重要的角色。随着数据源的多样化和数据量的迅速增长，企业需要具备强大的数据集成能力来高效地处理和分析数据。SeaTunnel通过其高度可扩展和灵活的架构，帮助企业快速实现多源数据的采集、处理和加载。

目标

本说明旨在为系统架构师、开发者、实施工程师和运维人员提供SeaTunnel的技术架构介绍，以支持其在项目实施和优化中进行高效设计和运维。

技术架构概述

整体架构说明

SeaTunnel 主要由一套数据同步处理的 API 和核心计算引擎组成，包括三个主要的服务：CoordinatorService、TaskExecutionService 和 SlotService

CoordinatorService

CoordinatorService 是集群的 Master 服务，提供每个作业从 LogicalDag 到 ExecutionDag，再到 PhysicalDag 的生成流程，并且最终创建作业的 JobMaster 进行作业的调度执行和状态监控。

TaskExecutionService

TaskExecutionService 是集群的 Worker 服务，提供作业中每个 Task 的真正运行时环境，TaskExecutionService 使用 Dynamic Thread Sharing 技术降低 CPU 使用

SlotService

SlotService 在集群每个节点上都会运行，主要负责节点上资源的划分、申请和回收

SeaTunnel 模块说明

模块名	介绍
seatunnel-api	SeaTunnel connector V2 API 模块
seatunnel-common	SeaTunnel 通用模块
seatunnel-connectors-v2	SeaTunnel connector V2 模块, connector V2 处于社区重点开发中
seatunnel-core/seatunnel-spark-starter	SeaTunnel connector V2 的 Spark 引擎核心启动模块
seatunnel-core/seatunnel-flink-starter	SeaTunnel connector V2 的 Flink 引擎核心启动模块
seatunnel-core/seatunnel-starter	SeaTunnel connector V2 的 SeaTunnel 引擎核心启动模块
seatunnel-e2e	SeaTunnel 端到端测试模块
seatunnel-examples	SeaTunnel 本地案例模块， 开发者可以用来单元测试和集成测试
seatunnel-engine	SeaTunnel 引擎模块, seatunnel-engine 是 SeaTunnel 社区新开发的计算引擎，用来实现数据同步
seatunnel-formats	SeaTunnel 格式化模块，用来提供格式化数据的能力
seatunnel-plugin-discovery	SeaTunnel 插件发现模块，用来加载类路径中的SPI插件
seatunnel-transforms-v2	SeaTunnel transform V2 模块, transform V2 处于社区重点开发中
seatunnel-translation	SeaTunnel translation 模块, 用来适配Connector V2 和其他计算引擎， 例如Spark、Flink等

架构层次

数据集成产品的架构设计具有高度的模块化和可扩展性，基于SeaTunnel Zeta高性能集成引擎，可以支持数据批量和实时高效同步、批流一体等能力，SeaTunnel整体分为以下几个主要层次：

• 数据源接入层(source)：包括各种数据源的连接器和适配器。
• 数据转换层(transform)：支持数据清洗和转换操作比如过滤数据、添加字段、自定义 SQL 等多种操作。
• 数据写入层(sink)：多种数据仓库和数据库等目标数据源的输出，包括关系型数据库、NoSQL 数据库、分布式存储系统等。

可接入的数据源类型

• 关系型数据库：支持 JDBC驱动，采用并行提取策略优化大数据量场景下的同步性能。
• NoSQL 数据库：如 MongoDB、Cassandra，支持以 JSON、BSON 等格式传输数据，保证高效的文档处理。
• 文件系统：支持从本地文件系统、HDFS、Amazon S3 等读取结构化和半结构化数据，支持 CSV、JSON、Parquet等格式。
• 消息队列：提供与 Kafka等消息系统 的原生集成，支持高并发下的流式数据消费。
• Sass接口：如 http接口等

数据处理流与执行逻辑

数据管道设计

数据管道设计考虑到不同的数据源、数据处理需求和目标存储，通过以下步骤实现：

• 数据提取 (Extract)：通过高效的多线程和批量拉取实现数据的快速提取。
• 数据加载 (Load)：数据的目标存储和输出，支持幂等性和分布式一致性。
• 数据转换 (transform)：包括数据清洗、格式转换、逻辑处理等。

Zeta引擎工作原理

SeaTunnel Engine 由三个主要的服务组成：CoordinatorService、TaskExecutionService 和 SlotService。

CoordinatorService

CoordinatorService 是集群的 Master 服务，提供了每个作业从 LogicalDag 到 ExecutionDag，再到 PhysicalDag 的生成流程，并最终创建作业的 JobMaster 进行作业的调度执行和状态监控。CoordinatorService中 主要由 4 个大的功能模块组成：

1. JobMaster，负责单个作业的 LogicalDag 到 ExecutionDag，再到 PhysicalDag 的生成流程，并由 PipelineBaseScheduler 进行调度运行。
2. CheckpointCoordinator，负责作业的 Checkpoint 流程控制。
3. ResourceManager，负责作业资源的申请和管理，目前支持 Standalone 模式，未来会支持 On Yarn 和 On K8s。
4. Metrics Service，负责作业监控信息的统计和汇总。

客户端在提交任务时会找到master节点并将任务提交到CoordinatorService服务上，CoordinatorService会缓存任务信息并等待任务执行结束。当任务结束后再对任务进行归档处理。

TaskExecutionService

TaskExecutionService 是集群的 Worker 服务，提供了作业中每个 Task 的真正运行时环境，TaskExecutionService 使用了 Dynamic Thread Sharing 技术降低 CPU 使用。

TaskExecutionService 是一个执行任务的服务，将在每个节点上运行一个实例。它从 JobMaster 接收 TaskGroup 并在其中运行 Task。并维护TaskID->TaskContext，对Task的具体操作都封装在TaskContext中。而Task内部持有OperationService，也就是说Task可以通过OperationService远程调用其他Task或JobMaster进行通信。

SlotService

SlotService 在集群每个节点上都会运行，主要负责节点上资源的划分、申请和回收。SlotService用于管理集群的可用Slot资源。SlotService运行在所有节点上并定期向master上报资源信息。

1. SeaTunnel 运行流程
   1. 第一步：jobconf转换为LogicDag
      通过job的配置文件定义job流程，因此 SeaTunnelClient 需要做的第一件事是解析job配置文件并生成action列表。action 类似于 Flink 中的 operator，是对 SeaTunnel API 的封装。一个action包含 SeaTunnelSource 或 SeaTunnelTransform 或 SeaTunnelSink 的实例。每个action都需要知道它自己的上游。
      这一块主要是 Action 接口来负责的。
      目前支持 SourceAction 、SinkAction 和 TransformAction 三种类型的action。如果只有一个 Source和一个Sink，和一个 Transform

如果有多个source或多个transform或多个sink，我们将依赖 source_table_name 和 result_table_name 来构建action pipeline。因此，在这种情况下，result_table_name对于source action是必需的，而所有result_table_name 和 source_table_name对于transform action都是必需的。最后，source_table_name 是 sink action必需的。

1. 1. 第二步：LogicPlan转换成PhysicalPlan

SeaTunnel引擎会收到客户端发送过来的逻辑计划，引擎需要将其转化为可以直接执行的物理计划。因此，需要对逻辑执行计划进行处理，通过转换生成物理计划。具体过程如下：

• 逻辑计划

收到逻辑计划，我们需要去除多余的Actions，并验证Schema（Transform2和Transform 5应该是一样的）

• 执行计划

转换为执行计划时：

（1）Transforms需要合并，合并的依据是Transform后数据是否会被拆分（如果没有shuffle则会将transform合并）。

（2）将shuffle action转换成队列。

（3）拆分多个pipeline。

• 物理计划

将Pipeline按照并行度拆分成单独的可执行任务，并且同时需要添加SourceSplitEnumerator和SinkAggregatedCommitter任务，可以将任务发送到executionService。然后任务就可以正常运行了。

1. 第三步：将taskGroup调度到指定node上等待运行

在master节点上将physicalPlan拆分成pipeline并将pipeline拆分成taskGroup分别调度到不同节点上进行执行。

• physicalPlan：用户提交的job被解析成可以运行的执行计划。
• pipeline：在pipeline中的任务只有pipeline的上游和下游算子，不同pipeline没有相关联的算子。
• taskGroup：每一个执行计划顶点将会创建一个taskGroup,一个taskGroup包含一个或者多个task，每一个taskGroup需要一个单位的计算资源。taskGroup是任务分配和执行的最小单位。如下：

下图是 SeaTunnel 整体运行流程图

运行流程图

Zeta 引擎的优势与特点

SeaTunnel Zeta 引擎采用无中心化架构，为了在不依赖第三方服务组件(如 Zookeeper)的情况下实现集群的自治和作业的容错，SeaTunnel Zeta使用了 Hazelcast作为底层依赖。Hazelcast 提供了一个分布式内存网络，让用户可以像在本地操作普通 Java 集合一样来操作一个分布式的集合，SeaTunnel 将作业的状态信息保存在Hazelcast 的内存网格中，当 Master 节点切换后，可以基于 Hazelcast 内存网格中的数据进行作业状态的恢复。同时，我们还实现了 Hazelcast 内存网格数据的持久化，以WAL 的方式将作业状态信息持久化到存储中（JDBC 协议的数据库、HDFS、云存储）。这样，即使整个集群挂掉重启，也可以修复作业的运行时信息。

数据缓存

SeaTunnel Zeta Engine 与传统的Spark/Flink计算引擎不同，是专门用来做数据同步的引擎。SeaTunnel 引擎天然支持数据 Cache，当集群中有多个同步作业共用一个数据源时，SeaTunnel 引擎会自动启用数据 Cache，由一个作业的 Source 将数据读取后写入 Cache 中，其它所有作业不再从数据源读取数据，而是自动被优化为从 Cache 中读取数据。这样做的好处是可以降低数据源的读取压力，降低数据同步对数据源的影响。

速度控制

SeaTunnel Engine 支持数据同步时的速度限制，这在高并发读取数据源时非常有用，合理的速度限制既可以保证数据按时同步完成，又可以尽量减小对数据源造成的压力影响。

共享连接池，降低数据库压力

数据库连接对数据库来说是昂贵的资源，过多的数据库连接会对数据库造成极大的压力，导致数据库读写延迟稳定性降低，这对业务数据库来说是非常严重的事故。为了解决这个问题，SeaTunnel Engine 使用共享连接池的方式，保证多张表可以共用 JDBC 连接，从而降低数据库连接的使用。

断点续传（增量/全量），让用户无感知

Zeta Engine 支持离线同步下的断点续传。在数据量较大时，一次数据同步作业往往需要运行几十分钟或几个小时，如果中间作业挂了重跑，那意味着浪费时间。SeaTunnel Engine 会在离线同步的过程中不断地进行状态的保存（检查点），作业挂掉重跑时会从上一次的检查点继续运行，这有效解决了节点宕机等硬件问题可能导致的数据延迟。

Schema evolution 的路线

模式演化是一种允许用户轻松更改表的当前模式以适应随时间变化的数据的功能。最常见的是，在执行追加或覆盖操作时使用它，以自动调整模式以包括一个或多个新列。

更细粒度的容错设计

Flink 的设计是整个作业级别的容错和回滚，表现为如果某一个 task 失败，那整个作业都会进行回滚重启操作。SeaTunnel Engine 在设计时考虑到了在数据同步场景下，很多q情况下一个 task 的失败应该只需要和它有上下游关系的 task 需要关注容错。基于这一设计原则，SeaTunnel Engine 会先按用户配置的作业配置文件生成逻辑 DAG，再对逻辑DAG 进行优化，最终生成以 Pipeline（一个作业 DAG 中的一个连通子图）为粒度进行作业的调用和容错。

一个典型的使用场景是：
使用 CDC 连接器从 MySQL 的 binlog 中读取数据后写入另一个 MySQL，如果使用 Flink 或 Spark 引擎，一旦目标端 MySQL 无法写入，会导致 CDC 读取 binlog 的任务也会中止，如果 MySQL 被设置了 log 的过期时间，会出现目标端 MySQL 问题解决了，但源 MySQL 的日志被清除了，进而引发数据丢失等问题。
SeaTunnel Engine 会自动优化这一同步任务，自动添加源到目标端的 Cache，再进一步将这个作业优化成两个 Pipeline，pipeline#1 负责从 CDC 读取数据并写入 SeaTunnel Cache，pipeline#2 负责从 SeaTunnel Cache 读取数据并写入目标MySQL。如果目标端 MySQL 有问题导致无法写入，这个同步作业的 pipeline#2 会中止，pipeline#1 依然正常运行。这种设计从根本上解决了上述的问题，更符合数据同步引擎的处理逻辑。

动态共享线程，减少资源占用

SeaTunnel Engine 的 Task 设计使用了共享线程的技术，区别于 Flink/Spark，SeaTunnel Engine 不会简单的让一个 Task 占用一个线程，而是通过一种动态感知的方式——动态线程共享(Dynamic Thread Sharing)来判断一个 Task 应该和其它 Task 共享一个线程还是应该独占一个线程。
多线程并行计算和单线程串行计算相比有更好的性能优势，但如果每个 Task 都占使用一个独立线程来运行，当数据同步的表比较多，Task 数量大时，会在 Worker 节点上启动非常多的线程。在 CPU 核心数固定的情况下，线程数并不是越多越好，当线程数量过多时，CPU 需要花大量的时间进行线程的上下文切换，这反而会影响计算性能。
Flink/Spark 通常会限制每个节点上最大运行的 Task 的数量，通过这种方式来可避免起动太多的线程，而 SeaTunnel Engine 为了能在一个节点上运行更多的 Task，通过共享线程技术可以让那些数据量较少的 Task 共享线程，而数据量较大的 Task 独占线程，这种方式使 SeaTunnel Engine 在一个节点上运行几百上千张表同步任务成为了可能，以更少的资源占用，完成更多的表的同步。