Azure HDInsight 中的 Hive Warehouse Connector API

项目
06/02/2023

本文列出了 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()

Azure HDInsight 中的 Hive Warehouse Connector API

先决条件

受支持的 API

其他资源