E-MapReduce:Spark Thrift Server セッションの管理

Python を使用して Spark Thrift Server に接続する

1. 次のコマンドを実行して、PyHive パッケージと Thrift パッケージをインストールします。

   pip install pyhive thrift

2. Spark Thrift Server に接続するための Python スクリプトを作成します。

   以下は、Hive に接続してデータベースのリストを表示する方法を示す Python スクリプトの例です。ネットワーク環境（パブリックまたは内部）に基づいて、適切な接続方法を選択できます。

   パブリックエンドポイントを使用して接続する

   from pyhive import hive
   
   if __name__ == '__main__':
       # <endpoint>、<username>、および <token> を実際の情報に置き換えます。
       cursor = hive.connect('<endpoint>', port=443, scheme='https', username='<username>', password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

   内部の同一リージョンエンドポイントを使用して接続する

   from pyhive import hive
   
   if __name__ == '__main__':
       # <endpoint>、<username>、および <token> を実際の情報に置き換えます。
       cursor = hive.connect('<endpoint>', port=80, scheme='http', username='<username>', password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

Java を使用して Spark Thrift Server に接続する

1. 次の Maven 依存関係を pom.xml ファイルに追加します。

   <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.1.0</version>
           </dependency>
       </dependencies>
   

   説明

   Serverless Spark の組み込み Hive バージョンは 2.x です。したがって、hive-jdbc 2.x バージョンのみがサポートされています。

2. Spark Thrift Server に接続するための Java コードを作成します。

   以下は、Spark Thrift Server に接続してデータベースリストを照会する Java コードのサンプルです。

   パブリックエンドポイントを使用して接続する

   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.sql.ResultSet;
   import java.sql.ResultSetMetaData;
   import org.apache.hive.jdbc.HiveStatement;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           String url = "jdbc:hive2://<endpoint>:443/;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 = "show databases";
           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 java.sql.Connection;
   import java.sql.DriverManager;
   import java.sql.ResultSet;
   import java.sql.ResultSetMetaData;
   import org.apache.hive.jdbc.HiveStatement;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           String url = "jdbc:hive2://<endpoint>:80/;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 = "show databases";
           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();
       }
   }

Spark Beeline を使用して Spark Thrift Server に接続する

• セルフビルドクラスタを使用している場合は、最初に Spark の bin ディレクトリに移動してから、beeline を使用して Spark Thrift Server に接続する必要があります。

  パブリックエンドポイントを使用して接続する

  cd /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3/bin/
  
  ./beeline -u "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>"

  内部の同一リージョンエンドポイントを使用して接続する

  cd /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3/bin/
  
  ./beeline -u "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>"

  説明

  このコードは、ECS クラスタ上の EMR の Spark インストールパスの例として /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3 を使用しています。クライアントの実際の Spark インストールパスに基づいてパスを調整する必要があります。Spark インストールパスがわからない場合は、env | grep SPARK_HOME コマンドを使用してパスを見つけることができます。

• ECS クラスタ上の EMR を使用している場合は、Spark Beeline クライアントを直接使用して Spark Thrift Server に接続できます。

  パブリックエンドポイントを使用して接続する

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

  内部の同一リージョンエンドポイントを使用して接続する

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

Hive Beeline を使用して Serverless Spark Thrift Server に接続するときに次のエラーが発生した場合、通常は Hive Beeline のバージョンが Spark Thrift Server と互換性がないことが原因です。したがって、Hive Beeline 2.x を使用することをお勧めします。

24/08/22 15:09:11 [main]: ERROR jdbc.HiveConnection: Error opening session
org.apache.thrift.transport.TTransportException: HTTP Response code: 404

Apache Superset を構成して Spark Thrift Server に接続する

Apache Superset は、シンプルな折れ線グラフから非常に詳細な地理空間グラフまで、豊富なグラフ形式を備えた最新のデータ探索および 可視化 プラットフォームです。Superset の詳細については、Superset をご覧ください。

1. 依存関係をインストールします。

   thrift パッケージバージョン 0.20.0 をインストール済みであることを確認します。インストールされていない場合は、次のコマンドを使用してインストールできます。

   pip install thrift==0.20.0

2. Superset を起動し、Superset インターフェースに移動します。

   Superset の起動方法の詳細については、Superset ドキュメント をご覧ください。

