Spark SQL操作Hive表

Spark SQL支持从Hive存储中读写数据。然而，Hive存在很多的依赖，而这些依赖又不包含在默认的各类Spark发型版本中。如果将Hive的依赖放入classpath中，Spark将自动加载它们。值得注意的是，这些依赖必须在所有节点中都存在。因为他们需要通过Hive的序列化和反序列化库(SerDes)来访问存储在Hive中的数据。

在Spark中配置Hive，需要将hive-site.xml, core-site.xml, hdfs-site.xml放置到Spark的conf/目录下。

需要操作Hive，必须先实例化SparkSession以获取Hive的支持。包括连接到一个持久化的Hive元存储，支持Hive的序列化以及Hive的UDF。没有部署Hive的用户仍然可以启用Hive支持。如果在hive-site.xml中没有进行配置，上下文将在当前目录自动创建metastore_db，并创建一个由spark.sql.warehouse.dir配置的目录。默认为Spark应用程序启动的当前目录下的Spark -warehouse目录。注意：hive-site.xml中的hive.metastore.warehouse.dir配置项在Spark 2.0.0以后被启用，取而代之的是spark.sql.warehouse.dir，以指定数据库在仓库中的默认位置。使用时，需要将写权限授予启动Spark应用程序的用户。

import java.io.File

import org.apache.spark.sql.{Row, SaveMode, SparkSession}

case class Record(key: Int, value: String)

// warehouseLocation指的是管理数据库和表的默认路径地址
val warehouseLocation = new File("spark-warehouse").getAbsolutePath

val spark = SparkSession
  .builder()
  .appName("Spark Hive Example")
  .config("spark.sql.warehouse.dir", warehouseLocation)
  .enableHiveSupport()
  .getOrCreate()

import spark.implicits._
import spark.sql

sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive")
sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")

// HiveQL
sql("SELECT * FROM src").show()
// +---+-------+
// |key|  value|
// +---+-------+
// |238|val_238|
// | 86| val_86|
// |311|val_311|
// ...

// 支持聚合查询。
sql("SELECT COUNT(*) FROM src").show()
// +--------+
// |count(1)|
// +--------+
// |    500 |
// +--------+

// SQL查询的结果本身就是DataFrame，支持所有普通函数。
val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")

// DataFrame中的项是Row类型的，可以按顺序访问每个列。
val stringsDS = sqlDF.map {
  case Row(key: Int, value: String) => s"Key: $key, Value: $value"
}
stringsDS.show()
// +--------------------+
// |               value|
// +--------------------+
// |Key: 0, Value: val_0|
// |Key: 0, Value: val_0|
// |Key: 0, Value: val_0|
// ...

// 在SparkSession中使用DataFrame创建临时视图。
val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s"val_$i")))
recordsDF.createOrReplaceTempView("records")

// 使用查询来连接DataFrame中的数据和Hive中的表
sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()
// +---+------+---+------+
// |key| value|key| value|
// +---+------+---+------+
// |  2| val_2|  2| val_2|
// |  4| val_4|  4| val_4|
// |  5| val_5|  5| val_5|
// ...

// 创建Hive管理的Parquet表，使用HQL语法，而不是Spark SQL的原生语法
// 使用hive
sql("CREATE TABLE hive_records(key int, value string) STORED AS PARQUET")

// 保存DataFrame到Hive管理表中
val df = spark.table("src")
df.write.mode(SaveMode.Overwrite).saveAsTable("hive_records")
// 插入数据
sql("SELECT * FROM hive_records").show()
// +---+-------+
// |key|  value|
// +---+-------+
// |238|val_238|
// | 86| val_86|
// |311|val_311|
// ...

// 准备一个Parquet数据目录
val dataDir = "/tmp/parquet_data"
spark.range(10).write.parquet(dataDir)

// 创建一个Hive外部表（Parquet格式）
sql(s"CREATE EXTERNAL TABLE hive_bigints(id bigint) STORED AS PARQUET LOCATION '$dataDir'")

// Hive外部表应该已经有数据了
sql("SELECT * FROM hive_bigints").show()
// +---+
// | id|
// +---+
// |  0|
// |  1|
// |  2|
// ... Order may vary, as spark processes the partitions in parallel.

// 开启Hive动态分区的标志
spark.sqlContext.setConf("hive.exec.dynamic.partition", "true")
spark.sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
// 使用DataFrame API创建Hive分区表
df.write.partitionBy("key").format("hive").saveAsTable("hive_part_tbl")
// 分区列' key '将被移到schema的末尾。
sql("SELECT * FROM hive_part_tbl").show()
// +-------+---+
// |  value|key|
// +-------+---+
// |val_238|238|
// | val_86| 86|
// |val_311|311|
// ...

spark.stop()

指定Hive表的存储格式

创建Hive表时，需定义该表从文件系统读/写以及文件系统从该表读/写数据的方式，例如：写入格式和读出格式。还需要定义这个表应该如何将数据反序列化为行，或将行序列化为数据。接下来的选项可以被用来指定存储的格式，比如：CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。默认的，将以纯文本的形式读取表文件。注意：创建表时不支持Hive存储处理程序，可以通过在Hive端使用存储处理程序来创建表，然后使用Spark SQL去读取。