Spark SQL 与 Hive 集成主要通过 Hive Metastore 实现，让 Spark 能够直接访问和管理 Hive 的表和数据。

1. 配置 Spark 与 Hive 集成

基础配置方式

# 启动 Spark Shell 时指定 Hive 配置
spark-shell --conf spark.sql.warehouse.dir=/user/hive/warehouse \
            --conf spark.sql.catalogImplementation=hive \
            --jars /path/to/hive-metastore.jar,/path/to/mysql-connector.jar

配置文件方式（spark-defaults.conf）

# 启用 Hive 支持
spark.sql.catalogImplementation=hive
spark.sql.warehouse.dir=hdfs://namenode:8020/user/hive/warehouse

# Hive Metastore 配置
spark.hadoop.hive.metastore.uris=thrift://metastore-host:9083
spark.hadoop.hive.metastore.warehouse.dir=hdfs://namenode:8020/user/hive/warehouse

# 数据库连接（如果使用外部数据库）
spark.hadoop.javax.jdo.option.ConnectionURL=jdbc:mysql://mysql-host:3306/hive
spark.hadoop.javax.jdo.option.ConnectionUserName=hive
spark.hadoop.javax.jdo.option.ConnectionPassword=password

编程方式配置

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("SparkHiveIntegration")
  .config("spark.sql.warehouse.dir", "/user/hive/warehouse")
  .config("spark.sql.catalogImplementation", "hive")
  .config("hive.metastore.uris", "thrift://metastore-host:9083")
  .enableHiveSupport()  // 关键：启用 Hive 支持
  .getOrCreate()

2. 在 Spark SQL 中查询 Hive 表

方法一：使用 Spark SQL 直接查询

// 查询所有数据库
spark.sql("SHOW DATABASES").show()

// 切换数据库
spark.sql("USE my_database")

// 查看表列表
spark.sql("SHOW TABLES").show()

// 简单查询
spark.sql("SELECT * FROM users LIMIT 10").show()

// 复杂查询
spark.sql("""
  SELECT 
    u.user_id,
    u.name,
    COUNT(o.order_id) as order_count,
    SUM(o.amount) as total_amount
  FROM users u
  LEFT JOIN orders o ON u.user_id = o.user_id
  WHERE u.create_date >= '2024-01-01'
  GROUP BY u.user_id, u.name
  HAVING COUNT(o.order_id) > 5
  ORDER BY total_amount DESC
""").show()

方法二：使用 DataFrame API 查询

// 直接读取 Hive 表为 DataFrame
val usersDF = spark.table("my_database.users")
val ordersDF = spark.table("my_database.orders")

// DataFrame 操作
val result = usersDF
  .join(ordersDF, "user_id")
  .filter($"create_date" >= "2024-01-01")
  .groupBy("user_id", "name")
  .agg(
    count("order_id").as("order_count"),
    sum("amount").as("total_amount")
  )
  .filter($"order_count" > 5)
  .orderBy(desc("total_amount"))

result.show()

方法三：使用 Catalog API 管理元数据

// 获取 catalog 实例
val catalog = spark.catalog

// 列出所有数据库
catalog.listDatabases().show()

// 列出指定数据库的表
catalog.listTables("my_database").show()

// 获取表详细信息
catalog.listColumns("my_database", "users").show()

// 检查表是否存在
if (catalog.tableExists("my_database", "users")) {
  println("表存在")
}

3. 读写 Hive 表操作

读取 Hive 表数据

// 读取整个表
val fullTableDF = spark.read.table("database_name.table_name")

// 读取特定分区
val partitionedDF = spark.sql("""
  SELECT * FROM sales 
  WHERE dt = '2024-01-15' AND region = 'north'
""")

// 使用 DataFrameReader
val df = spark.read
  .format("hive")
  .option("path", "/user/hive/warehouse/database.db/table")
  .load()

写入数据到 Hive 表

// 追加数据到现有表
resultDF.write
  .mode("append")
  .saveAsTable("my_database.result_table")

// 覆盖表数据
resultDF.write
  .mode("overwrite")
  .saveAsTable("my_database.result_table")

// 插入特定分区
resultDF.write
  .mode("append")
  .insertInto("my_database.partitioned_table")

// 动态分区写入
resultDF.write
  .mode("append")
  .partitionBy("dt", "region")  // 分区列
  .saveAsTable("my_database.dynamic_partition_table")

4. 高级集成特性

外部表与内部表处理

// 创建外部表（数据保留在HDFS）
spark.sql("""
  CREATE EXTERNAL TABLE external_users (
    user_id INT,
    name STRING,
    email STRING
  )
  LOCATION 'hdfs:///data/external/users'
""")

// 创建内部表（由Hive管理数据）
spark.sql("""
  CREATE TABLE managed_users (
    user_id INT,
    name STRING,
    email STRING
  )
  STORED AS ORC
""")

分区表操作

// 创建分区表
spark.sql("""
  CREATE TABLE partitioned_sales (
    product STRING,
    amount DOUBLE
  )
  PARTITIONED BY (sale_date STRING, region STRING)
  STORED AS PARQUET
""")

// 添加分区
spark.sql("ALTER TABLE partitioned_sales ADD PARTITION (sale_date='2024-01-01', region='north')")

// 分区查询优化
spark.sql("SELECT * FROM partitioned_sales WHERE sale_date = '2024-01-01'")

性能优化配置

// 启用动态分区
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")

// 向量化查询
spark.conf.set("spark.sql.hive.convertMetastoreParquet", "true")

// 谓词下推优化
spark.conf.set("spark.sql.parquet.filterPushdown", "true")
spark.conf.set("spark.sql.orc.filterPushdown", "true")

5. 故障排查与验证

验证连接状态

// 测试 Hive Metastore 连接
try {
  spark.sql("SHOW DATABASES").show()
  println("Hive 集成成功")
} catch {
  case e: Exception => 
    println(s"Hive 集成失败: ${e.getMessage}")
    // 检查配置
    println(s"Metastore URIs: ${spark.conf.get("hive.metastore.uris")}")
}

// 检查表属性
spark.sql("DESCRIBE FORMATTED my_table").show(false)

常见问题解决

// 如果遇到类路径问题，添加依赖
// 在 spark-submit 时包含 Hive 相关 JAR
spark-submit --jars $(echo /path/to/hive/lib/*.jar | tr ' ' ',') \
             --class MainClass app.jar

// 或者使用 --packages
spark-submit --packages org.apache.spark:spark-hive_2.12:3.4.0

通过以上配置和操作，Spark SQL 可以无缝集成 Hive，实现对 Hive 表的完整 CRUD 操作，同时利用 Spark 的性能优势进行高效的数据处理。