SeaTunnel HdfsFile Sink 连接器完全指南：从基础配置到 HA/Kerberos/ViewFS 实战

SeaTunnel HdfsFile Sink 连接器完全指南：从基础配置到 HA/Kerberos/ViewFS 实战

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

SeaTunnel 的 HdfsFile Sink 连接器用于将上游数据写入 HDFS 文件系统，支持 text、csv、parquet、orc、json、excel、xml、binary 八种文件格式，并提供事务提交、分区写入、自定义文件名、Kerberos 认证与 schema 演进等能力。本文以官方文档 docs/en/connectors/sink/HdfsFile.md 为主体，结合仓库中connector-file-hadoop与connector-file-base模块的源码，系统讲解该连接器的全部配置项、底层实现原理与可运行的配置示例，帮助你直接落地到实际同步任务中。

支持引擎与关键特性

HdfsFile Sink 同时支持三种引擎：

• Spark
• Flink
• SeaTunnel Zeta（引擎自带 Hadoop 依赖，无需额外集成）

关键特性一览（对应 connector-v2-features）：

• 多模态（multimodal）：底层使用二进制文件格式读写任意格式文件（视频、图片等），可实现任意文件的同步。
• 精确一次（exactly-once）：默认使用两阶段提交（2PC）保证数据不丢失、不重复。
• 多表写入（support multiple table write）：一个任务可同时向多张目标表写入。
• 文件格式：text、csv、parquet、orc、json、excel、xml、binary。
• 压缩编码：lzo、canal_json、debezium_json、maxwell_json（其中 canal_json / debezium_json / maxwell_json 实际是 CDC 事件的行序列化格式，与merge_update_event配合使用）。
• 不支持 timer flush：文件刷新时机完全由 checkpoint 与batch_size决定。

支持的 HDFS 版本：Hadoop 2.x 与 3.x。

使用提示：若在 Spark/Flink 上使用，必须确保集群已集成 Hadoop（文档中已验证的版本为 2.x）；若使用 SeaTunnel Zeta 引擎，安装包lib目录下已自动包含 Hadoop jar，可通过检查${SEATUNNEL_HOME}/lib下的 jar 包确认。

源码结构：HdfsFile Sink 的实现骨架

在深入配置之前，先了解该连接器在仓库中的代码组织，方便后续对照：

• 连接器入口模块：connector-file-hadoop（约 20 个 Java 文件，仅实现 HDFS 特有逻辑）
• 文件连接器通用模块：connector-file-base（文件读写、事务、分区、格式策略等通用实现）

核心类调用链如下：

组件	文件	职责
插件入口	HdfsFileSink.java	继承BaseMultipleTableFileSink，返回插件名HdfsFile
插件工厂	HdfsFileSinkFactory.java	声明OptionRule条件化校验、构建HadoopConf
配置定义	HdfsFileSinkOptions.java	直接继承FileBaseSinkOptions，无新增项
全部 Sink 选项	FileBaseSinkOptions.java	所有参数定义、默认值与约束（约 40 个 Option）
Hadoop 客户端配置	HadoopConf.java	构造Configuration，区分 hdfs / viewfs scheme
写入器	BaseFileSinkWriter.java	事务恢复、write/prepareCommit/snapshotState
提交器	FileSinkAggregatedCommitter.java	通过renameFile把临时文件 mv 到目标目录完成提交
写策略	AbstractWriteStrategy.java	文件名生成、事务目录管理、分区目录计算

Sink 参数完整参考表

以下参数即官方文档全量整理，默认值与约束与 FileBaseSinkOptions.java 中的定义一一对应。

