E-MapReduce：管理Kyuubi Gateway

管理Token

1. 在Kyuubi Gateway頁面，單擊目標Gateway操作列的Token管理。

2. 單擊建立Token。

3. 在建立Token對話方塊中，配置以下資訊，單擊確定。

   參數	說明
   名稱	建立Token的名稱。
   到期時間	設定該Token的到期時間。設定的天數應大於或等於1。預設情況下為開啟狀態，365天后到期。
   指派至	說明 如果您在資料目錄頁簽，預設使用的是DLF（之前稱為DLF 2.5），則需要配置該參數。 需確保配置的RAM使用者或RAM角擁有訪問DLF的許可權，授權操作請參見新增授權。 在下拉框中，選擇您在存取控制中添加的RAM使用者或者RAM角色。 指定分配Token的RAM使用者或RAM角色，用於在串連Kyuubi Gateway提交Spark任務時訪問DLF。

4. 複製Token資訊。

   重要

   Token建立完成後，請務必立即複製新Token的資訊，後續不支援查看。如果您的Token到期或遺失，請選擇建立Token或重設Token。

在串連Kyuubi Gateway時，請根據您的實際情況替換JDBC URL中的資訊：

• <endpoint>：您在總覽頁簽擷取的Endpoint資訊。

• <port>：連接埠號碼。使用外網網域名稱訪問時連接埠號碼為443，使用內網網域名稱訪問時連接埠號碼為80。

• <token>：您在Token管理頁簽複製的Token資訊。

• <tokenname>：Token名稱。您可以在Token管理頁簽擷取。

• <UserName/RoleName>：您在存取控制中添加的RAM使用者或RAM角色。

使用Beeline串連

在串連Kyuubi Gateway時，請確保您使用的Beeline版本與Kyuubi服務端版本相容。如果您未安裝Beeline，詳情請參見Getting Started - Apache Kyuubi。

請根據資料目錄頁面Catalog的預設使用方式，選擇以下對應的操作方式。

beeline -u "jdbc:hive2://<endpoint>:<port>/;transportMode=http;user=<UserName/RoleName>;httpPath=cliservice/token/<token>"

beeline -u "jdbc:hive2://<endpoint>:<port>/;transportMode=http;httpPath=cliservice/token/<token>"

使用Beeline串連時支援修改會話參數，例如beeline -u "jdbc:hive2://<endpoint>:<port>/;transportMode=http;httpPath=cliservice/token/<token>;#spark.sql.shuffle.partitions=100;spark.executor.instances=2;"。

使用Java串連

• 更新pom.xml。

  使用適當版本的依賴項以替換hadoop-common和hive-jdbc。

  <dependencies>
          <dependency>
              <groupId>org.apache.hadoop</groupId>
              <artifactId>hadoop-common</artifactId>
              <version>3.0.0</version>
          </dependency>
          <dependency>
              <groupId>org.apache.hive</groupId>
              <artifactId>hive-jdbc</artifactId>
              <version>2.3.9</version>
          </dependency>
      </dependencies>

