SeaTunnel FTP 文件连接器connector-file-ftp能力全景与演进史从 2.2.0-beta 到 dev 的源码级解读【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnelSeaTunnelApache SeaTunnel的 FTP 文件连接器位于seatunnel-connectors-v2/connector-file/connector-file-ftp它通过自研的 Hadoop FileSystem 适配层SeaTunnelFTPFileSystem对接 Apache Commons Net 的 FTPClient为上层统一的 File Source/Sink 框架提供 FTP 协议的读写能力。本文以仓库内 FTP 连接器 Change Log 为主线结合连接器源码与配置工厂实现系统梳理该连接器从 2.2.0-beta 到 dev 版本的能力演进、连接模式与可靠性机制、多表与保存模式并给出可直接落地的完整配置示例。读完本文你将掌握 FTP 连接器的全部核心参数、底层实现原理与版本能力边界能够独立完成基于 FTP 的数据同步任务设计与排障。连接器定位File 连接器家族中的 FTP 一员FTP 连接器是 SeaTunnel 文件连接器家族LocalFile、HdfsFile、SftpFile、OssFile、S3File 等中基于HDFS 协议模拟实现的一员。从源码结构看FtpConf.java它把 FTP 服务器封装成一个ftp://host:port的 Hadoop 文件系统协议 schema 固定为ftp底层实现类为org.apache.seatunnel.connectors.seatunnel.file.ftp.system.SeaTunnelFTPFileSystem用户名、密码以fs.ftp.user.host、fs.ftp.password.host的形式写入 Hadoop Configuration。SeaTunnelFTPFileSystem继承自 HadoopFileSystem并实现StreamingFileSystem接口SeaTunnelFTPFileSystem.java在其上实现了open、create、listStatus、mkdirs、delete、rename、getFileStatus等完整文件操作因此上层的 File Source / SinkFtpFileSource、FtpFileSink只需复用 connector-file-base 的通用实现即可工作。这一设计使得 FTP 连接器天然继承了文件连接器家族的全部能力多格式、多表、分区、压缩等。与 SFTP 类似FTP 连接器不维护工作目录状态setWorkingDirectory为空实现getWorkingDirectory恒返回主目录每次文件操作都会通过connect()建立独立连接、操作结束后disconnect()释放源码注释明确说明这是为了避免每次 API 调用都承担 TCP 连接开销的权衡。核心配置参数基础连接与连接模式FTP 专属参数定义在 FtpFileBaseOptions.java 中Source 与 Sink 通过FtpFileSourceOptions、FtpFileSinkOptions直接继承。参数类型必填默认值说明hostString是-FTP 服务器地址portInteger是-FTP 服务器端口底层未显式指定时使用 FTP 标准端口 21userString是-FTP 登录用户名passwordString是-FTP 登录密码connection_modeEnum否active_localFTP 连接模式可选active_local与passive_localremote_verification_enabledBoolean否true是否开启 FTP 数据通道的远程主机校验2.3.11 新增control_encodingString否UTF-8FTP 控制连接字符编码用于支持文件路径中的特殊字符其中host、port、user、password在 Sink 侧被 FtpFileSinkFactory.java 标记为required即写 FTP 时必须显式配置Source 侧FtpFileSourceFactory.java则标记为optional因为可以结合 Catalog / 多表配置注入。connection_mode主动/被动模式与自动降级连接模式枚举定义在 FtpConnectionMode.javaactive_localFTP 主动模式服务器主动回连客户端数据端口passive_localFTP 被动模式客户端主动连接服务器数据端口更适用于客户端位于 NAT/防火墙之后的场景。值得强调的是 Change Log 2.3.9 中“Fix FTP connector connection_mode is not effective”#7865这一修复。在 SeaTunnelFTPFileSystem.connect() 中连接模式通过fs.ftp.connection.mode配置项读取而 setFsFtpConnectionMode() 实现了主动模式的自动降级逻辑若配置为active_local先enterLocalActiveMode()随后尝试创建一个测试目录/.ftptesttimestamp若创建失败捕获 IOException 并自动切换为被动模式同时将fs.ftp.connection.mode更新为passive_local无论成败finally中都会清理测试目录。也就是说即使显式配置了主动模式在网络环境不支持时连接器也会自动回退到被动模式而不是直接失败。remote_verification_enabled数据通道远程主机校验2.3.112.3.11 新增的“FTP data channels remote host verification”选项#9324对应remote_verification_enabled参数。在 connect() 中该开关被透传到 Commons Net 的FTPClient.setRemoteVerificationEnabled()。默认开启true用于校验数据连接返回的 IP 是否与控制连接一致提升安全性在部分代理/NAT 环境服务器回连地址与预期不一致下若连接异常可显式设置为false关闭校验。control_encoding控制连接编码control_encoding默认UTF-8在FTPClient连接前通过setControlEncoding()设置见 SeaTunnelFTPFileSystem.java#L144-L146用于支持文件路径中的中文等特殊字符。源码注释特别强调该设置必须在连接建立之前完成。配置解析链路FtpConf.buildWithConfig() 负责将 SeaTunnel 配置转换为 Hadoop 配置构造ftp://host:port作为 defaultFS写入fs.ftp.user.host、fs.ftp.password.host、fs.ftp.connection.mode、fs.ftp.remote.verification.enabled、fs.ftp.control.encoding等键通过setExtraOptions挂载到 HadoopConf。支持的引擎、格式与数据类型FTP 连接器复用了 File 连接器家族的完整能力矩阵适用于 Spark / Flink / SeaTunnel Zeta 三种引擎具体集成方式参见 文件连接器通用说明。文件格式能力从 FtpFileSourceFactory.optionRule() 与 FtpFileSinkFactory.optionRule() 可见file_format_type支持text、csv、json、excel、xml、markdown、pdf、binary、parquet、orc等。其中binary格式2.3.6 “Supports the transfer of any file”#6826支持以二进制块读取任意文件实现视频、图片等任意文件的同步markdown、pdf是 dev 分支最新加入的解析能力Add markdown parser #9714各格式还有专属参数例如excel的sheet_name、excel_enginePOI/EasyExcel、poi_excel_max_file_sizexml的xml_row_tag、xml_use_attr_formattext/csv的field_delimiter、row_delimiter、skip_header_row_number等。数据类型映射文件本身没有强类型定义通过配置schema显式声明每列的目标 SeaTunnel 数据类型支持 STRING、SHORT、INT、BIGINT、BOOLEAN、DOUBLE、DECIMAL、FLOAT、DATE、TIME、TIMESTAMP、BYTES、ARRAY、MAP 等类型与 SFTP Source 文档 中的数据类型映射一致。压缩支持compress_codec按格式区分text/json/csv支持lzoorc支持lzo、snappy、lz4、zlibparquet支持lzo、snappy、lz4、gzip、brotli、zstdexcel 不支持压缩。2.3.4 还加入了 LZO 读取支持2.3.8 新增archive_compress_codec支持读取归档压缩文件。版本演进主线从 2.2.0-beta 到 devChange Logdocs/zh/connectors/changelog/connector-file-ftp.md完整记录了连接器的演进轨迹按主题可归纳为以下几条主线起步FTP Sink 与 Source 诞生2.2.0-betaFTP 文件 Sink 支持#2483最初仅支持将数据写入 FTPFTP Sink 重构并新增 FTP Source#2774补齐了读端能力连接器成为完整的读写双端组件同期还改进了 parquet 读取#2841并修复了 Hive ORC 读取问题#2845。选项体系与工厂机制成型2.3.0 ~ 2.3.12.3.0 是一次结构性大版本为文件连接器引入统一的Option 与 Factory 机制#3375、重构代码结构#3238、统一文件连接器异常处理#3525、补充 Hadoop3 uber 包#3755。2.3.1 阶段继续完善将file type统一更名为file_format_type#4249、重构 schema 解析#4157、为文件读写加入压缩支持#3899、改进文件连接器 option rule 与文档#3812、增加 get source 方法#3846。格式与过滤能力扩充2.3.2 ~ 2.3.62.3.2新增 Excel Sink 与 Source#41642.3.3新增file_filter_pattern文件过滤配置#5153支持按文件名、目录以path开头正则过滤2.3.4能力密集版本——Source/Sink 增加connection_mode#6077/#6099、多表 File API 下沉到 File Base 模块#6033、支持多 Hadoop 账号#5903、引入新错误定义规则#5793、统一文件连接器选项与文档#5680、支持 LZO 读取压缩#5083、支持读取空目录#5591、schema 支持 column/primaryKey/constraintKey#5564、text/csv 格式新增enable_header_write#55672.3.5为 FTP/SFTP/LocalFile/HdfsFile 等文件连接器增加XML 文件类型支持#63272.3.6parquet 支持将 fixed/timestamp 以 int96 写入#6971、支持任意文件的二进制传输#6826。多表与保存模式2.3.8 ~ 2.3.92.3.8FTP Sink 支持多表与 save mode#7665支持读取归档压缩文件#76332.3.9FTP Source 支持多表#7795修复connection_mode不生效问题#7865text 读取支持null_format自定义空值格式#8109(S)FTP 创建目录补充 debug 日志#8286。文件管理与可靠性增强2.3.10 ~ 2.3.122.3.10FTP 连接器目录操作可靠性修复#8959、新增filename_extension读写参数#8769、重构 connector common options#8634、Sink 支持无数据时创建空文件#8543、Sink 支持单文件模式#85182.3.11新增 FTP 数据通道远程主机校验选项remote_verification_enabled#9324、更新文件连接器配置#9034、text Sink 增加row_delimiter#90172.3.12text 文件处理支持自定义行分隔符#9608。dev 分支最新动态当前 dev 分支正在加入Markdown 解析器#9714用于文本类文件的 Markdown 解析与 RAG 元数据处理markdown_rag_metadata_enabled参数已出现在 Source option rule 中。多表Multiple Table与表管理能力从 2.3.8Sink与 2.3.9Source开始FTP 连接器支持多表配置。Source 侧table_configs与单表path互斥exclusive规则见 FtpFileSourceFactory.java#L58多表配置通过MultipleTableFTPFileSourceConfig解析每个表可独立指定path、schema与file_format_type。此外连接器还提供 FtpFileCatalog.java 与对应 Factory支持以 Catalog 方式管理 FTP 文件表元数据配合schema_save_mode、data_save_modeSink 侧实现表结构/数据的安全落库策略。实战配置示例以下示例均以当前仓库为基准编写可直接套用。示例一FTP → Consoletext 格式读取env { parallelism 2 job.mode BATCH } source { FTP { host 192.168.1.100 port 21 user seatunnel password your_password path /data/input file_format_type text field_delimiter \001 row_delimiter \n skip_header_row_number 1 connection_mode passive_local # 读取 FTP 后删除源文件 post_sync_action delete schema { fields { id INT name STRING ts TIMESTAMP } } } } sink { Console { parallelism 1 } }示例二多表 Source2.3.9source { FTP { host 192.168.1.100 port 21 user seatunnel password your_password table_configs [ { table_path /data/orders table_name orders file_format_type csv schema { fields { order_id BIGINT, amount DOUBLE } } }, { table_path /data/users table_name users file_format_type json schema { fields { user_id BIGINT, name STRING } } } ] } }示例三FTP SinkCSV 分区 单文件模式sink { FTP { host 192.168.1.100 port 21 user seatunnel password your_password path /data/output file_format_type csv field_delimiter , row_delimiter \n enable_header_write true # 分区能力 have_partition true partition_by [dt] partition_dir_expression ${dt} is_partition_field_write_in_file true # 单文件模式2.3.10所有数据写入一个文件 single_file_mode true # 无数据时也创建空文件2.3.10 create_empty_file_when_no_data true filename_extension .csv schema_save_mode CREATE_SCHEMA_WHEN_NOT_EXIST data_save_mode APPEND_DATA } }示例四任意文件同步binary 格式2.3.6source { FTP { host 192.168.1.100 port 21 user seatunnel password your_password path /data/images file_format_type binary binary_chunk_size 4096 } } sink { LocalFile { path /tmp/images_backup file_format_type binary } }环境依赖说明与 SFTP 文档 中说明一致若使用 Spark/Flink 引擎需确保集群已集成 Hadoop实测版本 2.x连接器要求Hadoop 2.9.X若使用 SeaTunnel Engine安装包已自动集成 Hadoop jar可在${SEATUNNEL_HOME}/lib下确认。连接器通过 HDFS 协议模拟访问 FTP因此这类依赖是必需的。测试保障与工程规范仓库为 FTP 连接器提供了对应的单元测试与工厂测试SeaTunnelFTPFileSystemTest.java验证 FTP 文件系统的核心操作逻辑FtpFileFactoryTest.java验证 Source/Sink Factory 的 option rule 与插件发现机制。Change Log 中的 2.3.1 条目还记录了 spotless 代码格式化、模块可读命名#4114等工程化改进2.3.9 的 metrics 关联改进#7786使指标信息可关联到逻辑计划节点属于 SeaTunnel 引擎层通用能力。总结SeaTunnel FTP 文件连接器通过“Hadoop FileSystem 适配层 文件连接器家族通用框架”的架构用较小成本实现了与 LocalFile、HdfsFile 等一致的丰富能力十余种文件格式、多表、save mode、压缩、分区、文件过滤、任意文件二进制传输以及主动/被动连接模式与自动降级、远程主机校验、控制连接编码等 FTP 专属可靠性机制。从 2.2.0-beta 的 Sink 起步到 2.3.x 的格式扩充、多表支持与工程重构再到 dev 的 Markdown/PDF 解析探索其演进主线清晰可循。读者可按需组合本文章节中的配置示例快速落地 FTP 数据集成任务更深入的实现细节可直接阅读 connector-file-ftp 源码目录。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考