数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载FtpFile Sink 是 Apache SeaTunnel 内置的文件型输出连接器负责把上游 Source 或 Transform 产出的数据以 text、csv、parquet、orc、json、excel、xml、binary 等格式写入 FTP 服务器指定目录。本文以官方文档 docs/en/connector-v2/sink/FtpFile.md 为主体结合仓库中connector-file-ftp与connector-file-base模块源码完整讲解该连接器的全部配置参数、运行环境要求、分区与事务机制、完整配置示例以及 2PC 提交与 FTP 文件系统适配的底层实现。读完本文你将能够独立完成一个从任意 Source 到 FTP 的 SeaTunnel 作业的配置、调优与排障。连接器简介与适用场景FtpFile 连接器的职责非常聚焦Output data to Ftp——将 SeaTunnel 管道中的SeaTunnelRow数据流式写入 FTP 服务器。它与同模块的 FtpFileSource 形成一对读写组合适用于数据落地 FTP、与外部系统通过 FTP 交换文件的场景。从源码看连接器插件名来自 FileSystemType.FTPSink 侧实现为 FtpFileSink它继承自文件连接器公共基类BaseFileSink因此所有文件型 Sink 的通用能力分区、事务、格式、压缩它都天然具备。运行环境前提官方文档对运行环境有明确提示务必先满足使用 Spark / Flink 运行时集群必须已经集成 Hadoop官方实测的 Hadoop 版本为 2.x。如果集群未集成 HadoopFTP 文件系统适配层将无法加载作业会启动失败。使用 SeaTunnel Engine 运行时无需额外操作下载并安装 SeaTunnel Engine 时已自动集成 Hadoop 相关 jar。可通过检查${SEATUNNEL_HOME}/lib目录下的 jar 包来确认。这一前提在源码中也得到印证FtpFileSink 通过 Hadoop 配置体系来桥接 FTP 协议见下文“FTP 文件系统适配”本质上仍依赖 Hadoop FileSystem 抽象。核心特性Key features特性说明exactly-once默认通过两阶段提交2PC保证exactly-once详见 connector-v2-features文件格式text、csv、parquet、orc、json、excel、xml、binary 共 8 种均完整支持其中 exactly-once 承诺的落地实现可追溯到 BaseFileSink它实现了SeaTunnelSinkSeaTunnelRow, FileSinkState, FileCommitInfo, FileAggregatedCommitInfo泛型签名分别提供了createWriter、restoreWriter从 checkpoint 状态恢复写以及createAggregatedCommitter返回FileSinkAggregatedCommitter这正是 2PC 中预提交与最终提交的载体详见下文“源码级原理”。完整 Options 参数表下表完整列出官方文档定义的连接器参数。标注 “Required yes” 的 6 个参数host、port、user、password、path、tmp_path在配置缺失时连接器会直接抛出配置校验异常。名称类型是否必填默认值说明hoststringyes-FTP 服务器地址portintyes-FTP 服务器端口userstringyes-FTP 用户名passwordstringyes-FTP 密码pathstringyes-目标目录路径tmp_pathstringyes/tmp/seatunnel结果文件先写入临时目录再通过mv提交到目标目录必须是 FTP 目录connection_modestringnoactive_localFTP 连接模式custom_filenamebooleannofalse是否自定义文件名file_name_expressionstringno${transactionId}仅当 custom_filename 为 true 时生效filename_time_formatstringnoyyyy.MM.dd仅当 custom_filename 为 true 时生效file_format_typestringnocsv输出文件格式field_delimiterstringno\001仅 text 格式生效列分隔符row_delimiterstringno\n仅 text 格式生效行分隔符have_partitionbooleannofalse是否开启分区处理partition_byarrayno-仅 have_partition 为 true 时生效按所选字段分区partition_dir_expressionstringno${k0}${v0}/${k1}${v1}/.../${kn}${vn}/仅 have_partition 为 true 时生效is_partition_field_write_in_filebooleannofalse仅 have_partition 为 true 时生效sink_columnsarrayno空为空时写入全部字段顺序决定文件实际写入顺序is_enable_transactionbooleannotrue是否开启事务保护batch_sizeintno1000000单个文件最大行数compress_codecstringnonone压缩编解码器common-optionsobjectno-Sink 公共参数max_rows_in_memoryintno-仅 excel 格式生效内存中缓存的最大数据条数sheet_namestringnoSheet${随机数}仅 excel 格式生效工作簿工作表名xml_root_tagstringnoRECORDS仅 xml 格式生效XML 根元素标签xml_row_tagstringnoRECORD仅 xml 格式生效XML 数据行标签xml_use_attr_formatbooleanno-仅 xml 格式生效是否使用标签属性格式parquet_avro_write_timestamp_as_int96booleannofalse仅 parquet 格式生效parquet_avro_write_fixed_as_int96arrayno-仅 parquet 格式生效encodingstringnoUTF-8仅 json、text、csv、xml 格式生效上表默认值均可在 BaseSinkConfig 中逐一找到对应常量如DEFAULT_TMP_PATH /tmp/seatunnel、DEFAULT_FILE_NAME_EXPRESSION ${transactionId}、DEFAULT_BATCH_SIZE 1000000参数定义与文档完全一致可作为排障时的权威依据。连接参数host / port / user / password / path / tmp_path / connection_mode连接五要素与目标路径host[string]FTP 服务器主机地址必填。port[int]FTP 服务端口必填通常为 21。user[string]FTP 登录用户名必填。password[string]FTP 登录密码必填。path[string]目标写入目录必填。最终文件会生成在该目录下。tmp_path[string]临时目录必填默认/tmp/seatunnel。写入机制是结果文件先写进临时目录完成后用mv命令整体提交到目标目录这保证了目标目录不会出现半成品文件。由于mv发生在 FTP 服务器侧tmp_path必须是一个 FTP 可达的目录且建议与path位于同一文件系统/挂载范围内。在 FtpFileSink.java 的prepare阶段会通过CheckConfigUtil.checkAllExists强制校验 host、port、user、password 四项path 的校验由工厂的OptionRule完成任一缺失都会抛出FileConnectorException错误码CONFIG_VALIDATION_FAILED。connection_mode主动/被动连接模式默认值active_local可选值active_local、passive_local。对应枚举定义在 FtpConnectionMode其模式字符串与 Apache Commons Net 的连接模式一致active_localFTP 主动模式服务器主动连接客户端的数据端口passive_localFTP 被动模式客户端主动连接服务器数据端口。生产环境中客户端位于 NAT/防火墙之后时通常需要切换为passive_local否则数据通道可能建立失败。FTP 文件系统适配源码级FtpFile 连接器通过 Hadoop FileSystem 抽象访问 FTP 协议这一桥接逻辑在 FtpConf 中实现文件系统实现类org.apache.seatunnel.connectors.seatunnel.file.ftp.system.SeaTunnelFTPFileSystem协议 schemeftp默认文件系统地址defaultFS由配置拼接为ftp://{host}:{port}认证与连接参数被映射为 Hadoop 配置项写入hadoopConffs.ftp.user.{host}→ FTP 用户名fs.ftp.password.{host}→ FTP 密码fs.ftp.connection.mode→ 连接模式仅在配置了connection_mode时写入。也就是说只要让BaseFileSinkWriter拿到这份HadoopConf写数据、建目录、mv提交等所有文件操作都复用了文件连接器公共模块的通用实现FTP 特有逻辑被收敛在配置映射这一层。这也是“先临时目录、后 mv 提交”机制能够透明工作的原因。文件名与时间格式控制custom_filename [boolean]是否自定义文件名默认false。file_name_expression [string]仅当custom_filename true时生效。描述要写入path的文件名表达式支持变量${now}与${uuid}例如test_${uuid}_${now}${now}表示当前时间其格式由filename_time_format定义需要注意若is_enable_transaction true会自动在文件名头部追加${transactionId}_前缀用于事务标识例如transactionId_20250101_xxxx这类形态。默认表达式为${transactionId}见 BaseSinkConfig 的DEFAULT_FILE_NAME_EXPRESSION。filename_time_format [string]仅当custom_filename true时生效。当file_name_expression中包含xxxx-${now}时该参数指定时间格式默认yyyy.MM.dd。常用符号如下符号含义y年YearM月Monthd日Day of monthH小时Hour in day 0-23m分钟Minute in hours秒Second in minute输出文件格式与行列分隔file_format_type [string]支持的格式text、csv、parquet、orc、json、excel、xml、binary。默认值为csv。请特别注意最终文件名会以格式后缀结尾其中 text 格式的后缀是txt即file_format_type text时生成xxx.txt文件。field_delimiter [string]行内列之间的分隔符仅 text 格式需要。默认\001SOH 控制符即\u0001这是 Hadoop 生态 text 文件常见的默认列分隔。row_delimiter [string]文件内行与行之间的分隔符仅 text 格式需要默认\n。历史上2.3.0-beta 之前存在无法从配置文件解析\t作为分隔符的缺陷相关修复可见文档 Changelog 部分配置时建议直接书写真实转义字符。encoding [string]仅 json、text、csv、xml 四种格式生效。写入文件的编码默认UTF-8该参数最终通过Charset.forName(encoding)解析。压缩配置compress_codeccompress_codec指定输出文件的压缩编解码器默认none。不同格式支持的压缩算法不同务必按下表对照使用excel 格式不支持任何压缩文件格式支持压缩算法txtlzo、nonejsonlzo、nonecsvlzo、noneorclzo、snappy、lz4、zlib、noneparquetlzo、snappy、lz4、gzip、brotli、zstd、none源码 BaseSinkConfig 中以TXT_COMPRESS、PARQUET_COMPRESS、ORC_COMPRESS三个单选项分别约束了 txt/json/csv、parquet、orc 的可选压缩集与文档表格完全一致同时 FtpFileSinkFactory 的OptionRule也按file_format_type做了条件绑定配置不合法时工厂校验即会报错。分区写入have_partition [boolean]是否开启分区处理默认false。partition_by [array]仅当have_partition true时生效。按所选字段进行分区支持多字段组合。partition_dir_expression [string]仅当have_partition true时生效。指定partition_by后连接器会根据分区信息生成对应的分区目录最终文件落在分区目录内。默认表达式为${k0}${v0}/${k1}${v1}/.../${kn}${vn}/其中k0是第一个分区字段名v0是第一个分区字段的值以此类推。例如按age分区时典型结果是age25/这样的目录层级。is_partition_field_write_in_file [boolean]仅当have_partition true时生效。若为true分区字段及其值也会写入数据文件内部若为false分区字段只体现在目录结构中。官方文档特别举例如果要产出 Hive 数据文件该值应设为falseHive 分区约定分区列不在数据文件内重复出现。写入列与事务保护sink_columns [array]需要写入文件的列默认值为 Source 或 Transform 输出的全部列。数组中字段的顺序即文件实际写入顺序可用于调整列序或裁剪列。is_enable_transaction [boolean]默认true。开启后连接器保证写入目标目录的数据不丢失、不重复。开启时文件名头部会自动追加${transactionId}_前缀。官方文档明确当前只支持true。batch_size [int]单个文件的最大行数默认1000000。在 SeaTunnel Engine 中一个文件的实际行数由batch_size与checkpoint.interval共同决定若checkpoint.interval足够大sink writer 会持续写入当前文件直到行数超过batch_size才切换新文件若checkpoint.interval较小则每次 checkpoint 触发时sink writer 就会创建一个新文件以 checkpoint 边界切分文件便于与事务提交对齐。因此想要控制文件大小应同时调大checkpoint.interval并设置合适的batch_size。Excel / XML / Parquet 专属参数max_rows_in_memory [int]仅 excel 格式生效。内存中可缓存的最大数据条数用于控制写 Excel 时的内存占用。sheet_name [string]仅 excel 格式生效。写入的工作簿工作表名默认值为Sheet${随机数}。xml_root_tag [string]仅 xml 格式生效。指定 XML 文件的根元素标签名默认RECORDS。xml_row_tag [string]仅 xml 格式生效。指定 XML 文件中数据行的标签名默认RECORD。xml_use_attr_format [boolean]仅 xml 格式生效。是否使用标签属性格式处理数据。parquet_avro_write_timestamp_as_int96 [boolean]仅 parquet 格式生效默认false。支持将 timestamp 类型写为 Parquet INT96。parquet_avro_write_fixed_as_int96 [array]仅 parquet 格式生效。支持将 12 字节字段写为 Parquet INT96。完整配置示例示例一text 格式的基础配置最简单的 text 落地配置官方文档原始示例FtpFile { host xxx.xxx.xxx.xxx port 21 user username password password path /data/ftp file_format_type text field_delimiter \t row_delimiter \n sink_columns [name,age] }该配置将上游数据以name、age两列顺序写入/data/ftp下的*.txt文件列间以\t分隔行间以\n分隔未开启自定义文件名默认使用${transactionId}表达式并带事务前缀未开启分区。示例二分区 自定义文件名 列裁剪的组合配置同时启用have_partition、custom_filename与sink_columns的完整配置官方文档原始示例FtpFile { host xxx.xxx.xxx.xxx port 21 user username password password path /data/ftp/seatunnel/job1 tmp_path /data/ftp/seatunnel/tmp file_format_type text field_delimiter \t row_delimiter \n have_partition true partition_by [age] partition_dir_expression ${k0}${v0} is_partition_field_write_in_file true custom_filename true file_name_expression ${transactionId}_${now} sink_columns [name,age] filename_time_format yyyy.MM.dd }行为推演结合前文参数语义数据先写入临时目录/data/ftp/seatunnel/tmp提交时mv到目标目录/data/ftp/seatunnel/job1按age字段分区分区目录表达式为${k0}${v0}例如age25/is_partition_field_write_in_file true分区列age也会写入数据文件文件名形如${transactionId}_2025.01.01.txt${now}按yyyy.MM.dd格式化text 后缀为 txt仅写入name、age两列顺序按数组声明。源码级工作原理与验证Sink 主类与配置校验FtpFileSink 通过AutoService(SeaTunnelSink.class)注册插件名为FileSystemType.FTP对应的字符串。其prepare方法强制校验 host/port/user/password随后调用FtpConf.buildWithConfig构造 Hadoop 配置其余逻辑全部继承自 BaseFileSink。2PC exactly-once 的实现路径从 BaseFileSink 的接口实现可以看出完整的提交链路createWriter/restoreWriter创建或恢复BaseFileSinkWriter写入发生在临时路径createAggregatedCommitter返回FileSinkAggregatedCommitter在 checkpoint 提交阶段统一把各 writer 的临时文件mv到目标目录三个序列化器FileCommitInfo、FileAggregatedCommitInfo、FileSinkState保证 checkpoint 状态可持久化、可恢复。这套“写临时文件 → checkpoint 预提交 → 聚合提交mv 到目标目录”的机制正是“数据不丢失、不重复”与文件名带transactionId前缀的原因。文档中的“Next version”Changelog 提到修复了“从 states 恢复 writer 时获取 transaction 直接失败”的问题正是这条链路中的状态恢复环节。工厂参数约束与单元测试FtpFileSinkFactory 通过OptionRule声明了参数依赖关系path、host、port、user、password为必填field_delimiter、row_delimiter、压缩等参数仅在对应file_format_type下生效file_name_expression、filename_time_format仅在custom_filename true时生效partition_by等仅在have_partition true时生效。仓库中的单元测试 FtpFileFactoryTest 对 Sink 与 Source 工厂的optionRule()均做了非空断言确保参数规则可用。Sink 与 Source 的能力差异注意同一 FTP 模块下的 FtpFileSource 在读取侧有明确限制Ftp 文件 Source 仅支持读取 text、csv、json 三种格式配置为 orc、parquet 会直接抛出ILLEGAL_ARGUMENT异常而本篇文章所讲的 FtpFileSink则支持 8 种格式的写入。在规划“FTP 读 → FTP 写”的自同步任务时需注意两侧格式能力并不对称。版本演进 Changelog2.2.0-beta2022-09-26新增 Ftp File Sink 连接器。2.3.0-beta2022-10-20三处缺陷修复——Windows 环境下路径不正确PR 2980文件系统获取报错PR 3117配置文件无法解析\t作为分隔符PR 3083。Next version缺陷修复PR 3258上游字段为 null 时抛 NullPointerExceptionsink 列映射失败从 states 恢复 writer 时直接获取 transaction 失败能力增强支持为每个文件设置 batch_sizePR 3625支持文件压缩PR 3899。常见问题速查现象排查方向作业启动报配置校验失败检查 host、port、user、password、path 是否齐全对应FtpFileSink.prepare与OptionRule的必填声明Spark/Flink 下无法写入确认集群已集成 Hadoop 2.xFTP 文件系统适配依赖 Hadoop FileSystem 抽象数据通道建立失败客户端处于 NAT/防火墙后时将connection_mode改为passive_local文件一直不落到目标目录确认tmp_path为 FTP 可达目录mv提交依赖该目录可写同时理解文件切分由batch_size与checkpoint.interval共同决定目标目录出现未知前缀文件名is_enable_transaction true时文件名自动带${transactionId}_前缀属预期行为写 Hive 数据文件列重复分区写入时确认is_partition_field_write_in_file false压缩参数报错对照“压缩配置”表格excel 不支持压缩orc/parquet 各有允许的算法白名单更多 Sink 公共参数如date_format、datetime_format、time_format、表头写入等可查阅 Sink Common OptionsFtpFile 专属配置之外的能力均可复用公共参数体系。赞分享数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载相关推荐SeaTunnel FtpFile Sink 连接器实战指南将数据可靠写入 FTP/SFTPSeaTunnel FtpFile Sink 连接器实战指南将数据可靠写入 FTP/SFTP 本文是 SeaTunnel 文件类 Sink 连接器系列中的 F数据集成ETL大数据批处理流处理变更数据捕获SeaTunnel FtpFile Sink Connector 使用指南将数据输出到 FTP 服务器SeaTunnel FtpFile Sink Connector 使用指南将数据输出到 FTP 服务器 SeaTunnel 的 FtpFile Sink 插件数据集成ETL大数据批处理流处理变更数据捕获Apache SeaTunnel SLS Sink 连接器将 SeaTunnel 行数据写入阿里云日志服务的实现与配置指南Apache SeaTunnel SLS Sink 连接器将 SeaTunnel 行数据写入阿里云日志服务的实现与配置指南 本文基于 SeaTunnel 官方数据集成ETL大数据批处理流处理变更数据捕获上一篇Isar数据库最佳实践构建高性能Flutter应用的25个技巧下一篇CANN/asc-devkit Sign接口缓冲区因子大小获取创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考