E-MapReduce:Kyuubi Gateway の管理

トークンの管理

1. [Kyuubi Gateway] ページで、ターゲットゲートウェイを見つけ、[操作] 列の [トークン管理] をクリックします。

2. [トークンの作成] をクリックします。

3. [トークンの作成] ダイアログボックスで、パラメーターを設定し、[OK] をクリックします。

   パラメーター	説明
   名前	新しいトークンの名前。
   有効期限	トークンの有効期限を設定します。日数は 1 以上である必要があります。デフォルトでは、これは有効になっており、トークンは 365 日後に有効期限が切れます。
   オブジェクトの割り当て	説明 [データカタログ] タブでデフォルトで DLF (旧 DLF 2.5) を使用する場合は、このパラメーターを設定する必要があります。 設定された Resource Access Management (RAM) ユーザーまたは RAM ロールに DLF へのアクセス権限があることを確認してください。権限を付与する方法の詳細については、「権限付与の追加」をご参照ください。 [Resource Access Management] で追加した RAM ユーザーまたは RAM ロールをドロップダウンリストから選択します。 トークンが割り当てられる RAM ユーザーまたは RAM ロールを指定します。これは、Kyuubi Gateway に接続して Spark ジョブを送信する際に DLF にアクセスするために使用されます。

4. トークン情報をコピーします。

   重要

   トークンは作成後にすぐにコピーする必要があります。後で取得することはできません。トークンの有効期限が切れたり、紛失したりした場合は、新しいトークンを作成するか、トークンをリセットする必要があります。

Kyuubi Gateway に接続する際、JDBC URL のプレースホルダーを置き換えます:

• <endpoint>: [概要] タブから取得できるエンドポイント情報。

• <port>: ポート番号。ポート番号は、パブリックエンドポイントの場合は 443、内部の同一リージョンエンドポイントの場合は 80 です。

• <token>: [トークン管理] ページからコピーしたトークン情報。

• <tokenname>: トークン名。[トークン管理] ページから取得できます。

• <UserName/RoleName>: [Resource Access Management] に追加した RAM ユーザーまたは RAM ロール。

Beeline を使用した接続

Kyuubi Gateway に接続する前に、Beeline のバージョンが Kyuubi サーバーのバージョンと互換性があることを確認してください。Beeline がインストールされていない場合は、「Getting Started - Apache Kyuubi」をご参照ください。

[データカタログ] ページで設定されているデフォルトのカタログに基づいて、次のいずれかの方法を選択します。

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>

• Kyuubi Gateway に接続するための Java コードを記述します。

  [データカタログ] ページで設定されているデフォルトのカタログに基づいて、次のいずれかの方法を選択します。

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

  別のカタログの使用

  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. Kyuubi Gateway に接続するための Python スクリプトを記述します。

   次の Python スクリプトは、Kyuubi Gateway に接続してデータベースのリストを表示する方法の例です。

   [データカタログ] ページで設定されているデフォルトのカタログに基づいて、次のいずれかの方法を選択します。

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

   別のカタログの使用

   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 は、HTTP を介した Kyuubi サービスとの対話をサポートする、オープンソース互換の Representational State Transfer (REST) API を提供します。現在、次の API パスのみがサポートされています:

• /api/v1/sessions/*

• /api/v1/operations/*

• /api/v1/batches/*

次の例は、REST API を使用して Kyuubi Gateway に接続する方法を示しています。

• 例 1: セッションを開始して SQL クエリを実行する。

  1. セッションを作成し、Spark 構成を指定します。

     [データカタログ] ページで設定されているデフォルトのカタログに基づいて、次のいずれかの方法を選択します。

     説明

     • 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>"
       }
     }'

     別のカタログの使用

     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 セッションハンドルを示します。このトピックでは、この値を <sessionHandle> と呼びます。

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

  2. 文を作成します。

     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"
       }
     }'

     別のカタログの使用

     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 操作ハンドルを示します。このトピックでは、この値を <operationHandle> と呼びます。

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

  3. 文のステータスを取得します。

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

     別のカタログの使用

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

  4. 文の結果を取得します。

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

     別のカタログの使用

     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 API を使用してバッチジョブを送信する。

  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'

  別のカタログの使用

  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'