Azure HDInsight'ta Hive Ambarı Bağlayıcı API'leri

Bu makalede Hive ambarı bağlayıcısı tarafından desteklenen tüm API'ler listelenmektedir. Aşağıda gösterilen tüm örnekler spark-shell ve hive ambarı bağlayıcı oturumu kullanılarak çalıştırılır.

Hive ambarı bağlayıcı oturumu oluşturma:

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

Önkoşul

Hive Ambarı Bağlayıcısı kurulum adımlarını tamamlayın.

Desteklenen API'ler

• Veritabanını ayarlayın:

  hive.setDatabase("<database-name>")
  

• Tüm veritabanlarını listeleyin:

  hive.showDatabases()
  

• Geçerli veritabanındaki tüm tabloları listeleme

  hive.showTables()
  

• Tabloyu açıklama

  // 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>")
  

• Veritabanını bırakma

  // ifExists and cascade are boolean variables
  hive.dropDatabase("<database-name>", ifExists, cascade)
  

• Geçerli veritabanına tablo bırakma

  // ifExists and purge are boolean variables
  hive.dropTable("<table-name>", ifExists, purge)
  

• Veritabanı oluşturma

  // ifNotExists is boolean variable
  hive.createDatabase("<database-name>", ifNotExists)
  

• Geçerli veritabanında tablo oluşturma

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

  Create-table oluşturucusu yalnızca aşağıdaki işlemleri destekler:

  // 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()
  

  Not

  Bu API, varsayılan konumda ORC biçimli bir tablo oluşturur. Diğer özellikler/seçenekler için veya hive sorgularını kullanarak tablo oluşturmak için API'yi kullanın executeUpdate .

• Tabloyu okuma

  // Returns a Dataset<Row> that contains data of <table-name> in the current database
  hive.table("<table-name>")
  

• HiveServer2'de DDL komutları yürütme

  // 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 sorgusunu yürütme ve Veri Kümesinde sonuç yükleme

  • LLAP daemon'ları aracılığıyla sorgu yürütülüyor. [Önerilen]

    // <hive-query> should be a hive query 
    hive.executeQuery("<hive-query>")
    

  • Sorguyu JDBC aracılığıyla HiveServer2 aracılığıyla yürütme.

    Bu API'yi kullanmak için false Spark oturumuna başlamadan önce Spark yapılandırmalarında olarak ayarlayın spark.datasource.hive.warehouse.smartExecution

    hive.execute("<hive-query>")
    

• Hive ambarı bağlayıcısı oturumlarını kapatma

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

• Hive Birleştirme sorgusu yürütme

  Bu API, aşağıdaki biçimde bir Hive birleştirme sorgusu oluşturur

  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
  

  Oluşturucu aşağıdaki işlemleri destekler:

  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 Tablosuna Toplu Olarak Veri Kümesi Yazma

  df.write.format("com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector")
     .option("table", tableName)
     .mode(SaveMode.Type)
     .save()
  

  • TableName veya <table>biçiminde <db>.<table> olmalıdır. Veritabanı adı sağlanmazsa, tablo geçerli veritabanında aranacak/oluşturulur

  • SaveMode türleri şunlardır:

    • Ekleme: Veri kümesini verilen tabloya ekler

    • Üzerine yaz: Veri kümesiyle verilen tablodaki verilerin üzerine yazar

    • Yoksay: Tablo zaten varsa yazma atlar, hata oluşmaz

    • ErrorIfExists: Tablo zaten varsa hata oluşturur

• HiveStreaming kullanarak Hive Tablosuna Veri Kümesi Yazma

  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()
  

  Not

  Akış yazma işlemleri her zaman verileri ekler.

• Hive Tablosuna spark akışı yazma

  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()