• 編寫Java代碼，串連Kyuubi Gateway。

  請根據資料目錄頁面Catalog的預設使用方式，選擇以下對應的操作方式。

  使用DLF（之前稱為DLF 2.5）

  import org.apache.hive.jdbc.HiveStatement;
  import java.sql.Connection;
  import java.sql.DriverManager;
  import java.sql.ResultSet;
  import java.sql.ResultSetMetaData;
  
  public class Main {
      public static void main(String[] args) throws Exception {
          String url = "jdbc:hive2://<endpoint>:<port>/;transportMode=http;httpPath=cliservice/token/<token>;user=<UserName/RoleName>";
          Class.forName("org.apache.hive.jdbc.HiveDriver");
          Connection conn = DriverManager.getConnection(url);
          HiveStatement stmt = (HiveStatement) conn.createStatement();
  
  
          String sql = "select * from students;";
          System.out.println("Running " + sql);
          ResultSet res = stmt.executeQuery(sql);
  
          ResultSetMetaData md = res.getMetaData();
          String[] columns = new String[md.getColumnCount()];
          for (int i = 0; i < columns.length; i++) {
              columns[i] = md.getColumnName(i + 1);
          }
          while (res.next()) {
              System.out.print("Row " + res.getRow() + "=[");
              for (int i = 0; i < columns.length; i++) {
                  if (i != 0) {
                      System.out.print(", ");
                  }
                  System.out.print(columns[i] + "='" + res.getObject(i + 1) + "'");
              }
              System.out.println(")]");
          }
  
          conn.close();
      }
  }

  使用其他Catalog

  import org.apache.hive.jdbc.HiveStatement;
  import java.sql.Connection;
  import java.sql.DriverManager;
  import java.sql.ResultSet;
  import java.sql.ResultSetMetaData;
  
  public class Main {
      public static void main(String[] args) throws Exception {
          String url = "jdbc:hive2://<endpoint>:<port>/;transportMode=http;httpPath=cliservice/token/<token>";
          Class.forName("org.apache.hive.jdbc.HiveDriver");
          Connection conn = DriverManager.getConnection(url);
          HiveStatement stmt = (HiveStatement) conn.createStatement();
  
  
          String sql = "select * from students;";
          System.out.println("Running " + sql);
          ResultSet res = stmt.executeQuery(sql);
  
          ResultSetMetaData md = res.getMetaData();
          String[] columns = new String[md.getColumnCount()];
          for (int i = 0; i < columns.length; i++) {
              columns[i] = md.getColumnName(i + 1);
          }
          while (res.next()) {
              System.out.print("Row " + res.getRow() + "=[");
              for (int i = 0; i < columns.length; i++) {
                  if (i != 0) {
                      System.out.print(", ");
                  }
                  System.out.print(columns[i] + "='" + res.getObject(i + 1) + "'");
              }
              System.out.println(")]");
          }
  
          conn.close();
      }
  }

使用Python串連

1. 執行以下命令，安裝PyHive和Thrift包。

   pip3 install pyhive thrift