名称	类型	必填	默认值	说明
fs.defaultFS	string	是	-	Hadoop 集群地址。支持hdfs://hadoopcluster、hdfs://namenode:9000（标准 HDFS）、viewfs://mycluster（ViewFS 联邦 HDFS），ViewFS 配置示例见下文
path	string	是	-	目标目录路径，必填
tmp_path	string	是	/tmp/seatunnel	结果文件先写入临时路径，提交时通过mv移动到目标目录，必须是 HDFS 路径
hdfs_site_path	string	否	-	hdfs-site.xml的路径，用于加载 NameNode 的 HA 配置
custom_filename	boolean	否	false	是否需要自定义文件名
file_name_expression	string	否	"${transactionId}"	仅当custom_filename=true时生效。描述生成到path的文件名表达式，支持${now}与${uuid}变量，如test_${uuid}_${now}；${now}的格式由filename_time_format定义。注意：当is_enable_transaction=true时，会自动在文件名头部追加${transactionId}_
filename_time_format	string	否	"yyyy.MM.dd"	仅当custom_filename=true时生效。指定file_name_expression中${now}的时间格式。常用占位符：y（年）、M（月）、d（日）、H（小时 0-23）、m（分钟）、s（秒）
file_format_type	string	否	"csv"	文件类型：text、csv、parquet、orc、json、excel、xml、binary。最终文件名会以文件格式后缀结尾，其中 text 文件后缀为txt
filename_extension	string	否	-	覆盖默认文件名后缀，例如.xml、.json、dat、.customtype
field_delimiter	string	否	text 为 '\001'，csv 为 ','	仅 text/csv 格式生效，字段间分隔符。源码中默认值来自TextFormatConstant.SEPARATOR[0]（\001）
row_delimiter	string	否	"\n"	仅 text、csv、json 格式生效，行间分隔符
have_partition	boolean	否	false	是否进行分区处理
partition_by	array	否	-	仅当have_partition=true时生效，按所选字段分区
partition_dir_expression	string	否	"${k0}=${v0}/${k1}=${v1}/.../${kn}=${vn}/"	仅当have_partition=true时生效，根据分区信息生成分区目录，k0为第一个分区字段，v0为其值
is_partition_field_write_in_file	boolean	否	false	仅当have_partition=true时生效。为true时分区字段及其值会写入数据文件；例如要写 Hive 数据文件，该值应为false
sink_columns	array	否	空（全部字段）	需要写入文件的列。为空时写入Transform或Source传来的所有字段，字段顺序决定实际写入顺序
is_enable_transaction	boolean	否	true	为true时保证写入目标目录的数据不丢失、不重复，且自动在文件名头部追加${transactionId}_。当前仅支持true
batch_size	int	否	1000000	单个文件的最大行数。SeaTunnel Zeta 引擎下文件行数由batch_size与checkpoint.interval共同决定：若 checkpoint 间隔足够大，writer 会持续写入直到超过batch_size；若 checkpoint 间隔小，则每次 checkpoint 触发时新建文件
compress_codec	string	否	none	文件压缩编码。各格式支持：txt：lzo、none；json：lzo、none；csv：lzo、none；orc：lzo、snappy、lz4、zlib、none；parquet：lzo、snappy、lz4、gzip、brotli、zstd、none。excel 不支持任何压缩
krb5_path	string	否	/etc/krb5.conf	Kerberos 的 krb5 配置文件路径
kerberos_principal	string	否	-	Kerberos principal
kerberos_keytab_path	string	否	-	Kerberos keytab 路径
common-options	object	否	-	Sink 插件公共参数，详见 Sink Common Options
max_rows_in_memory	int	否	-	仅 excel 格式生效，内存中可缓存的最大数据条数
sheet_max_rows	int	否	1048576	仅 excel 格式生效
sheet_name	string	否	Sheet${随机数}	仅 excel 格式生效，写入的工作表名称
csv_string_quote_mode	enum	否	MINIMAL	仅 csv 格式生效，字符串引号模式（见下文详解）
xml_root_tag	string	否	RECORDS	仅 xml 格式生效，XML 根元素标签名
xml_row_tag	string	否	RECORD	仅 xml 格式生效，数据行标签名
xml_use_attr_format	boolean	否	-	仅 xml 格式生效，是否使用标签属性格式处理数据
single_file_mode	boolean	否	false	每个并行度只输出一个文件。开启后batch_size不生效，输出文件名不带文件块后缀
create_empty_file_when_no_data	boolean	否	false	上游无数据同步时，仍生成对应的空数据文件
parquet_avro_write_timestamp_as_int96	boolean	否	false	仅 parquet 格式生效，是否将 timestamp 写入 Parquet INT96
parquet_avro_write_fixed_as_int96	array	否	-	仅 parquet 格式生效，是否将 12 字节字段写入 Parquet INT96
enable_header_write	boolean	否	false	仅 text、csv 格式生效。false 不写表头，true 写表头
encoding	string	否	"UTF-8"	仅 json、text、csv、xml 格式生效，输出文件编码
remote_user	string	否	-	HDFS 远程用户名
schema_evolution_enabled	boolean	否	false	为 CDC 管道启用 schema 演进，true时 ADD/DROP/RENAME/MODIFY 列事件无需重启任务即可应用到 sink。binary 格式不支持
schema_save_mode	string	否	CREATE_SCHEMA_WHEN_NOT_EXIST	目录已存在时的处理方式
data_save_mode	string	否	APPEND_DATA	已有数据的处理方式
multi_table_sink_replica	int	否	1	多表 Sink 任务中每张表的 sink writer 副本数
merge_update_event	boolean	否	false	仅 canal_json、debezium_json、maxwell_json 格式生效。为true时将 UPDATE_AFTER 与 UPDATE_BEFORE 事件合并为 UPDATE 事件数据

