Edit

Hive Warehouse Connector 2.1 and 2.0 APIs in Azure HDInsight

This article lists all the APIs supported by Hive warehouse connector 2.1 and 2.0. All the examples shown are how to run using spark-shell and hive warehouse connector session.

How to create Hive warehouse connector session:

import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()

Prerequisite

Complete the Hive Warehouse Connector setup steps.

Supported APIs

  • Set the database:

    hive.setDatabase("")

  • List all databases:

    hive.showDatabases()

  • List all tables in the current database

    hive.showTables()

  • Describe a table

    // Describes the table  in the current database
hive.describeTable("")

    // Describes the table  in 
hive.describeTable(".")

  • Drop a database

    // ifExists and cascade are boolean variables
hive.dropDatabase("", ifExists, cascade)

  • Drop a table in the current database

    // ifExists and purge are boolean variables
hive.dropTable("", ifExists, purge)

  • Create a database

    // ifNotExists is boolean variable
hive.createDatabase("", ifNotExists)

  • Create a table in current database

    // Returns a builder to create table
val createTableBuilder = hive.createTable("")

    Builder for create-table supports only the below operations:

    // Create only if table does not exists already
createTableBuilder = createTableBuilder.ifNotExists()

    // Add columns
createTableBuilder = createTableBuilder.column("", "")

    // Add partition column
createTableBuilder = createTableBuilder.partition("", "")

    // Add table properties
createTableBuilder = createTableBuilder.prop("", "")

    // Creates a bucketed table,
// Parameters are numOfBuckets (integer) followed by column names for bucketing
createTableBuilder = createTableBuilder.clusterBy(numOfBuckets, "", .... , "")

    // Creates the table
createTableBuilder.create()

    Note

    This API creates an ORC formatted table at default location. For other features/options or to create table using hive queries, use executeUpdate API.

  • Read a table

    // Returns a Dataset that contains data of  in the current database
hive.table("")

  • Execute DDL commands on HiveServer2

    // Executes the  against HiveServer2
// Returns true or false if the query succeeded or failed respectively
hive.executeUpdate("")

    // Executes the  against HiveServer2
// Throws exception, if propagateException is true and query threw exception in HiveServer2
// Returns true or false if the query succeeded or failed respectively
hive.executeUpdate("", propagateException) // propagate exception is boolean value

  • Execute Hive query and load result in Dataset

    • Executing query via LLAP daemons. [Recommended]

      //  should be a hive query 
hive.executeQuery("")

    • Executing query through HiveServer2 via JDBC.

      Set spark.datasource.hive.warehouse.smartExecution to false in spark configs before starting spark session to use this API

      hive.execute("")

  • Close Hive warehouse connector session

    // Closes all the open connections and
// release resources/locks from HiveServer2
hive.close()

  • Execute Hive Merge query

    This API creates a Hive merge query in the format

    MERGE INTO . AS  USING  AS 
ON 
WHEN MATCHED [AND ] THEN UPDATE SET  ... 
WHEN MATCHED [AND ] THEN DELETE
WHEN NOT MATCHED [AND ] THEN INSERT VALUES  ... 

    val mergeBuilder = hive.mergeBuilder() // Returns a builder for merge query

    Builder supports the following operations:

    mergeBuilder.mergeInto("", "")

    mergeBuilder.using("", "")

    mergeBuilder.on("")

    mergeBuilder.whenMatchedThenUpdate("", "", ... , "")

    mergeBuilder.whenMatchedThenDelete("")

    mergeBuilder.whenNotMatchedInsert("", "", ... , "");

    // Executes the merge query
mergeBuilder.merge()

  • Write a Dataset to Hive Table in batch

    df.write.format("com.microsoft.hwc.v2")
   .option("table", tableName)
   .mode(SaveMode.Type)
   .save()

    • TableName should be of form .

      or
      . If no database name is provided, the table will searched/created in the current database

    • SaveMode types are:

      • Append: Appends the dataset to the given table

      • Overwrite: Overwrites the data in the given table with dataset

      • Ignore: Skips write if table already exists, no error thrown

      • ErrorIfExists: Throws error if table already exists

    • Write a Dataset to Hive Table using HiveStreaming

      df.write.format("com.microsoft.hwc.v2.batch.stream.write")
   .option("database", databaseName)
   .option("table", tableName)
   .option("metastoreUri", "")
// .option("metastoreKrbPrincipal", principal), add if executing in ESP cluster
   .save()

 // To write to static partition
 df.write.format("com.microsoft.hwc.v2.batch.stream.write")
   .option("database", databaseName)
   .option("table", tableName)
   .option("partition", partition)
   .option("metastoreUri", "")
// .option("metastoreKrbPrincipal", principal), add if executing in ESP cluster
   .save()

      Note

      Stream writes always append data.

    • Writing a spark stream to a Hive Table

      stream.writeStream
    .format("com.microsoft.hwc.v2")
    .option("metastoreUri", "")
    .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()

      • Next steps

      • Last updated on