3. ページの右上隅にある [DATABASE] をクリックして、[Connect A Database] ページに移動します。

4. [Connect A Database] ページで、[Apache Spark SQL] を選択します。

   

5. 接続文字列を入力し、関連するデータソースパラメーターを構成します。

   パブリックエンドポイントを使用して接続する

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

   内部の同一リージョンエンドポイントを使用して接続する

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

6. [FINISH] をクリックして、接続と検証が成功したことを確認します。

Hue を構成して Spark Thrift Server に接続する

Hue は、Hadoop エコシステムと対話するための一般的なオープンソース Web インターフェースです。Hue の詳細については、Hue 公式ドキュメント をご覧ください。

1. 依存関係をインストールします。

   thrift パッケージバージョン 0.20.0 をインストール済みであることを確認します。インストールされていない場合は、次のコマンドを使用してインストールできます。

   pip install thrift==0.20.0

2. Spark SQL 接続文字列を Hue 構成ファイルに追加します。

   Hue 構成ファイル（通常は /etc/hue/hue.conf にあります）を見つけて、次の内容をファイルに追加します。

   パブリックエンドポイントを使用して接続する

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

   内部の同一リージョンエンドポイントを使用して接続する

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

3. Hue を再起動します。

   構成を変更した後、次のコマンドを実行して Hue サービスを再起動し、変更を有効にする必要があります。

   sudo service hue restart

4. 接続を確認します。

   Hue が正常に再起動されたら、Hue インターフェースにアクセスして、Spark SQL オプションを見つけます。構成が正しい場合は、Spark Thrift Server に正常に接続して SQL クエリを実行できるはずです。

   

DataGrip を使用して Spark Thrift Server に接続する

DataGrip は、開発者向けのデータベース管理環境であり、データベースのクエリ、作成、および管理を容易にするように設計されています。データベースは、ローカル、サーバー、またはクラウドで実行できます。 DataGrip の詳細については、DataGrip をご覧ください。

1. DataGrip をインストールします。詳細については、DataGrip のインストール をご覧ください。

   この例で使用されている DataGrip のバージョンは 2025.1.2 です。

2. DataGrip クライアントを開き、DataGrip インターフェースに移動します。

3. プロジェクトを作成します。

   1. をクリックし、[Flie] > [New] > > [Project] を選択します。

      

   2. [New Project] ダイアログボックスで、Spark などのプロジェクト名を入力し、[OK] をクリックします。

4. [Database Explorer] メニューバーのアイコンをクリックします。 [Data Source] > [Other] > [Apache Spark] を選択します。

   

5. [Data Sources And Drivers] ダイアログボックスで、次のパラメーターを構成します。

   

   タブ	パラメーター	説明
   [General]	名前	カスタム接続名。例: spark_thrift_server。
   	[Authentication]	認証方法。この例では、[No Auth] が選択されています。 本番環境では、[User & Password] を選択して、承認されたユーザーのみが SQL タスクを送信できるようにします。これにより、システムのセキュリティが向上します。
   	[Driver]	[Apache Spark] をクリックし、次に [Go To Driver] をクリックして、ドライバーのバージョンが ver. 1.2.2 であることを確認します。 説明 現在の Serverless Spark エンジンバージョンは 3.x であるため、システムの安定性と機能の互換性を確保するには、ドライバーのバージョンを 1.2.2 にする必要があります。
   	[URL]	Spark Thrift Server に接続するための URL。ネットワーク環境（パブリックまたは内部）に基づいて、適切な接続方法を選択できます。 パブリックエンドポイントを使用して接続する jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token> 内部の同一リージョンエンドポイントを使用して接続する jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>
   [Options]	[Run Keep-alive Query]	このパラメーターはオプションです。タイムアウトによる自動切断を防ぐには、このパラメーターを選択します。

6. [Test Connection] をクリックして、データソースが正常に構成されていることを確認します。

   

7. [OK] をクリックして構成を完了します。

8. DataGrip を使用して Spark Thrift Server を管理します。

   DataGrip が Spark Thrift Server に正常に接続した後、データ開発を実行できます。詳細については、「DataGrip ヘルプ」をご参照ください。

   たとえば、作成された接続で、ターゲットテーブルを右クリックし、[New] > > [Query Console] を選択します。次に、右側の SQL エディターで SQL スクリプトを作成して実行し、テーブルデータ情報を表示できます。