参数间的条件化校验

从 HdfsFileSinkFactory.java 的optionRule()可以看到，连接器使用conditional实现了参数间的依赖校验，配置不符合条件会在任务启动时直接报错：

• file_format_type=text时，才接受row_delimiter、field_delimiter、TXT_COMPRESS（lzo/none）、enable_header_write；
• file_format_type=csv时，接受row_delimiter、TXT_COMPRESS、enable_header_write；
• file_format_type=json时，接受row_delimiter、TXT_COMPRESS；
• file_format_type=orc时，接受ORC_COMPRESS（lzo/snappy/lz4/zlib/none）；
• file_format_type=parquet时，接受PARQUET_COMPRESS（lzo/snappy/lz4/gzip/brotli/zstd/none）及两个 INT96 选项；
• file_format_type=xml时，接受xml_use_attr_format、xml_root_tag、xml_row_tag；
• custom_filename=true时，才接受file_name_expression、filename_time_format；
• have_partition=true时，才接受partition_by、partition_dir_expression、is_partition_field_write_in_file；
• file_format_type为 text/json/csv/xml 时，才接受encoding。

这也解释了压缩格式矩阵：不同文件格式的压缩选项在代码中以singleChoice枚举白名单约束（见 FileBaseSinkOptions.java），例如 ORC 不允许 gzip/brotli/zstd，parquet 不允许 zlib。

schema_save_mode 与 data_save_mode：目录与数据生命周期管理

这两个参数控制任务启动前对目标目录的预处理策略，与文件连接器的 SaveMode 机制对应（源码见 sink/config/SaveMode.java）。

schema_save_mode（目录已存在时的处理）

• RECREATE_SCHEMA：目录不存在则创建，目录存在则删除后重建；
• CREATE_SCHEMA_WHEN_NOT_EXIST：目录不存在则创建，存在则跳过（默认）；
• ERROR_WHEN_SCHEMA_NOT_EXIST：目录不存在时报错；
• IGNORE：忽略对表的处理。

data_save_mode（已有数据的处理）

• DROP_DATA：保留目录，删除已有数据文件；
• APPEND_DATA：保留目录，保留已有数据文件（默认）；
• ERROR_WHEN_DATA_EXISTS：已有数据文件时报错。

从代码约束看，data_save_mode仅允许DROP_DATA、APPEND_DATA、ERROR_WHEN_DATA_EXISTS三个取值（FileBaseSinkOptions.java）。典型使用场景：落地 Hive 数仓表时建议schema_save_mode=CREATE_SCHEMA_WHEN_NOT_EXIST+data_save_mode=DROP_DATA（先清空再写，保证当天数据无残留）；增量场景用APPEND_DATA。

schema_evolution_enabled：CDC 管道下的在线 Schema 演进

