企业官网建设流程全解析

• 数据集成
• ETL
• 大数据
• 批处理
• 流处理
• 变更数据捕获

本文以仓库中的 Hive 连接器变更日志 为骨架，结合 Hive Sink 文档、Hive Source 文档 与connector-hive模块源码，完整梳理 SeaTunnel Hive 连接器从引入到成熟的能力演进路径，深入讲解 overwrite 覆盖写、多表读写、Kerberos 认证、多文件系统支持、自动建表等核心能力的底层实现，并给出可直接复用的配置示例。读完本文，你将掌握 Hive 源/汇连接器的完整选项体系、精确一次语义的实现原理，以及排查认证与小文件问题的实战思路。

1. 从变更日志看 Hive 连接器：一条清晰的能力演进主线

connector-hive模块是 SeaTunnel Connector-V2 体系中与 Hive 生态集成最深的连接器：它既能把 Hive 表作为数据源读取，也能把数据写入 Hive 表，且基于 File 连接器底座实现了文本、CSV、Parquet、ORC、JSON 等多种格式的读写。

仓库中的 Hive 连接器变更日志 记录了从2.2.0-beta至今（dev 分支）的全部重要变更。把时间线拉通，可以看到一条非常清晰的能力演进主线：

从代码结构看，Hive 连接器的能力几乎全部建立在seatunnel-connectors-v2/connector-hive模块之上，核心包结构如下：

• config/：HiveBaseOptions、HiveSinkOptions、HiveSourceOptions定义全部配置项；
• sink/：HiveSink（Sink 入口）、HiveSaveModeHandler（SaveMode 处理）、HiveSinkFactory；
• commit/：HiveSinkAggregatedCommitter（2PC 提交与 overwrite 删除逻辑）；
• source/：HiveSource、MultipleTableHiveSourceConfig、MultipleTableHiveSourceSplitEnumerator（多表分片分配）；
• storage/：StorageFactory、StorageType（HDFS/S3/OSS/COS/BOS/FILE 多文件系统适配）；
• utils/：HiveMetaStoreCatalog（元数据客户端封装）、HiveTypeConvertor（类型转换）、HiveTableTemplateUtils（建表模板）。

下文将按"读写能力 → 认证安全 → 存储与元数据 → 运维实践"的顺序逐一展开。

2. Hive Sink：从 ORC/Parquet 写入到 overwrite 覆盖写的完整演进

2.1 文件格式与写入底座

Hive Sink 在2.2.0-beta阶段分别合入了 Parquet 与 ORC 写入能力（#2310、#2311），并在2.3.1把file type统一改名为file_format_type（#4249）。从当前 HiveSink.java 源码可以看出，HiveSink 复用 File Sink 的WriteStrategyFactory生成写入策略，并通过HiveTableUtils.parseFileFormat(tableInformation)解析 Hive 表元数据中的 InputFormat/OutputFormat 来决定实际写入格式：

