Azure HDInsight 中的 Hive Warehouse Connector API
本文列出了 Hive Warehouse Connector 支持的所有 API。 下面显示的所有示例都使用 spark-shell 和 Hive Warehouse Connector 会话运行。
如何创建 Hive Warehouse Connector 会话:
import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()
先决条件
完成 Hive Warehouse Connector 设置步骤。
受支持的 API
设置数据库:
hive.setDatabase("<database-name>")
列出所有数据库:
hive.showDatabases()
列出当前数据库中的所有表
hive.showTables()
描述一个表
// Describes the table <table-name> in the current database hive.describeTable("<table-name>")
// Describes the table <table-name> in <database-name> hive.describeTable("<database-name>.<table-name>")
删除数据库
// ifExists and cascade are boolean variables hive.dropDatabase("<database-name>", ifExists, cascade)
删除当前数据库中的一个表
// ifExists and purge are boolean variables hive.dropTable("<table-name>", ifExists, purge)
创建数据库
// ifNotExists is boolean variable hive.createDatabase("<database-name>", ifNotExists)
在当前数据库中创建一个表
// Returns a builder to create table val createTableBuilder = hive.createTable("<table-name>")
create-table 的生成器仅支持以下操作:
// Create only if table does not exists already createTableBuilder = createTableBuilder.ifNotExists()
// Add columns createTableBuilder = createTableBuilder.column("<column-name>", "<datatype>")
// Add partition column createTableBuilder = createTableBuilder.partition("<partition-column-name>", "<datatype>")
// Add table properties createTableBuilder = createTableBuilder.prop("<key>", "<value>")
// Creates a bucketed table, // Parameters are numOfBuckets (integer) followed by column names for bucketing createTableBuilder = createTableBuilder.clusterBy(numOfBuckets, "<column1>", .... , "<columnN>")
// Creates the table createTableBuilder.create()
注意
此 API 在默认位置创建 ORC 格式的表。 如需其他功能/选项或者要使用 hive 查询来创建表,请使用
executeUpdate
API。读取表
// Returns a Dataset<Row> that contains data of <table-name> in the current database hive.table("<table-name>")
在 HiveServer2 上执行 DDL 命令
// Executes the <hive-query> against HiveServer2 // Returns true or false if the query succeeded or failed respectively hive.executeUpdate("<hive-query>")
// Executes the <hive-query> against HiveServer2 // Throws exception, if propagateException is true and query threw excpetion in HiveServer2 // Returns true or false if the query succeeded or failed respectively hive.executeUpdate("<hive-query>", propagateException) // propagate exception is boolean value
执行 Hive 查询并在数据集中加载结果
通过 LLAP 守护程序执行查询。 [建议]
// <hive-query> should be a hive query hive.executeQuery("<hive-query>")
使用 JDBC 通过 HiveServer2 执行查询。
在启动 Spark 会话以使用此 API 之前,在 Spark 配置中将
spark.datasource.hive.warehouse.smartExecution
设置为false
hive.execute("<hive-query>")
关闭 Hive Warehouse Connector 会话
// Closes all the open connections and // release resources/locks from HiveServer2 hive.close()
执行 Hive 合并查询
此 API 创建以下格式的 Hive 合并查询
MERGE INTO <current-db>.<target-table> AS <targetAlias> USING <source expression/table> AS <sourceAlias> ON <onExpr> WHEN MATCHED [AND <updateExpr>] THEN UPDATE SET <nameValuePair1> ... <nameValuePairN> WHEN MATCHED [AND <deleteExpr>] THEN DELETE WHEN NOT MATCHED [AND <insertExpr>] THEN INSERT VALUES <value1> ... <valueN>
val mergeBuilder = hive.mergeBuilder() // Returns a builder for merge query
生成器支持以下操作:
mergeBuilder.mergeInto("<taget-table>", "<targetAlias>")
mergeBuilder.using("<source-expression/table>", "<sourceAlias>")
mergeBuilder.on("<onExpr>")
mergeBuilder.whenMatchedThenUpdate("<updateExpr>", "<nameValuePair1>", ... , "<nameValuePairN>")
mergeBuilder.whenMatchedThenDelete("<deleteExpr>")
mergeBuilder.whenNotMatchedInsert("<insertExpr>", "<value1>", ... , "<valueN>");
// Executes the merge query mergeBuilder.merge()
将数据集批量写入 Hive 表
df.write.format("com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector") .option("table", tableName) .mode(SaveMode.Type) .save()
TableName 的格式应为
<db>.<table>
或<table>
。 如果未提供数据库名称,将在当前数据库中搜索/创建表SaveMode 类型为:
Append:将数据集追加到给定表
Overwrite:使用数据集覆盖给定表中的数据
Ignore:如果表已存在,则跳过写入操作,不引发任何错误
ErrorIfExists:如果表已存在,则引发错误
使用 HiveStreaming 将数据集写入 Hive 表
df.write.format("com.hortonworks.spark.sql.hive.llap.HiveStreamingDataSource") .option("database", databaseName) .option("table", tableName) .option("metastoreUri", "<HMS_URI>") // .option("metastoreKrbPrincipal", principal), add if executing in ESP cluster .save() // To write to static partition df.write.format("com.hortonworks.spark.sql.hive.llap.HiveStreamingDataSource") .option("database", databaseName) .option("table", tableName) .option("partition", partition) .option("metastoreUri", "<HMS URI>") // .option("metastoreKrbPrincipal", principal), add if executing in ESP cluster .save()
注意
流写入始终追加数据。
将 Spark 流写入 Hive 表
stream.writeStream .format("com.hortonworks.spark.sql.hive.llap.streaming.HiveStreamingDataSource") .option("metastoreUri", "<HMS_URI>") .option("database", databaseName) .option("table", tableName) //.option("partition", partition) , add if inserting data in partition //.option("metastoreKrbPrincipal", principal), add if executing in ESP cluster .start()