当schema_evolution_enabled=true时，文件 Sink 可在运行期处理 CDC 的 Schema 变更事件（ADD COLUMN、DROP COLUMN、RENAME COLUMN、MODIFY COLUMN），无需重启任务。每次 schema 变更时，当前输出文件会被关闭，并以更新后的 schema 打开新文件写入。

• 支持的格式：除binary外的所有文件格式。若file_format_type=binary却开启该选项，任务启动时会被配置校验拦截并报错。
• 分区约束：当have_partition=true时，不允许 DROP 掉partition_by中列出的分区列，否则会快速失败（fail fast）。分区列必须跨 schema 变更保持稳定。
• 当schema_evolution_enabled=false（默认）时：如果上游 CDC 源开启了schema-changes.enabled=true，且AlterTableEvent到达 sink，任务会立即抛出可操作的错误：

Received AlterTableEvent but schema_evolution_enabled=false at this sink. Either set schema_evolution_enabled=true to handle schema changes, or set schema-changes.enabled=false at the CDC source to suppress them.

使用默认 CDC 源配置（schema-changes.enabled=false）的用户完全不受影响。

• 已知限制：schema 变更与 checkpoint 不是原子的。若任务在文件轮转与 schema 元数据更新之间的窗口内崩溃，恢复后写入的行可能仍使用变更前的 schema。这是 SeaTunnel 其他 Sink 共有的架构性缺口，完整的"重启 + DDL 恢复"正确性需要配合后续 CDC 源修复（另行跟踪）。

源码层面，写入器通过SupportSchemaEvolutionSinkWriter接口暴露applySchemaChange(SchemaChangeEvent)（见 BaseFileSinkWriter.java），将事件转交给WriteStrategy完成文件轮转与 schema 更新。格式侧还有 FileSchemaEvolutionTest.java 与 ParquetWriteStrategyEvolutionTest.java 等测试覆盖该能力。

CDC 管道中的示例：

HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/seatunnel/cdc/${table_name}" file_format_type = "parquet" schema_evolution_enabled = true }

multi_table_sink_replica 与 merge_update_event

multi_table_sink_replica：多表 Sink 任务中每张表使用的 sink writer 副本数，默认1。仅当单张表需要更高的 writer 并行度时才需要调大。

merge_update_event：仅canal_json、debezium_json、maxwell_json三种 CDC 序列化格式生效。为true时，UPDATE_AFTER 与 UPDATE_BEFORE 事件被合并为 UPDATE 事件数据；为false时二者会作为独立事件分别序列化。对应源码定义见 FileBaseSinkOptions.java。

csv_string_quote_mode：CSV 字符串引号模式

当文件格式为 CSV 时的字符串引号策略（对应 CsvStringQuoteMode）：

• ALL：所有字符串字段都加引号；
• MINIMAL（默认）：仅当字段包含特殊字符（如字段分隔符、引号字符或行分隔符字符串中的任意字符）时才加引号；
• NONE：永不加引号。当数据中出现分隔符时，printer 会用转义字符作为前缀；若未设置转义字符，格式校验会抛出异常。

任务示例：从简单到生产级

最简配置（FakeSource → HDFS）

以下任务通过 FakeSource 自动生成 16 行多种类型的测试数据，写入 HDFS 的 ORC 文件：

# Defining the runtime environment env { parallelism = 1 job.mode = "BATCH" } source { # This is a example source plugin **only for test and demonstrate the feature source plugin** FakeSource { parallelism = 1 plugin_output = "fake" row.num = 16 schema = { fields { c_map = "map<string, smallint>" c_array = "array<int>" c_string = string c_boolean = boolean c_tinyint = tinyint c_smallint = smallint c_int = int c_bigint = bigint c_float = float c_double = double c_decimal = "decimal(30, 8)" c_bytes = bytes c_date = date c_timestamp = timestamp } } } } transform { # If you would like to get more information about how to configure seatunnel and see full list of transform plugins, # please go to https://seatunnel.apache.org/docs/transforms } sink { HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/hive/warehouse/test2" file_format_type = "orc" } # If you would like to get more information about how to configure seatunnel and see full list of sink plugins, # please go to https://seatunnel.apache.org/docs/connectors/sink }

