E-MapReduce:EMR Serverless Spark と他の VPC 間のネットワーク接続を確立する

前提条件

データソースを準備する必要があります。このトピックでは、DataLake クラスターを例として使用します。クラスターは EMR on ECS ページで作成し、Hive サービスを含め、[メタデータ] に [内蔵 MySQL] データベースを使用する必要があります。詳細については、「クラスターの作成」をご参照ください。

制限事項

現在、次のゾーンの vSwitch のみを使用できます。

• 中国リージョン

  リージョン名	リージョン ID	ゾーン名
  中国 (杭州)	cn-hangzhou	ゾーン H ゾーン I ゾーン J
  中国 (上海)	cn-shanghai	ゾーン B ゾーン L ゾーン F ゾーン G
  中国 (北京)	cn-beijing	ゾーン F ゾーン G ゾーン H ゾーン K
  中国 (深圳)	cn-shenzhen	ゾーン E ゾーン F
  中国 (香港)	cn-hongkong	ゾーン B ゾーン C

• その他の国とリージョン

  リージョン名	リージョン ID	ゾーン名
  ドイツ (フランクフルト)	eu-central-1	ゾーン A ゾーン B
  インドネシア (ジャカルタ)	ap-southeast-5	ゾーン A ゾーン B
  シンガポール	ap-southeast-1	ゾーン B ゾーン C
  米国 (バージニア)	us-east-1	ゾーン A ゾーン B
  米国 (シリコンバレー)	us-west-1	ゾーン A ゾーン B
  日本 (東京)	ap-northeast-1	ゾーン B ゾーン C

ステップ 1: ネットワーク接続の追加

1. ネットワーク接続ページに移動します。

   1. EMR コンソールにログインします。

   2. 左側のナビゲーションウィンドウで、EMR Serverless > Spark を選択します。

   3. [Spark] ページで、対象のワークスペースの名前をクリックします。

   4. [EMR Serverless Spark] ページで、左側のナビゲーションウィンドウにある ネットワーク接続をクリックします。

2. [ネットワーク接続] ページで、[ネットワーク接続の追加] をクリックします。

3. [ネットワーク接続の追加] ダイアログボックスで、パラメーターを設定し、[OK] をクリックします。

   パラメーター	説明
   接続名	新しい接続の名前を入力します。
   VPC	EMR クラスターと同じ VPC を選択します。 利用可能な VPC がない場合は、[VPC の作成] をクリックして VPC コンソールに移動し、VPC を作成します。詳細については、「VPC と vSwitch」をご参照ください。 説明 Serverless Spark がインターネットにアクセスする必要がある場合は、ネットワーク接続がパブリックネットワークアクセス権を持っていることを確認してください。たとえば、VPC にインターネット NAT ゲートウェイをデプロイできます。これにより、Serverless Spark インスタンスはゲートウェイを介してインターネットにアクセスできます。詳細については、「インターネット NAT ゲートウェイの SNAT 機能を使用してインターネットにアクセスする」をご参照ください。
   VSwitch	EMR クラスターと同じ VPC 内の vSwitch を選択します。 現在のゾーンで利用可能な vSwitch がない場合は、[仮想スイッチ] をクリックして VPC コンソールに移動し、vSwitch を作成します。詳細については、「vSwitch の作成と管理」をご参照ください。 重要 特定のゾーンの vSwitch のみを選択できます。詳細については、「制限事項」をご参照ください。

   ネットワーク接続の [ステータス] が [成功] に変わると、ネットワーク接続が追加されます。

   

ステップ 2: EMR クラスターのセキュリティグループルールを追加する

1. ネットワーク接続で指定された vSwitch の CIDR ブロックを取得します。

   VPC コンソールにログインし、[VSwitches] ページに移動して vSwitch の CIDR ブロックを取得できます。

   

