SeaTunnel Connector V2 特性详解多引擎适配、批量流统一与 Source/Sink 核心能力体系【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文围绕 docs/en/introduction/concepts/connector-v2-features.md 展开系统讲解 SeaTunnel Connector V2 的诞生背景、与 V1 的本质差异以及 Source/Sink 连接器各自的核心能力exactly-once、column projection、batch/stream、parallelism、multimodal、CDC、多表读写等并结合仓库源码SeaTunnelSource、SeaTunnelSink、translation 层等揭示这些特性背后的实现原理帮助开发者在选型、配置和自研连接器时建立完整的能力认知。一、Connector V2 是什么从 V1 到 V2 的演进SeaTunnel 在 issue #1608 之后正式引入了 Connector V2 体系。Connector V2 是基于 SeaTunnel Connector API 接口定义的连接器规范与 Connector V1 相比它在架构上实现了多项关键突破。理解这些差异是理解整个 Connector V2 特性体系的前提。1.1 多引擎支持Multi Engine SupportSeaTunnel Connector API 是一套**与底层执行引擎无关engine independent**的 API。基于该 API 开发的连接器可以运行在多个引擎之上。从当前仓库结构看SeaTunnel 官方维护了 Flink 与 Spark 两套翻译层实现seatunnel-translation/seatunnel-translation-flink/含 Flink 13/15/20 三个版本seatunnel-translation/seatunnel-translation-spark/含 Spark 2.4/3.3 两个版本seatunnel-translation/seatunnel-translation-base/存放与引擎无关的翻译基础组件翻译层的核心作用是把 SeaTunnel 连接器 API 的语义翻译成具体引擎的原生算子。以 Source 为例BaseSourceFunction 定义了引擎无关的open()、run(CollectorT)、snapshotState(long)三个核心方法而 ParallelSource 和 CoordinatedSource 则分别对应两种分布式读取模式详见下文 parallelism 小节。1.2 多引擎版本支持Multi Engine Version Support在 V1 时代连接器与引擎版本强耦合底层引擎升级大版本往往意味着大量连接器需要修改代码重新适配。Connector V2 通过translation layer翻译层将连接器与引擎解耦连接器开发者只需要面向 SeaTunnel Connector API 编写一次新引擎或新版本引擎的适配工作收敛到翻译层这一个地方完成。这正是seatunnel-translation目录下按引擎、按大版本拆分子模块如seatunnel-translation-flink-13、seatunnel-translation-spark-3.3的原因——每个引擎版本的翻译实现独立演进互不干扰。1.3 统一的批量与流式处理Unified Batch And StreamConnector V2 允许同一个连接器同时支持批量batch处理和流式stream处理无需为两种模式分别开发连接器。其底层支撑是 Boundedness 枚举public enum Boundedness { /** A BOUNDED stream is a stream with finite records. */ BOUNDED, /** A UNBOUNDED stream is a stream with infinite records. */ UNBOUNDED }连接器通过SeaTunnelSource#getBoundedness()声明自身的数据边界性质引擎据此决定作业的运行方式。以 Kafka 连接器为例KafkaSource 会根据配置动态返回BOUNDED或UNBOUNDED当配置了起始/结束位点等有界读取条件时按批量作业运行否则按无限流作业持续运行。1.4 JDBC/日志连接复用Multiplexing JDBC/Log ConnectionConnector V2 支持JDBC 资源复用多个读取任务共享同一批数据库连接资源以及共享数据库日志解析多个表共享同一份变更日志解析能力。这一能力在多表同步场景下显著降低了对数据库连接数的占用和对源端日志的重复消费。1.5 多模态数据集成Multimodal Data IntegrationConnector V2 支持多模态multimodal数据集成覆盖结构化数据、非结构化文本、视频、图片、二进制文件等多种数据形态。这一能力同时出现在 Source 与 Sink 的公共特性列表中见下文。二、Source 连接器核心特性Source 连接器共享一组核心特性不同连接器对这些特性的支持程度各不相同。下面逐项说明并给出仓库中的实现佐证。2.1 exactly-once精确一次读取定义如果数据源中的每条数据只会被 Source 向下游发送一次则认为该 Source 连接器支持 exactly-once。SeaTunnel 的实现思路在 checkpoint 时将已读取的Split及其offset数据在该 Split 中的读取位置如行号、字节偏移量等保存为StateSnapshot。任务重启后从最后一次StateSnapshot恢复定位上次读取到的 Split 与 offset从断点继续向下游发送数据从而避免重复读取。这一机制在 API 层有清晰的落点SourceSplitEnumerator#snapshotState(long checkpointId) 负责在 checkpoint 时快照哪些 Split 已分配/未分配的枚举器状态SourceReader#snapshotState(long checkpointId) 返回当前 reader 持有的 Split 列表含各自 offset 状态用于故障后恢复SeaTunnelSource#restoreEnumerator(...)与SeaTunnelSource#getEnumeratorStateSerializer()分别负责从 checkpoint 状态重建枚举器、以及状态的跨进程序列化传输。支持该特性的典型连接器包括File、Kafka等。2.2 column projection列裁剪定义连接器支持只从数据源读取指定列。需要注意的是先读取全部列、再用 schema 过滤掉不需要的列并不算真正的 column projection——真正的列裁剪应发生在读取阶段从而减少数据搬运量与反序列化开销。典型对比JDBCSource可以通过 SQL 精确指定读取列select col1, col2 from ...属于真正的 column projectionKafkaSource会读取 topic 中的全部内容再通过schema过滤不需要的列这不是column projection。2.3 batch批量模式定义批量作业模式下读取的数据是**有界bounded**的作业在完成全部数据读取后自动停止。在 API 层面批量 Source 的getBoundedness()返回Boundedness.BOUNDED可参考KafkaSource对 getBoundedness() 的按配置动态返回实现。2.4 stream流式模式定义流式作业模式下读取的数据是**无界unbounded**的作业持续运行不停止。对应的getBoundedness()返回Boundedness.UNBOUNDED。Boundedness枚举的注释也明确指出batch 模式对应BOUNDEDstreaming 模式对应UNBOUNDED。2.5 parallelism并行度定义支持parallelism配置的 Source 称为并行 Source每个并行度都会创建一个 task 来读取数据。实现机制在并行 Source 中数据源会被切分成多个Split然后由SplitEnumerator枚举器将 Split 分配给各个SourceReader处理。从源码可以看到这一分工的完整链路SourceSplitEnumerator 运行在 master 端负责枚举 Split 并管理分配。其核心方法包括run()启动时枚举 Split、registerReader(int subtaskId)注册 reader、handleSplitRequest(int subtaskId)响应 reader 的取数请求、addSplitsBack(ListSplitT, int subtaskId)reader 故障时把未完成 checkpoint 的 Split 收回重新分配、signalNoMoreSplits(int)通知 reader 不再有新 Split。SourceReader 运行在 worker 端通过pollNext(CollectorT)持续产出数据其Context#sendSplitRequest()可以向枚举器主动请求新的 Split。在翻译层ParallelSourceCoordinatedSource的并行变体与CoordinatedSource分别承载reader 自行拉取 Split与枚举器统一协调分配两种执行模式最终都会被翻译为具体引擎Flink/Spark的并行 Source 算子。2.6 multimodal多模态Source 支持多模态数据集成包括结构化与非结构化文本、视频、图片、二进制文件等数据形态。2.7 support user-defined split用户自定义 Split支持用户自定义 Split 规则。这意味着数据切分方式可以由用户按需配置而不是由连接器固定死为倾斜数据的均衡分配提供了灵活性。2.8 support multiple table read多表读取支持在一个 SeaTunnel 作业中同时读取多张表。这一能力与 Sink 侧的多表写入配合构成了多表同步multi-table sync的完整闭环。三、Sink 连接器核心特性Sink 连接器同样共享一组核心特性。在阅读本节前建议先了解 Sink 侧的职责划分SeaTunnelSink 接口通过createWriter()/createCommitter()/createAggregatedCommitter()把写数据与提交commit两个阶段解耦分别对应 SinkWriter、SinkCommitter、SinkAggregatedCommitter这是理解 Sink exactly-once 的关键。3.1 exactly-once精确一次写入定义当任意一条数据流入分布式系统时如果系统在整个处理过程中只精确处理一次且结果正确则认为该系统满足精确一次一致性。对 Sink 连接器而言支持 exactly-once 意味着任何一条数据只会被写入目标端一次。一般有两种实现路径路径一目标数据库支持主键去重利用目标端的主键约束天然去重。例如MySQL、Kudu等数据库即使重复写入主键冲突也会被幂等处理。路径二目标端支持 XA 事务 两阶段提交XA 事务可以跨会话使用即使创建事务的程序已结束新启动的程序只需要知道上一个事务的 ID即可重新提交或回滚该事务。在此基础上利用**两阶段提交Two-phase Commit**保证 exactly-once。典型实现如File、MySQL连接器。在 API 层面这一路径的落点是SinkWriter与 committer 的配合SinkWriter#prepareCommit(long checkpointId) 在snapshotState之前调用如需使用 2PC可在该方法中返回 commit info如 XA 事务 ID随后交给SinkCommitter#commit(...)处理SinkWriter#abortPrepare()用于在prepareCommit失败时回滚其副作用当前仅在 Spark 引擎中调用SeaTunnelSink通过createCommitter()/createAggregatedCommitter()提供单点提交与聚合提交两种粒度getCommitInfoSerializer()/getAggregatedCommitInfoSerializer()则保证 commit info 可以跨进程传输。3.2 cdc变更数据捕获定义如果 Sink 连接器支持基于主键写入多种行类型row kinds——INSERT/UPDATE_BEFORE/UPDATE_AFTER/DELETE——则认为该连接器支持 CDC变更数据捕获。这一能力使 CDC 管道如 MySQL CDC → Sink中的增删改语义能够被完整透传而不仅仅是追加写入。3.3 support multiple table write多表写入定义支持在一个 SeaTunnel 作业中写入多张表用户可以通过**配置占位符placeholders**动态指定目标表的标识符。占位符机制详见仓库文档 Sink Options Placeholders原文档中的../configuration/sink-options-placeholders.md链接其仓库根路径为docs/en/introduction/configuration/sink-options-placeholders.md其核心内容如下支持的占位符表达式用于获取上游 CatalogTable 的元数据占位符含义${database_name}上游表的 database可带默认值${database_name:default_my_db}${schema_name}上游表的 schema可带默认值${schema_name:default_my_schema}${table_name}上游表的 table可带默认值${table_name:default_my_table}${schema_full_name}上游表的 schema 全路径database schema${table_full_name}上游表的全路径database schema table${primary_key}上游表的主键字段${unique_key}上游表的唯一键字段${field_names}上游表的字段名列表${comment}上游表的注释${partition_keys}上游表的分区键使用前提所使用的 Sink 连接器必须实现了TableSinkFactoryAPI。以 JDBC 为例JdbcSinkFactory 即实现了TableSinkFactory并承载多表写入能力的创建逻辑。配置示例MySQL CDC → JDBC 多表同步env { // ignore... } source { MySQL-CDC { // ignore... } } transform { // ignore... } sink { jdbc { url jdbc:mysql://localhost:3306 driver com.mysql.cj.jdbc.Driver user root password 123456 database ${database_name}_test table ${table_name}_test primary_keys [${primary_key}] } }配置示例Oracle CDC → JDBC注意 schema 占位符sink { jdbc { url jdbc:mysql://localhost:3306 driver com.mysql.cj.jdbc.Driver user root password 123456 database ${schema_name}_test table ${table_name}_test primary_keys [${primary_key}] } }主键占位符的展开行为${primary_key}会展开为上游表元数据中定义的全部主键列。例如上游主键为(f1, f2)则primary_keys [${primary_key}]会被展开为primary_keys [f1, f2]不支持将${primary_key}与静态列名混用例如primary_keys [${primary_key}, tenant_id]是不支持的——只有当${primary_key}是列表中的唯一元素时占位符替换才会作用于列表值该行为对单表作业与多表作业一致占位符替换会在连接器启动前完成确保 Sink 配置就绪后再使用如果变量未被替换通常意味着上游表元数据缺少对应选项例如mysql源不提供${schema_name}、oracle源不提供${database_name}等。3.4 multimodal多模态Sink 支持多模态数据集成包括结构化与非结构化文本、视频、图片、二进制文件等数据形态。四、如何在项目中查阅与验证这些特性API 定义Source 侧完整契约见 SeaTunnelSourceSink 侧见 SeaTunnelSink配套的Boundedness、SourceSplitEnumerator、SourceReader、SinkWriter等接口均位于seatunnel-api/src/main/java/org/apache/seatunnel/api/下。翻译层实现seatunnel-translation/目录下按引擎与版本组织的子模块Flink 13/15/20、Spark 2.4/3.3是多引擎版本支持特性的直接体现。连接器实例可在 seatunnel-connectors-v2 中逐个查看各连接器的特性支持情况。例如connector-kafka的KafkaSource展示了Boundedness的动态判定connector-jdbc的JdbcSinkFactory展示了TableSinkFactory与多表写入的实现。多表占位符完整语法与边界行为见 docs/en/introduction/configuration/sink-options-placeholders.md。五、小结Connector V2 通过引擎无关的 API 与翻译层设计将连接器与 Flink/Spark 及其大版本解耦通过Boundedness实现了批量/流式统一通过 Split-Enumerator 模型统一了并行读取与断点恢复通过 Writer-Committer 分离与两阶段提交实现了 Sink 端 exactly-once并通过占位符机制把多表读写变成了可配置、可动态化的能力。无论是使用现成连接器做数据集成还是基于 Connector API 自研插件上述特性清单都是能力评估与架构设计的重要参考。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
