Spark 与 Hive 集成
概述
Spark 与 Hive 的集成是数仓场景的标配：Spark 复用 Hive Metastore 的元数据，用 Spark 引擎执行 Hive SQL 或读写 Hive 表。本文讲清两种集成方式（Spark on Hive / Hive on Spark）、Metastore 共享与性能差异。
一、两种集成方式
1.1 Spark on Hive
Spark 作为执行引擎，读写 Hive 元数据与数据（生产主流）：
Spark SQL / DataFrame
│ 连接 Hive Metastore（元数据）
├── 读 Hive 表数据（HDFS/Parquet）
└── 写结果回 Hive 表
Spark 引擎执行（Catalyst + Tungsten）1.2 Hive on Spark
Hive 作为入口，把 SQL 翻译后交给 Spark 执行：
Hive CLI / HiveServer2
│ Hive 解析、优化 SQL
│ 选择执行引擎（Tez / Spark / MR）
└── Spark 作为执行后端| 对比 | Spark on Hive | Hive on Spark |
|---|---|---|
| 入口 | Spark | Hive |
| 优化器 | Catalyst | Hive 优化器 |
| 适用 | 数据开发、ETL | 保留 Hive 生态、存量 SQL |
| 性能 | 更优（Catalyst+Tungsten） | 依赖 Hive 翻译 |
二、Metastore 共享
2.1 为什么共享
Hive 表元数据（表结构、分区、存储位置）存在 Metastore（MySQL 中）。共享后 Spark 能直接读写 Hive 表，无需重复建表。
2.2 配置
bash
# spark-defaults.conf 或 spark-submit
--conf spark.sql.catalogImplementation=hive
--conf spark.sql.hive.metastore.version=3.1.3
--conf spark.sql.hive.metastore.jars=path
--conf spark.sql.hive.metastore.jars.path=hdfs:///libs/hive-metastore.jar
# 连接 Metastore 地址（Thrift 协议）
--conf spark.hadoop.hive.metastore.uris=thrift://metastore-host:9083| 配置项 | 作用 |
|---|---|
spark.sql.catalogImplementation=hive | 启用 Hive 元数据 |
hive.metastore.uris | Metastore 服务地址 |
hive.metastore.version | 元数据版本 |
javax.jdo.option.ConnectionURL | 直接连 MySQL（可选） |
2.3 三种访问方式
| 方式 | 说明 |
|---|---|
| 内嵌 Metastore | 仅本地测试 |
| 远程 Metastore（推荐） | Thrift 服务，多引擎共享 |
| 直接连 MySQL | 绕过服务直连（不推荐） |
三、Spark 读 Hive 表
3.1 方式
scala
val df = spark.read.table("dw.ods_orders")
// 或
val df = spark.sql("SELECT * FROM dw.ods_orders")3.2 行为
| 行为 | 说明 |
|---|---|
| 表类型 | 内外部表均可读 |
| 分区表 | 自动分区裁剪 |
| 数据格式 | Parquet/ORC/Text 均支持 |
| 转换 | Hive 表默认转原生 Spark 读（Parquet 时） |
3.3 关键参数
| 参数 | 默认 | 说明 |
|---|---|---|
spark.sql.hive.convertMetastoreParquet | true | 用 Spark 原生读 Parquet |
spark.sql.hive.convertMetastoreOrc | false | ORC 原生转换（可开） |
spark.sql.hive.metastorePartitionPruning | true | 元数据层分区裁剪 |
spark.sql.hive.manageFilesourcePartitions | true | 文件源分区管理 |
四、Spark 写 Hive 表
4.1 写外部表
scala
df.write.mode("overwrite")
.partitionBy("dt")
.saveAsTable("dw.dws_report")4.2 写静态/动态分区
scala
// 动态分区
spark.sql("SET spark.sql.sources.partitionOverwriteMode=dynamic")
df.write.mode("overwrite").insertInto("dw.dws_report")| 模式 | 说明 |
|---|---|
| static | 分区列常量，覆盖该分区 |
| dynamic | 按数据分区列自动写，只覆盖命中的分区 |
4.3 写 Hive 注意事项
| 注意 | 说明 |
|---|---|
| 建表归属 | saveAsTable 若表不存在会新建 |
| 小文件 | 控制输出分区，必要时合并 |
| 事务表 | Hive ACID 表支持有限，注意兼容 |
| 权限 | 依赖 Hive 授权（Ranger/Sentry） |
五、Hive on Spark 配置
5.1 开启 Spark 引擎
sql
-- Hive 会话内设置
SET hive.execution.engine=spark;5.2 对比执行引擎
| 引擎 | 特点 |
|---|---|
| MapReduce | 稳定但慢 |
| Tez | 优于 MR，DAG 化 |
| Spark | 最快（内存 + 代码生成） |
| 参数 | 说明 |
|---|---|
spark.executor.memory | Spark 执行内存 |
spark.executor.cores | 并行度 |
hive.spark.client.server.connect.timeout | 会话建立超时 |
六、性能对比
6.1 Spark 原生 vs Hive 翻译
| 维度 | Spark 原生 SQL | Hive on Spark |
|---|---|---|
| 优化 | Catalyst 全量 | Hive 优化器 |
| 代码生成 | WholeStage | 部分 |
| 读取 | 原生向量化 | 经 Hive 层 |
| 复杂查询 | 更优 | 一般 |
结论：Spark on Hive 场景（Spark 入口）性能显著优于 Hive on Spark，生产首选前者；Hive on Spark 用于保留 Hive 生态的场景。
6.2 调优方向
1. 元数据：确认远程 Metastore 稳定，避免频繁元数据调用
2. 读：开 convertMetastoreParquet + 分区裁剪
3. 写：动态分区覆盖，控制小文件
4. 统计信息：ANALYZE TABLE 收集，辅助优化器常见问题速查
| 问题 | 原因与处理 |
|---|---|
| 找不到表 | Metastore 地址/版本不匹配，检查 catalogImplementation |
| 元数据慢 | 远程 Metastore 性能，升级或加缓存 |
| Parquet 读不了 | convertMetastoreParquet 与版本兼容，关掉转换 |
| 写覆盖错误 | 分区覆盖模式设错，用 dynamic |
| Hive on Spark 连接超时 | 调整 spark 会话建立与资源参数 |
| 权限拒绝 | 配置 Hive 授权与 Kerberos |