Spark基础：读写Hive

2020-07-01 00:00:00 创建配置默认属性相关

1 Hive相关的操作

Spark SQL支持读写Hive，不过Hive本身包含了大量的依赖，这些依赖spark默认是没有的。如果Hive的依赖在Classpath中，那么Spark可以自动加载（注意Spark的worker节点也需要提供这些依赖）。默认配置Hive只需要把相关的hive-site.xml core-site.xml hdfs-site.xml 放到conf目录下即可。

当使用hive时，需要在 SparkSession 中开启hive，从而获得hive相关的serdes以及函数。如果没有现成的Hive环境们也可以使用，spark会自动在当前目录创建metastore_db，目录的位置可以通过参数 spark.sql.warehouse.dir 指定，默认是启动Spark应用程序的目录。注意在spark2.0之前使用的参数hive.metastore.warehouse.dir属性，已经废弃。另外不要忘记赋予spark程序读写对应目录的权限。

// 创建spark session 并 指定hive地址
val spark = SparkSession
  .builder()
  .appName("Spark Hive Example")
  .config("spark.sql.warehouse.dir", warehouseLocation)
  .enableHiveSupport()
  .getOrCreate()

import spark.implicits._
import spark.sql

// 执行hive操作
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")

// 查询hive表
sql("SELECT * FROM src").show()

// 执行聚合操作
sql("SELECT COUNT(*) FROM src").show()

// sql转换DataFrame
val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")
val stringsDS = sqlDF.map {
  case Row(key: Int, value: String) => s"Key: $key, Value: $value"
}
stringsDS.show()

// 在sparksession中创建虚拟表
val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s"val_$i")))
recordsDF.createOrReplaceTempView("records")
sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()

// 使用hive命令创建表
sql("CREATE TABLE hive_records(key int, value string) STORED AS PARQUET")
val df = spark.table("src")
df.write.mode(SaveMode.Overwrite).saveAsTable("hive_records")
sql("SELECT * FROM hive_records").show()

// 创建hive外部表
val dataDir = "/tmp/parquet_data"
spark.range(10).write.parquet(dataDir)
sql(s"CREATE EXTERNAL TABLE hive_bigints(id bigint) STORED AS PARQUET LOCATION '$dataDir'")
sql("SELECT * FROM hive_bigints").show()

// 动态配置hive属性
spark.sqlContext.setConf("hive.exec.dynamic.partition", "true")
spark.sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
df.write
  .partitionBy("key")
  .format("hive")
  .saveAsTable("hive_part_tbl")

相关文章