FileFormat fileFormat = HiveTableUtils.parseFileFormat(tableInformation); switch (fileFormat) { case TEXT: // 从 SerDe parameters 读取 field.delim / line.delim case PARQUET: case ORC: default: throw ... "Hive connector only support [text parquet orc] table now" }

因此 Hive Sink 支持的格式与 Hive 表自身的STORED AS子句必须一致，支持ORC、PARQUET、TEXT、JSON、SEQUENCE等（TEXT 场景下会自动从表元数据读取field.delim、line.delim作为分隔符）。写入的文件名表达式默认使用${transactionId}，配合 2PC 提交实现"精确一次"。

2.2 overwrite 覆盖写：2.3.12 的关键能力（重点）

2.3.12合入了 Hive Sink 支持 overwrite 模式（PR #7891）。当前实现中 overwrite 的语义由 HiveSinkAggregatedCommitter.java 承担，它继承自FileSinkAggregatedCommitter，在commit()阶段做数据清理：

• 批模式（BATCH）：提交前删除目标路径已有数据。非分区表删除整个表目录；分区表删除本次提交涉及的分区目录，然后写入新数据；
• 流模式（STREAMING）：由于commit()会在每个 checkpoint 完成后触发一次，为了避免每个 checkpoint 都重复删除导致数据丢失，committer 通过deletedTargetDirectories（ConcurrentHashMap去重集合）保证每个目标目录（表目录/分区目录）在一个 job attempt 内最多删除一次，空提交（源数据为空）会跳过删除；
• 恢复语义：committer 还会从事务目录名（如T_job_0_2中的 checkpoint id）推断是否处于恢复场景（minCheckpointIdSeen > 1视为从 checkpoint 恢复），此时删除为 best-effort 并可能被跳过，以保护已提交的数据。

配置上overwrite与data_save_mode是等价的，源码中做了归一化：

boolean overwrite = readonlyConfig.get(HiveSinkOptions.OVERWRITE); this.dataSaveMode = overwrite ? DataSaveMode.DROP_DATA : configured;

对应单测 HiveSinkAggregatedCommitterOverwriteStreamingTest.java 覆盖了三个关键场景：跨 checkpoint 分区目录只删一次、多个新分区各自只删一次、以及真实文件累积（checkpoint 2 与 3 的文件都保留在分区目录中）。测试断言deleteFile(partitionDir)恰好调用一次，直接印证了上述"最多删除一次"的实现。

2.3 SaveMode：schema_save_mode 与自动建表模板

HiveSink 实现了SupportSaveMode接口，通过 HiveSaveModeHandler.java 处理表结构与数据的保存策略。schema_save_mode支持四种取值（见 HiveSinkOptions.java）：

自动建表依赖save_mode_create_template模板，可用变量为${database}、${table}、${rowtype_fields}、${rowtype_partition_fields}、${table_location}。默认模板为 PARQUET 非分区表：

CREATE TABLE IF NOT EXISTS `${database}`.`${table}` ( ${rowtype_fields} ) STORED AS PARQUET LOCATION '${table_location}'

从 HiveSaveModeHandler.java 源码可见建表路径：若配置了模板，先尝试通过 HiveServer2 JDBC 执行建表 SQL（tryExecuteSqlViaJdbc）；JDBC 不可用时回退到 Metastore Client 方式构建Table对象并调用createTableFromTemplate。模板中STORED AS子句会驱动存储描述符选择（PARQUET/ORC/TEXTFILE 分别对应不同的 InputFormat、OutputFormat 与 SerDe），分区字段从模板的PARTITIONED BY中提取，且会自动排除在普通列之外。表位置默认通过HiveLocationUtils.qualifiedDefaultLocation基于 HDFS 或本地文件系统动态限定。

data_save_mode控制写入前如何处理目标端已有数据：APPEND_DATA（默认，保留既有数据追加写入）、DROP_DATA（等价于overwrite=true）、CUSTOM_PROCESSING/ERROR_WHEN_DATA_EXISTS（不建议在 Hive Sink 下使用）。注意overwrite=true与data_save_mode="DROP_DATA"二者择一即可，勿同时设置。

abort_drop_partition_metadata（默认false）控制 abort（中止）操作时是否从 Hive Metastore 删除分区元数据；即使该参数为false，同步过程中生成的分区数据仍会被删除，只是元数据保留。parquet_avro_write_timestamp_as_int96（默认false）支持以 INT96 写入时间戳，仅对 Parquet 文件有效（对应2.3.10的#8509变更）。

2.4 最小可运行示例

env { parallelism = 3 job.mode = "BATCH" job.name = "test_hive_source_to_hive" } source { Hive { table_name = "test_hive.test_hive_source" metastore_uri = "thrift://ctyun7:9083" } } sink { Hive { table_name = "test_hive.test_hive_sink_text_simple" metastore_uri = "thrift://ctyun7:9083" hive.hadoop.conf = { bucket = "s3a://mybucket" fs.s3a.aws.credentials.provider = "com.amazonaws.auth.InstanceProfileCredentialsProvider" } } }

需要整表替换已有数据时，追加overwrite = true（或data_save_mode = "DROP_DATA"）：

sink { Hive { table_name = "warehouse.employees" metastore_uri = "thrift://metastore:9083" overwrite = true } }

3. Hive Source：多表读取、正则匹配与分区/列投影

3.1 从单表到多表（2.3.6 → 2.3.9）

2.3.6合入的[Feature] Hive Source/Sink support multiple table (#5929)让 Hive Source 支持一次读取多张表，2.3.9的[Improve][API] Unified tables_configs and table_list (#8100)则把多表配置 API 统一收敛。当前 HiveSource.java 通过MultipleTableHiveSourceConfig持有多个HiveSourceConfig，由MultipleTableHiveSourceSplitEnumerator负责分片分配，MultipleTableHiveSourceReader负责读取，支持并行度（SupportParallelism）与列投影（SupportColumnProjection），但Boundedness为BOUNDED（仅批处理，不支持流式）。

多表读取推荐使用tables_configs：

source { Hive { tables_configs = [ { table_name = "default.seatunnel_orc_1" metastore_uri = "thrift://namenode001:9083" }, { table_name = "default.seatunnel_orc_2" metastore_uri = "thrift://namenode001:9083" } ] } }

注意：Hive 是结构化数据源，多表读取应使用tables_configs；旧的table_list已废弃，仅为向后兼容保留，新作业请勿使用。

3.2 use_regex：整库/整 Hive 同步

use_regex（默认false）开启后，table_name被视为正则表达式进行匹配（HiveSourceOptions.java）。匹配规则要点：

• 点号（.）是数据库与表的分隔符（Hive 仅支持database.table），全表达式中只允许出现 1 个未转义的点号；
• 正则中的点号必须转义为\.，在 HOCON 字符串里需写成\\.；
• db0.\\.*表示同步db0库下的所有表（整库同步）；
• \\.*.\\.*表示同步所有库下的所有表（整 Hive 同步）；
• a.tmp_\\.*表示同步a库下表名匹配tmp_.*的表。

3.3 分区与列投影

• read_partitions：指定要读取的目标分区，不设置则读取全表。注意分区列表中的每个分区必须具有相同的目录层级，例如[par1=xxx, par1=yyy/par2=zzz]是不合法的；
• read_columns：读取列列表，用于字段投影（Schema Projection）。该能力对应2.3.1的[Improve][Connector-V2][Hive] Support read text table & Column projection (#4105)。

3.4 Markdown 解析（dev 分支新能力）

变更日志最新的[Feature][File] Add markdown parser (#9714)为 Hive Source 增加了 markdown 文件读取能力：SeaTunnel 可以解析存储在 Hive 表中的 markdown 文件，提取标题、段落、列表、代码块、表格等元素，每条元素转换为一条结构化记录，schema 包含element_id、element_type（Heading、Paragraph、ListItem 等）、heading_level（1-6，非标题元素为 null）、text、page_number（默认 1）、position_index、parent_id、child_ids。注意 Markdown 格式仅支持读取，不支持写入。

4. 认证与安全：Kerberos 与多 Hadoop 账号

4.1 Kerberos 支持的时间线

Kerberos 支持经历了一个逐步修复的过程，这也解释了为何当前配置项如此完整：

从 HiveMetaStoreProxyUtils.java 源码可以看出 Kerberos 的启用判定：只有同时配置了kerberos_principal与kerberos_keytab_path才返回 true；只配置了其中一个会直接抛出IllegalArgumentException（分别提示 "Please set kerberosPrincipal" / "Please set kerberosKeytabPath"）。这正是#6307修复后的严谨判定逻辑。

4.2 Kerberos 配置示例

source { Hive { table_name = "default.test_hive_sink_on_hdfs_with_kerberos" metastore_uri = "thrift://metastore:9083" hive_site_path = "/tmp/hive-site.xml" kerberos_principal = "hive/metastore.seatunnel@EXAMPLE.COM" kerberos_keytab_path = "/tmp/hive.keytab" krb5_path = "/tmp/krb5.conf" plugin_output = hive_source } } sink { Assert { plugin_input = hive_source rules { row_rules = [ { rule_type = MAX_ROW rule_value = 3 } ] # field_rules 可对 pk_id / name / score 做 NOT_NULL 校验 } } }

在 HiveMetaStoreCatalog.java 中，loginWithKerberos通过HadoopLoginFactory.loginWithKerberos(authConf, krb5Path, principal, keytabPath, ...)完成登录，并在每次getClient()时调用maybeRelogin()——若 UGI 来自 keytab，则执行checkTGTAndReloginFromKeytab()以应对 TGT 过期，保证长作业的认证不中断。未配置 Kerberos 时，可通过remote_user指定 HDFS 远端用户名（对应2.3.4的Support using multiple hadoop account (#5903)）。

4.3 配置文件加载优先级

hive-site.xml/hdfs-site.xml/core-site.xml的加载在2.3.6前后经过多次修复（#7069、#7017）。当前buildHiveConf()的加载顺序为：

1. metastore_uri写入hive.metastore.uris；
2. hive.hadoop.conf-path目录下的hive-site.xml（HADOOP_CONF_FILES）；
3. hive_site_path指定的hive-site.xml文件；
4. hive.hadoop.confMap 中的属性逐条写入（优先级最高）。

5. 存储与元数据：多文件系统与 Metastore 高可用

5.1 多文件系统支持（2.3.6）

[Improve][Connector-v2] The hive connector support multiple filesystem (#6648)使 Hive 连接器不再局限于 HDFS。当前 StorageFactory.java 依据 Hive 表Sd.Location的前缀自动选择存储实现：

if (location.startsWith("s3")) -> S3Storage else if (location.startsWith("oss")) -> OSSStorage else if (location.startsWith("cos")) -> COSStorage else if (location.startsWith("bos")) -> BOSStorage else if (location.startsWith("file")) -> HDFSStorage(location.replace("file:", "file:/")) else -> HDFSStorage(location) // 默认 HDFS

StorageType.java 定义了S3 / OSS / COS / BOS / FILE / HDFS六种类型，对应测试（BosStorageTest、CosStorageTest等）覆盖各存储的配置构建。使用对象存储时，需在hive.hadoop.conf中补充 bucket 与凭据 Provider，例如：

sink { Hive { table_name = "test_hive.test_hive_sink_on_s3" metastore_uri = "thrift://ip-192-168-0-202:9083" hive.hadoop.conf-path = "/home/ec2-user/hadoop-conf" hive.hadoop.conf = { bucket = "s3://ws-package" fs.s3a.aws.credentials.provider = "com.amazonaws.auth.InstanceProfileCredentialsProvider" } } }

5.2 metastore_uri 多 URI 高可用与 Glue Data Catalog

metastore_uri支持逗号分隔配置多个 URI 用于高可用/故障切换（自动去除空格），SeaTunnel 会将其写入hive.metastore.uris，并优先使用 Hive 的RetryingMetaStoreClient实现重试/切换：

sink { Hive { table_name = "default.seatunnel_orc" metastore_uri = "thrift://metastore-1:9083,thrift://metastore-2:9083" } }

从 HiveMetaStoreCatalog.java 源码可见，tryCreateRetryingClient会反射调用RetryingMetaStoreClient.getProxy(HiveConf, boolean)（同时兼容 Hive 2.x 的HiveConf签名与 Hive 3.x 的Configuration签名），失败则回退到HiveMetaStoreClient。此外还支持通过 Hive 标准的 metastore 客户端工厂接入 AWS Glue Data Catalog（配置hive.metastore.client.factory.class），此时可省略metastore_uri；若 URI 与客户端工厂均未配置，SeaTunnel 不会回退到内嵌 metastore，Sink 初始化会直接失败。

6. 精确一次语义与异常处理

Hive Sink 默认使用 2PC 提交保证"精确一次"（对应 Sink 文档 中的关键特性说明）。写入数据会先落在临时事务目录（文件名表达式${transactionId}），提交阶段由HiveSinkAggregatedCommitter.commit()完成：

1. 若data_save_mode == DROP_DATA（含overwrite=true），先删除目标目录（受"最多删除一次"保护）；
2. 调用父类super.commit()将临时文件移动到最终目录；
3. 调用hiveMetaStore.addPartitions(dbName, tableName, partitions)为分区表注册新分区；分区已存在时捕获AlreadyExistsException并告警（对应2.3.5的#6577修复）。

异常路径上，abort()会先清理临时文件，再根据abort_drop_partition_metadata决定是否删除分区元数据。历史上多次资源泄漏、空指针、线程安全问题的修复（如#6721、#7205、#9282、#3258、#3506）也体现了该提交链路在持续加固。另一个值得注意的细节是2.2.0-beta的filter '_SUCCESS' file in file list (#2235)：Hive Source 枚举文件时会过滤_SUCCESS标记文件，避免把 Spark/Hive 任务的成功标记当作数据文件读取。

7. 运维实战：依赖部署与常见问题

7.1 依赖准备

使用 SeaTunnel 引擎（Zeta）时，需将seatunnel-shade-hadoop3-uber-3.1.4-3.0.0.jar、hive-exec-3.1.3.jar和libfb303-0.9.3.jar放入$SEATUNNEL_HOME/lib/目录。测试过的 Hive 版本为 2.3.9 与 3.1.3（2.3.6的[Improve][Hive] update hive3 version (#6699)即为此适配）。若使用 Hive on S3/OSS/COS 等对象存储，还需在$SEATUNNEL_HOME/plugins/Hive/lib下准备对应的hive-exec及对象存储 SDK jar（如 AWS EMR 环境的hadoop-aws、emrfs-hadoop-assembly等；OSS 场景使用 JindoSDK 并移除冲突的hadoop-aliyun-*.jar）。

7.2 小文件治理

当任务并行度较高或批次较小时，Hive 表会产生大量小文件。治理手段：

• 降低env块中的parallelism；
• 增大batch_size，让每个 task 写入更大的文件；
• 定期执行ALTER TABLE ... CONCATENATE或使用 Spark merge 任务合并小文件。

7.3 Schema 演变注意事项

Hive Sink 会从 Hive Metastore 读取当前表 schema；若上游新增了列，需先手动执行ALTER TABLE更新 Hive 表结构，SeaTunnel 不会自动修改 Hive schema。分区表的写入依赖partition_by指定分区字段（或由自动建表模板的PARTITIONED BY派生），SeaTunnel 会按分区字段自动写入正确的分区目录。

7.4 Hive 版本与兼容性提醒

• Hive Source 仅支持批处理（BOUNDED），不支持流式；
• 连接器依赖 Hive 元数据 API（Thrift），metastore 服务的可用性直接影响作业启动；metastore_uri的多 URI 仅做客户端连接端点切换，元数据一致性仍需由 metastore 部署本身保证；
• 自动建表默认不自动创建数据库（RECREATE_SCHEMA/CREATE_SCHEMA_WHEN_NOT_EXIST模式下数据库不存在都会报错，需要预先建库）。

8. 延伸阅读：仓库内相关路径导航

• 数据集成
• ETL
• 大数据
• 批处理
• 流处理
• 变更数据捕获