2. 編寫Python指令碼，串連Kyuubi Gateway。

   以下是一個Python指令碼樣本，展示如何串連到Kyuubi Gateway並顯示資料庫列表。

   請根據資料目錄頁面Catalog的預設使用方式，選擇以下對應的操作方式。

   使用DLF（之前稱為DLF 2.5）

   from pyhive import hive
   
   if __name__ == '__main__':
       cursor = hive.connect('<endpoint>',
                             port="<port>",
                             scheme='http',
                             username='<UserName/RoleName>',
                             password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

   使用其他Catalog

   from pyhive import hive
   
   if __name__ == '__main__':
       cursor = hive.connect('<endpoint>',
                             port="<port>",
                             scheme='http',
                             username='<tokenname>',
                             password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

使用REST API串連

Kyuubi Gateway提供了開源相容的REST API，支援通過HTTP協議與Kyuubi服務進行互動。目前僅支援以下API路徑：

• /api/v1/sessions/*

• /api/v1/operations/*

• /api/v1/batches/*

本文通過以下樣本為您介紹如何使用REST API串連Kyuubi Gateway。

• 樣本1：啟動Session並進行SQL查詢。

  1. 建立Session並指定Spark配置。

     請根據資料目錄頁面Catalog的預設使用方式，選擇以下對應的操作方式。

     說明

     • spark.emr.serverless.kyuubi.engine.queue用於指定Spark任務運行時所使用的隊列。請根據實際情況替換<dev_queue>為具體的隊列名。

     • <UserName/Rolename>：替換為實際的使用者名稱或角色名稱。

     • <password>：僅作為預留位置，可填寫任意值。

     使用DLF（之前稱為DLF 2.5）

     curl -X 'POST' \
       'http://<endpoint>:<port>/api/v1/sessions/token/<token>' \
       -H 'accept: application/json' \
       -H 'Content-Type: application/json' \
       -u '<UserName/Rolename>:<password>' \
       -d '{
       "configs": {
         "set:hivevar:spark.emr.serverless.kyuubi.engine.queue": "<dev_queue>"
       }
     }'

     使用其他Catalog

     curl -X 'POST' \
       'http://<endpoint>:<port>/api/v1/sessions/token/<token>' \
       -H 'accept: application/json' \
       -H 'Content-Type: application/json' \
       -d '{
       "configs": {
         "set:hivevar:spark.emr.serverless.kyuubi.engine.queue": "<dev_queue>"
       }
     }'

     返回如下類似資訊。其中，identifier表示Kyuubi的Session Handle，用於唯一標識一個會話。本文中將以<sessionHandle>指代該值。

     {"identifier":"619e6ded-xxxx-xxxx-xxxx-c2a43f6fac46","kyuubiInstance":"0.0.0.0:10099"}

  2. 建立Statement。

     使用DLF（之前稱為DLF 2.5）

     curl -X 'POST' \
       'http://<endpoint>:<port>/api/v1/sessions/<sessionHandle>/operations/statement/token/<token>' \
       -H 'accept: application/json' \
       -H 'Content-Type: application/json' \
       -u '<UserName/RoleName>:<password>' \
       -d '{
       "statement": "select * from test;",
       "runAsync": true,
       "queryTimeout": 0,
       "confOverlay": {
         "additionalProp1": "string",
         "additionalProp2": "string"
       }
     }'

     使用其他Catalog

     curl -X 'POST' \
       'http://<endpoint>:<port>/api/v1/sessions/<sessionHandle>/operations/statement/token/<token>' \
       -H 'accept: application/json' \
       -H 'Content-Type: application/json' \
       -d '{
       "statement": "select * from test;",
       "runAsync": true,
       "queryTimeout": 0,
       "confOverlay": {
         "additionalProp1": "string",
         "additionalProp2": "string"
       }
     }'

     返回如下類似資訊。這裡的identifier表示Kyuubi的Operation Handle，用於唯一標識一個具體的操作。本文中將以<operationHandle>指代該值。

     {"identifier":"a743e8ff-xxxx-xxxx-xxxx-a66fec66cfa4"}

  3. 擷取Statement狀態。

     使用DLF（之前稱為DLF 2.5）

     curl --location -X 'GET' \
       'http://<endpoint>:<port>/api/v1/operations/<operationHandle>/event/token/<token>' \
       -H 'accept: application/json' \
       -u '<UserName/RoleName>:<password>'

     使用其他Catalog

     curl --location -X 'GET' \
       'http://<endpoint>:<port>/api/v1/operations/<operationHandle>/event/token/<token>' \
       -H 'accept: application/json'

  4. 擷取Statement結果。

     使用DLF（之前稱為DLF 2.5）

     curl --location -X 'GET' \
       'http://<endpoint>:<port>/api/v1/operations/<operationHandle>/rowset/token/<token>/?maxrows=100&fetchorientation=FETCH_NEXT' \
       -H 'accept: application/json'  \
       -u '<UserName/RoleName>:<password>'

     使用其他Catalog

     curl --location -X 'GET' \
       'http://<endpoint>:<port>/api/v1/operations/<operationHandle>/rowset/token/<token>/?maxrows=100&fetchorientation=FETCH_NEXT' \
       -H 'accept: application/json'

• 樣本2：使用batches介面提交批作業。

  通過REST API提交一個Spark批處理任務到Kyuubi Gateway。Kyuubi Gateway會根據請求中的參數啟動一個Spark應用程式，並執行指定的任務。

  在本樣本中，除過替換<endpoint>、<port>、<token>等資訊外，還需單擊spark-examples_2.12-3.3.1.jar，直接下載測試JAR包。

  說明

  該JAR包是Spark內建的一個簡單樣本，用於計算圓周率π的值。

  使用DLF（之前稱為DLF 2.5）

  curl --location \
    --request POST 'http://<endpoint>:<port>/api/v1/batches/token/<token>' \
    --user '<UserName/RoleName>:<password>' \
    --form 'batchRequest="{
      \"batchType\": \"SPARK\",
      \"className\": \"org.apache.spark.examples.SparkPi\",
      \"name\": \"kyuubi-spark-pi\",
      \"resource\": \"oss://bucket/path/to/spark-examples_2.12-3.3.1.jar\"
    }";type=application/json'

  使用其他Catalog

  curl --location \
    --request POST 'http://<endpoint>:<port>/api/v1/batches/token/<token>' \
    --form 'batchRequest="{
      \"batchType\": \"SPARK\",
      \"className\": \"org.apache.spark.examples.SparkPi\",
      \"name\": \"kyuubi-spark-pi\",
      \"resource\": \"oss://bucket/path/to/spark-examples_2.12-3.3.1.jar\"
    }";type=application/json'