SeaTunnel Fluss Source 连接器实战指南配置、语义与源码实现解析【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnelFluss Source 连接器用于在 SeaTunnel 批处理或流处理作业中从已有的 Fluss 为核心结合仓库中seatunnel-connectors-v2/connector-fluss的源码实现完整讲解该连接器支持的引擎、读取语义、主键表行为、全部配置项、数据类型映射与四种实战作业配置帮助你在 Spark、Flink 和 SeaTunnel Zeta 上正确、可靠地从 Fluss 表拉取数据。一、连接器概览与支持引擎Fluss Source 是 SeaTunnel 连接器体系connector-v2中的一个 Source 插件插件标识为Fluss源码见 FlussBaseOptions.java 中的CONNECTOR_IDENTITY。它支持以下执行引擎SparkFlinkSeaTunnel Zeta连接器能力矩阵对应 connector-v2-features 中的特性定义特性支持情况批处理✅流处理✅精确一次exactly-once✅列投影❌并行度✅用户自定义分片❌其中「精确一次」依赖的是 SeaTunnel 的 checkpoint 机制每个 bucket 的读取位置保存在 checkpoint 状态中作业可从上次中断处恢复配合下游 Sink 的事务能力实现端到端精确一次。二、核心读取原理Log Scanner、Bucket 分片与 RowKind2.1 读取路径与分片模型Fluss Source 通过 Fluss 客户端的log scanner读取数据而非 KV 快照。其分片split模型非常直接每个表 bucket 对应一个分片split读取并行度与 bucket 数量一致。这一点在 FlussSourceSplitEnumerator.java 的discoverSplits()中得到了印证作业启动时枚举器通过FlussAdminClient读取TableInfo拿到numBuckets与tableId随后为每个 bucket 构造一个FlussSourceSplit。分片按照「表全名 hashCode * 31 bucketId」计算归属 subTask实现跨并行度实例的负载均衡。// FlussSourceSplitEnumerator.discoverSplits() 核心逻辑简化 TableInfo tableInfo adminClient.getTableInfo(tablePath); int numBuckets tableInfo.getNumBuckets(); long tableId tableInfo.getTableId(); ListInteger buckets IntStream.range(0, numBuckets).boxed().collect(Collectors.toList());2.2 变更类型到 RowKind 的映射每条 Fluss 日志记录都带有变更类型ChangeType连接器将其映射为 SeaTunnel 的RowKind。映射逻辑位于 FlussSourceSplitReader.javaprivate static RowKind toRowKind(ChangeType changeType) { switch (changeType) { case UPDATE_BEFORE: return RowKind.UPDATE_BEFORE; case UPDATE_AFTER: return RowKind.UPDATE_AFTER; case DELETE: return RowKind.DELETE; case APPEND_ONLY: case INSERT: default: return RowKind.INSERT; } }由此引出两条读取语义日志表append-only所有记录都以INSERT的追加形式读取主键表changelog以其变更日志形式读取即INSERT/UPDATE_BEFORE/UPDATE_AFTER/DELETE四种 RowKind 都会出现下游可以据此还原增删改语义。2.3 有界性与start_mode读取是否有界由作业模式job.mode决定getBoundedness()实现在 FlussSource.javaBATCH模式Source 有界BOUNDED。每个 bucket 读取到作业启动时捕获的最新 log offset 后分片即告结束STREAMING模式Source 无界UNBOUNDED持续读取新追加的 log 记录。每个 bucket 的读取位置保存在 checkpoint 状态中作业可从中断处恢复。每个 bucket 的起始读取位点通过start_mode选择取值含义说明earliest默认从最早可用的 offset 开始读取整个 log有界 / 无界模式均适用latest只读取作业启动之后新追加的记录仅对流处理作业有意义BATCH 模式下会被直接拒绝start_modelatest在 BATCH 模式下的拒绝逻辑写在 FlussSource.java 的checkStartMode()中——它在setJobContext()阶段即抛出IllegalArgumentException属于 fail-fast 校验static void checkStartMode(JobMode jobMode, StartMode startMode) { if (JobMode.BATCH.equals(jobMode) StartMode.LATEST.equals(startMode)) { throw new IllegalArgumentException( Fluss source option start_modelatest is not supported in BATCH mode.); } }2.4 起始位点的源码级实现差异从 FlussSourceSplitEnumerator.discoverSplits() 可以看到不同模式创建分片的方式流处理 latest枚举器通过adminClient.latestOffsets(...)查询每个 bucket 当前最新 offset作为分片的起始位点流处理 earliest分片起始位点使用LogScanner.EARLIEST_OFFSET哨兵值 -2由 Fluss 服务端在每次 poll 时动态解析为当前日志起始位置因此天然不会因日志保留而过期批处理枚举器并行查询每个 bucket 的earliest与latestoffset 得到边界BucketBoundsearliest latest 的空 bucket 会被直接跳过不生成分片非空 bucket 则以[EARLIEST_OFFSET, latest)为读取范围。三、主键表语义只读 Changelog不做完整初始加载:::caution 主键表务必阅读对于主键表连接器只从最早可用的 log offset 开始读取表的 changelog而不会先读取 KV 快照。因此它能捕获持续发生的变更插入 / 更新 / 删除并带上正确的RowKind但不保证对已存在数据的完整初始加载任何因日志保留retention或压缩compaction而从 changelog 中被清除的记录都会缺失目前尚不支持主键表的「快照 增量」完整同步。如果需要完整的当前状态请优先使用日志表append-only——log scanner 总能将其完整读取。:::这意味着当你的目标是把 Fluss 主键表的存量数据完整迁移出来时需要自行评估 changelog 保留策略是否覆盖了表创建至今的全部记录否则读取结果会缺少被清理的旧记录。四、已知限制以下限制来自官方文档并已在源码中得到确认仅支持单表每个 source 只读取一张表通过databasetable配置不支持在一个 source 中读取多张表。FlussSourceConfig.getProducedCatalogTables()返回的正是单个CatalogTable见 FlussSourceConfig.java。不支持指定任意起始位点起始位置只能通过start_modeearliest或latest选择不支持从指定的 log offset 开始读取。不支持分区表将 source 指向分区 Fluss 表会在作业启动时直接失败fail-fast。校验位于 FlussSourceConfig.java读取TableInfo后若tableInfo.isPartitioned()为真立即抛出UnsupportedOperationException。请使用非分区表。主键表仅按 changelog 读取连接器读取表的 changelog 而非 KV 快照不做完整初始加载详见上文第三节。五、运行前提与依赖声明5.1 前置条件运行作业前Fluss database 和 table 必须已经存在。Source不会自动创建 Fluss database 或 table。值得注意的一点是表结构会自动从 Fluss 集群读取因此不需要配置schema选项。这在 FlussSourceConfig.java 中体现得很清楚——构造配置时即通过FlussAdminClient.getTableInfo()拉取远端表结构再调用toCatalogTable()结合FlussTypeConverter自动构建 SeaTunnel 的CatalogTable字段名、类型、长度、精度、可空性、主键全部自动生成。5.2 依赖坐标Fluss Source 连接器依赖fluss-client当前仓库使用的是0.7.0版本dependency groupIdcom.alibaba.fluss/groupId artifactIdfluss-client/artifactId version0.7.0/version /dependency六、Source 选项详解名称类型是否必填默认值描述bootstrap.serversstring是-Fluss coordinator 地址例如fluss-coordinator:9123。databasestring是-要读取的 Fluss database。tablestring是-要读取的 Fluss table。client.configmap否-传递给 Fluss 连接的额外 Fluss 客户端选项。start_modestring否earliest每个 bucket 的起始读取位点earliest整个 log或latest仅作业启动后新追加的记录。latest在BATCH模式下会被拒绝。poll.timeout.mslong否10000单次 Fluss log scanner poll 的最大阻塞时间单位毫秒。common-options-否-Source 通用选项详见 Source 通用选项。6.1 选项的源码定义以上选项在源码中的定义位置如下可供对照查阅必填三项bootstrap.servers/database/table与可选的client.config定义在 FlussBaseOptions.javapoll.timeout.ms默认值由Duration.ofSeconds(10).toMillis()得出即 10000 毫秒定义在 FlussSourceOptions.javastart_mode枚举类型可选值earliest/latest默认earliest枚举定义见 StartMode.java默认值与描述见 FlussSourceOptions.java。FlussSourceFactory通过OptionRule声明了这些约束见 FlussSourceFactory.java三个必填项都附加了notBlank校验poll.timeout.ms附加了greaterThan(..., 0L)校验——即poll.timeout.ms必须大于 0配置为 0 或负数会在作业校验阶段直接报错。6.2 client.config 的用法使用client.config传递额外的 Fluss 客户端配置例如调整请求超时client.config { request.timeout 30s }支持的配置项请参考 Fluss 客户端文档。在实现上FlussSourceConfig.buildFlussConfig() 会把bootstrap.servers与client.config中的全部键值对合并进 Fluss 的Configuration对象用于创建连接。注意SeaTunnel 侧不会校验这些键的合法性错误或拼写有误的键名会直接透传给 Fluss 客户端通常表现为客户端默认值被使用或连接时告警。6.3 Source 通用选项common-options除上述专属选项外还支持所有 Source 共用的通用选项包括plugin_output、parallelism、metadata_datasource_id等完整说明见 Source 通用选项。其中两个高频用法plugin_output为当前 Source 注册一个可被下游插件通过plugin_input直接引用的数据集临时表未指定时数据不会注册为可访问数据集。parallelism覆盖 env 中的全局并行度。由于 Fluss 的分片数等于 bucket 数通常建议并行度与 bucket 数匹配以充分利用分片。七、数据类型映射连接器在读取时会把 Fluss 数据类型转换为 SeaTunnel 数据类型。下表为官方映射关系Fluss 数据类型SeaTunnel 数据类型BOOLEANBOOLEANTINYINTTINYINTSMALLINTSMALLINTINTINTBIGINTBIGINTFLOATFLOATDOUBLEDOUBLEDECIMALDECIMALCHARSTRINGSTRINGSTRINGBINARYBYTESBYTESBYTESDATEDATETIMETIMETIMESTAMPTIMESTAMPTIMESTAMP_LTZTIMESTAMP_TZ这张映射表与源码 FlussTypeConverter.java 的toSeaTunnelType()完全一致。从实现层面还可以看到一些细节CHAR/BINARY的长度length与DECIMAL的精度、时间类型的秒精度precision/scale会通过columnLength()/columnScale()一并保留到 SeaTunnel 的物理列元数据中见 FlussTypeConverter.java值转换时toSeaTunnelValue见 FlussTypeConverter.javaCHAR/STRING由 Fluss 的BinaryString转为 JavaStringDECIMAL转为BigDecimalDATE按 epoch day 转为LocalDateTIME转为LocalTimeTIMESTAMPNTZ转为LocalDateTimeTIMESTAMP_LTZ转为 UTC 时区的OffsetDateTime如果 Fluss 表中出现上述列表之外的数据类型toSeaTunnelType会抛出unsupportedDataType异常fail-fast避免产生静默错误的数据。八、任务示例以下示例均来自官方文档并保持完整可运行。8.1 批处理读取将 Fluss 表以批模式一次性读取并输出到控制台env { parallelism 1 job.mode BATCH } source { Fluss { bootstrap.servers fluss-coordinator:9123 database fluss_db table fluss_table plugin_output fluss_source } } sink { Console { plugin_input fluss_source } }批处理模式下每个 bucket 读取到作业启动时捕获的最新 offset 后分片结束因此该作业会在数据读完后正常终止。8.2 流处理读取以latest位点持续消费新追加的数据env { parallelism 1 job.mode STREAMING } source { Fluss { bootstrap.servers fluss-coordinator:9123 database fluss_db table fluss_table start_mode latest } } sink { Console { } }start_mode latest意味着作业启动之前已存在的数据不会被读取只消费启动之后追加的记录适合「接入即开始消费」的实时场景。8.3 将一张 Fluss 表流式写入另一张 Fluss 表本示例在流处理模式下把一个 Fluss 源表复制到 Fluss 目标表Fluss 作为 Source 与 Sink 同时使用。Source 从最早可用的 log offset 开始读取连接器在每次 checkpoint 时提交每个 bucket 的读取位置作业重启后可以从断点恢复env { parallelism 1 job.mode STREAMING checkpoint.interval 30000 } source { Fluss { bootstrap.servers fluss-coordinator:9123 database fluss_stream_db table fluss_stream_src start_mode earliest poll.timeout.ms 10000 plugin_output fluss_stream } } sink { Fluss { bootstrap.servers fluss-coordinator:9123 database fluss_stream_db table fluss_stream_sink plugin_input fluss_stream } }要点说明使用plugin_output/plugin_input将 Source 的数据集显式传给 Sink注意一旦使用plugin_output下游必须使用plugin_input引用参见 Source 通用选项显式配置checkpoint.interval 30000配合读取位置随 checkpoint 提交的机制实现断点续读若目标表为 Fluss 主键表Source 侧产生的RowKind变更日志含UPDATE_BEFORE等将被完整透传实现「源到目标」的变更语义复制。8.4 为高延迟集群调优 poll 超时当 Fluss coordinator 位于高延迟网络或返回较大批次时可以增大poll.timeout.ms让 log scanner 在空轮询之间等待更长时间减少往返次数env { parallelism 2 job.mode STREAMING checkpoint.interval 60000 } source { Fluss { bootstrap.servers fluss-coordinator:9123 database fluss_db table fluss_table start_mode latest poll.timeout.ms 60000 client.config { request.timeout 30s } } }这里将poll.timeout.ms从默认 10000 提高到 60000同时通过client.config把 Fluss 客户端的request.timeout设为 30s。注意poll.timeout.ms仅影响单次 poll 的阻塞上限而request.timeout是 Fluss 客户端的 RPC 请求超时两者语义不同可按需分别调整。九、源码级深度分片发现、容错与恢复机制9.1 分片发现与并行度分配FlussSourceSplitEnumerator.java 负责分片的发现与分配流处理 latest作业启动时捕获每个 bucket 的最新 offset 作为起点此时在枚举器内完成一次 offset 查询见discoverSplits()批处理以BucketBoundsearliest/latest 边界确定读取范围空 bucket 跳过分片归属getSplitOwner()基于「表全名 bucketId」的哈希做桶分配HashUtils.bucketIndex保证同一 bucket 始终被同一 subTask 消费恢复restoreEnumerator()会把 checkpoint 中保存的FlussSourceState含未完成分片集合交还给新枚举器跳过重复发现直接继续分配未完成分片见 FlussSource.java。9.2 读取器与 offset 越界恢复FlussSourceSplitReader.java 实现了每个分片的实际拉取逻辑其中有三处值得关注的设计earliest哨兵机制新分片批 / 流的起始位点为LogScanner.EARLIEST_OFFSET-2由 Fluss 服务端在每次 poll 时动态解析为实时日志起点因此始终在有效范围内不会越界offset 越界自动重置流处理中若慢消费者落后导致已订阅 offset 被日志保留清理poll()会抛出包含LogOffsetOutOfRangeException的FetchException。读取器会查询这些 bucket 的最新 earliest把落后 bucket 重置到 earliest 继续读取行为对齐 Kafka 的auto.offset.resetearliest若重置后仍无法恢复则抛出原始异常让作业失败避免无限空转见resetTruncatedBuckets()FlussSourceSplitReader.java有界分片的完成判定批模式下当nextOffset endOffset即标记该分片完成并回报给框架同时isDrainedAtAssignment()会在分片分配阶段就识别「起点已到达终点」的空分片并直接结束见 FlussSourceSplitReaderTest.java 的单元测试对isDrainedAtAssignment(100L, 100L)/(150L, 100L)返回 true、对(50L, 100L)返回 false 的验证。9.3 精确一次的落点读取器在 checkpoint 时通过snapshotState()将未完成分片含每个 bucket 的当前读取位置持久化到状态后端见 FlussSourceSplitEnumerator.java重启后从该位置恢复这是「精确一次」能力中 Source 侧的关键保证。十、测试与进一步阅读仓库中为 Fluss 连接器提供了单元测试可作为行为契约参考FlussSourceSplitReaderTest.java验证分片分配 / 恢复时的空分片判定、split-change 类型防护等FlussSourceSplitEnumeratorTest.java验证分片发现与分配逻辑FlussSourceFactoryTest.java验证选项规则校验。连接器完整实现位于 connector-fluss 模块包含 source读取与 sink写入两侧。若需要将 Fluss 作为写入端可阅读同模块下sink/目录的实现。关于连接器通用特性批 / 流 / 精确一次 / 并行度的定义可参阅 connector-v2-featuresSource 通用选项的完整清单见 Source 通用选项。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考