ORC 格式最简配置

HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/hive/warehouse/test2" file_format_type = "orc" }

text 格式 + 分区 + 自定义文件名 + 列裁剪

HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/hive/warehouse/test2" 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}" filename_time_format = "yyyy.MM.dd" sink_columns = ["name","age"] is_enable_transaction = true }

parquet 格式 + 分区 + 自定义文件名 + 列裁剪

HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/hive/warehouse/test2" 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}" filename_time_format = "yyyy.MM.dd" file_format_type = "parquet" sink_columns = ["name","age"] is_enable_transaction = true }

Kerberos 认证最简配置

HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/hive/warehouse/test2" hdfs_site_path = "/path/to/your/hdfs_site_path" kerberos_principal = "[e-mail]" kerberos_keytab_path = "/path/to/your/keytab/file.keytab" }

压缩配置

HdfsFile { fs.defaultFS = "hdfs://hadoopcluster" path = "/tmp/hive/warehouse/test2" compress_codec = "lzo" }

深入原理：事务提交与文件名生成

两阶段提交：tmp_path 与 rename

连接器的 exactly-once 承诺源自"临时目录 + 原子 mv"机制，链路如下：

1. 数据先写入tmp_path下的事务目录（默认/tmp/seatunnel）。每个 checkpoint 对应一个事务，事务目录结构由getTransactionDir(transactionId)计算，事务 ID 格式为T_{jobId}_{uuidPrefix}_{subTaskIndex}_{checkpointId}（见 AbstractWriteStrategy.java），uuidPrefix是 10 位随机 UUID，保证跨任务不冲突。
2. 每个 checkpoint 到达时，writer 调用prepareCommit()关闭当前文件，返回FileCommitInfo（含needMoveFiles映射：临时文件 → 目标文件，见 AbstractWriteStrategy.java）。
3. checkpoint 完成时，FileSinkAggregatedCommitter.java 遍历mvFileEntry，调用hadoopFileSystemProxy.renameFile(临时路径, 目标路径, true)将临时文件原子 mv 到目标目录，即文档所述"先写 tmp，再用 mv 提交到目标目录"。
4. 若任务从 checkpoint 恢复，writer 会找出未包含在恢复状态中的孤儿事务并调用abortPrepare(transaction)删除对应事务目录，只由 aggregated committer 重放已 checkpoint 的事务（BaseFileSinkWriter.java）。这也解释了is_enable_transaction=true时文件名自动带${transactionId}_前缀的原因：不同事务、不同并行子任务之间的文件必须可区分，才能在恢复时精确取舍。

文件名生成规则

generateFileName(transactionId)（AbstractWriteStrategy.java）按以下顺序拼接：

1. 若配置了filename_extension，直接使用该后缀（自动补.前缀）；否则使用文件格式后缀 + 压缩编码后缀（例如 parquet + lzo 会得到类似.parquet.lzo的结果）；
2. 对file_name_expression执行变量替换：${transactionId}→ 当前事务 ID、${now}→ 按filename_time_format格式化的当前时间、${uuid}→ 随机 UUID；
3. 若未开启single_file_mode，追加_+ 子任务编号（partId），保证同一事务内不同并行子任务的文件不冲突；
4. 追加文件后缀。

single_file_mode 与 binary 自定义文件名的启动校验

BaseFileSinkWriter.java 的preCheckConfig还包含两条启动期校验：

• binary + 自定义文件名 + 并行度大于 1：文件名表达式必须包含${transactionId}或${uuid}，否则报错——因为并行子任务共享同一文件名时会导致互相覆盖；
• single_file_mode + 并行度大于 1：文件名表达式必须包含${transactionId}，否则报错。

这两条保证了多并行度下输出文件名的唯一性。

ViewFS（联邦 HDFS）配置实战