2. セキュリティグループルールを追加します。

   1. EMR on ECS コンソールにログインします。

   2. [EMR On ECS] ページで、対象クラスターの ID をクリックします。

   3. [基本情報] タブの [セキュリティ] セクションで、[クラスターセキュリティグループ] の横にあるリンクをクリックします。

   4. [セキュリティグループの詳細] ページの [アクセスルール] セクションで、[ルールの追加] をクリックします。次のパラメーターを設定し、[OK] をクリックします。

      パラメーター	説明
      プロトコル	許可するネットワーク通信プロトコルを指定します。デフォルトは TCP です。 説明 ネットワーク接続が Kerberos 認証に使用される場合は、UDP プロトコルを選択し、ポート 88 を開きます。Kerberos 認証の詳細については、「Kerberos 認証を有効にする」をご参照ください。
      ソース	前のステップで取得した vSwitch の CIDR ブロックを入力します。 重要 外部攻撃によるセキュリティリスクを防ぐため、[Authorization Object] を 0.0.0.0/0 に設定しないでください。
      宛先 (このインスタンス)	アクセスを許可する宛先ポートを指定します。例: 9083。

(オプション) ステップ 3: Hive サービスに接続してテーブルデータをクエリする

すでに Hive テーブルを作成して設定している場合は、このステップをスキップできます。

1. Secure Shell (SSH) を使用してクラスターのマスターノードにログインします。詳細については、「クラスターへのログイン」をご参照ください。

2. 次のコマンドを実行して、Hive コマンドラインに入ります。

   hive

3. 次のコマンドを実行して、テーブルを作成します。

   CREATE TABLE my_table (id INT,name STRING);

4. 次のコマンドを実行して、テーブルにデータを挿入します。

   INSERT INTO my_table VALUES (1, 'John'); 
   INSERT INTO my_table VALUES (2, 'Jane');

5. 次のコマンドを実行して、データをクエリします。

   SELECT * FROM my_table;

(オプション) ステップ 4: リソースファイルの準備とアップロード

JAR ジョブを使用する場合は、リソースファイルを準備する必要があります。SparkSQL ジョブタイプを使用する場合は、このステップをスキップできます。

1. ローカルマシンで新しい Maven プロジェクトを作成します。

   プロジェクトには次の内容が含まれています。

   package com.example;
   
   import org.apache.spark.sql.Dataset;
   import org.apache.spark.sql.Row;
   import org.apache.spark.sql.SparkSession;
   
   public class DataFrameExample {
       public static void main(String[] args) {
           // SparkSession を作成します。
           SparkSession spark = SparkSession.builder()
                   .appName("HMSQueryExample")
                   .enableHiveSupport()
                   .getOrCreate();
   
           // クエリを実行します。
           Dataset<Row> result = spark.sql("SELECT * FROM default.my_table");
   
           // クエリ結果を出力します。
           result.show();
   
           // SparkSession を閉じます。
           spark.stop();
       }
   }

   pom.xml ファイルには次の内容が含まれています。

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="http://maven.apache.org/POM/4.0.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
       <modelVersion>4.0.0</modelVersion>
   
       <groupId>org.example</groupId>
       <artifactId>sparkDataFrame</artifactId>
       <version>1.0-SNAPSHOT</version>
   
       <properties>
           <maven.compiler.source>8</maven.compiler.source>
           <maven.compiler.target>8</maven.compiler.target>
           <spark.version>3.3.1</spark.version>
           <scala.binary.version>2.12</scala.binary.version>
       </properties>
   
       <dependencies>
           <dependency>
               <groupId>org.apache.spark</groupId>
               <artifactId>spark-core_${scala.binary.version}</artifactId>
               <version>${spark.version}</version>
           </dependency>
           <dependency>
               <groupId>org.apache.spark</groupId>
               <artifactId>spark-sql_${scala.binary.version}</artifactId>
               <version>${spark.version}</version>
           </dependency>
           <dependency>
               <groupId>org.apache.spark</groupId>
               <artifactId>spark-hive_${scala.binary.version}</artifactId>
               <version>${spark.version}</version>
           </dependency>
       </dependencies>
   </project>

2. mvn package コマンドを実行します。プロジェクトがコンパイルおよびパッケージ化されると、sparkDataFrame-1.0-SNAPSHOT.jar ファイルが生成されます。

3. 対象のワークスペースの EMR Serverless Spark ページで、左側のナビゲーションウィンドウにある [ファイル管理] をクリックします。

4. [ファイル管理] ページで、[ファイルのアップロード] をクリックします。

5. sparkDataFrame-1.0-SNAPSHOT.jar ファイルをアップロードします。

ステップ 5: ジョブの作成と実行