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'

使用Apache Superset串連

Apache Superset是一個現代資料探索和可視化平台，具有豐富的圖表形態。更多Superset資訊，請參見Superset。

pip install thrift==0.20.0

1. 啟動Superset，進入Superset介面。更多啟動操作資訊，請參見Superset文檔。

2. 在頁面右上方單擊DATABASE，進入Connect a database頁面。

3. 在Connect a database頁面，選擇Apache Spark SQL。

4. 在SQLAlchemy URI欄位，根據訪問方式填寫對應的串連串。

   公網訪問

   hive+https://<username>:<token>@<endpoint>:443/<db_name>

   VPC內網訪問

   hive+http://<username>:<token>@<endpoint>:80/<db_name>

   參數說明如下：

   參數	說明
   username	使用者名稱，可填寫任意值。
   Token	訪問Token。
   Endpoint	Kyuubi Gateway的訪問端點，需要區分公網endpoint與內網endpoint。
   db_name	資料庫名稱。

5. 單擊Connect完成串連配置。

使用HUE串連

pip install thrift==0.20.0

1. 編輯Hue設定檔/etc/hue/hue.conf，根據您的訪問方式添加以下配置。

   公網訪問

   [[[sparksql]]]
        name = Spark Sql
        interface=sqlalchemy
        options='{"url": "hive+https://<username>:<token>@<endpoint>:443/"}'

   VPC內網訪問

   [[[sparksql]]]
        name = Spark Sql
        interface=sqlalchemy
        options='{"url": "hive+http://<username>:<token>@<endpoint>:80/"}'

   參數說明如下：

   參數	說明
   username	使用者名稱，可填寫任意值。
   Token	訪問Token。
   Endpoint	Kyuubi Gateway的訪問端點。

2. 重啟Hue服務。

   sudo service hue restart

3. 登入Hue Web UI，在左側導覽列選擇Query > Editor > SparkSql，驗證串連是否成功。

使用DataGrip串連

1. 開啟DataGrip，建立新專案。

2. 在右側Database Explorer面板中，單擊+ > Data Source > Other > Apache Spark。

3. 在Data Sources and Drivers對話方塊中，配置以下參數。

   參數	說明
   Name	自訂串連名稱。
   Authentication	選擇No auth或User & Password（使用者名稱可任意填寫）。
   Driver	選擇Apache Spark ver. 1.2.2。
   URL	公網訪問 jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token> VPC內網訪問 jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token> 其中，<endpoint>為Kyuubi Gateway的訪問端點，<token>為訪問Token。

4. 單擊Test Connection驗證串連。

5. 單擊OK儲存配置。