ViewFS 可将多个 HDFS 集群或 namespace 统一挂载为一个逻辑命名空间，适用于 HDFS Federation 场景。连接器在 HadoopConf.java 中通过检测fs.defaultFS是否以viewfs://开头，自动选择org.apache.hadoop.fs.viewfs.ViewFileSystem实现并切换fs.viewfs.impl与fs.defaultFS配置。

HdfsFile { fs.defaultFS = "viewfs://mycluster" path = "/data/output" file_format_type = "parquet" hdfs_site_path = "/path/to/core-site.xml" data_save_mode = "DROP_DATA" }

在core-site.xml中配置挂载表：

<?xml version="1.0" encoding="UTF-8"?> <configuration> <property> <name>fs.viewfs.mounttable.mycluster.link./data</name> <value>hdfs://namenode1:9000/data</value> </property> <property> <name>fs.viewfs.mounttable.mycluster.link./logs</name> <value>hdfs://namenode2:9000/logs</value> </property> <property> <name>fs.viewfs.mounttable.mycluster.link./tmp</name> <value>hdfs://namenode3:9000/tmp</value> </property> </configuration>

注意hdfs_site_path指向包含挂载表配置的core-site.xml（或hdfs-site.xml）。在 HadoopConf.setExtraOptionsForConfiguration 中，该 XML 会被加载进 HadoopConfiguration，且会先 unsetfs.defaultFS、fs.{schema}.impl、fs.{schema}.impl.disable.cache三个键，避免外部 XML 覆盖连接器自身构造的核心配置。

写入 HA HDFS 集群（启用 Kerberos）

写入使用 Kerberos 的 HA HDFS 集群时，除 nameservice URI 外还需提供 principal/keytab。连接器复用 Hadoop 工具链的同一套认证机制，因此 principal 的 HDFS 权限必须允许写入目标目录。

sink { HdfsFile { fs.defaultFS = "hdfs://mycluster" path = "/data/landing/events" file_format_type = "parquet" hdfs_site_path = "/etc/hadoop/conf/hdfs-site.xml" kerberos_principal = "[e-mail]" krb5_path = "/etc/krb5.conf" } }

关键说明：

• kerberos_principal与krb5_path会被转发给 Hadoop FileSystem 客户端（见 HdfsFileSinkFactory.initHadoopConf）；连接器本身不执行kinit，因此 keytab 必须已能在每个 worker 节点被发现（通常通过KRB5CCNAME或kinitcron 实现），或通过标准 Hadoop 认证工具提供给同一 JVM。
• HA 场景下hdfs_site_path用于加载 NameNode HA 配置（nameservice → active/standby 映射）。
• 若出现集群级认证问题，检查 worker 日志中的LoginException/KrbException信息——这些表明凭据问题，而非连接器缺陷。remote_user参数可显式指定 HDFS 远端用户名。

源码级测试验证

仓库为 HdfsFile 连接器提供了较为完整的测试覆盖，可作为配置正确性的参考：

• HdfsFileSinkTest.java：Sink 工厂与配置解析测试；
• HdfsFileFactoryTest.java：插件工厂行为测试；
• 通用写入策略测试（text/csv/orc/parquet 等）集中在 connector-file-base 的 writer 测试目录，例如 CsvWriteStrategyTest.java、OrcWriteStrategyTest.java、ParquetWriteStrategyTest.java；
• 两阶段提交行为见 FileSinkAggregatedCommitterTest.java；
• Kerberos 相关工具类见 HadoopFileSystemProxyKerberosRenewTest.java。

小结与选型建议

HdfsFile Sink 是 SeaTunnel 面向 HDFS 数仓落地的核心出口：通过 tmp_path 事务目录与 rename 提交实现 exactly-once，通过 WriteStrategy 抽象统一八种文件格式与压缩编码，通过 SaveMode 管理目录与数据生命周期，并针对联邦 HDFS、HA + Kerberos、CDC schema 演进等生产场景提供了完整参数。上手时建议遵循以下顺序验证：先跑通最简 ORC 配置确认连通性 → 按需开启分区与自定义文件名 → 接入生产集群时再启用hdfs_site_path/ Kerberos 认证与合适的schema_save_mode/data_save_mode组合，避免数据残留或误删。

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel