管理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'