E-MapReduce:JAR 開発のクイックスタート

ステップ 1： JAR パッケージの開発

EMR Serverless Spark は、JAR 開発のための統合開発環境 (IDE) を提供していません。ローカルまたはスタンドアロンの開発環境で Spark アプリケーションをコーディングし、JAR ファイルにパッケージ化する必要があります。

Maven プロジェクトの pom.xml ファイルに、必要な Spark の依存関係を追加します。EMR Serverless Spark の実行環境にはこれらの依存関係がすでに含まれているため、scope を provided に設定することで、重複したパッケージングやバージョンの競合を防ぎつつ、コンパイルとテストのために依存関係を利用可能な状態に保つことができます。

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.5.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.5.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.12</artifactId>
    <version>3.5.2</version>
    <scope>provided</scope>
</dependency>

public class HiveTableAccess {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("DlfTableAccessExample")
                .enableHiveSupport()
                .getOrCreate();
        spark.sql("SELECT * FROM test_table").show();
        spark.stop();
    }
}

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import java.util.ArrayList;
import java.util.List;
/**
 * 円周率の近似値を計算します
 * 使用法： JavaSparkPi [partitions]
 */
public final class JavaSparkPi {
  public static void main(String[] args) throws Exception {
    SparkSession spark = SparkSession
      .builder()
      .appName("JavaSparkPi")
      .getOrCreate();
    JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
    int slices = (args.length == 1) ? Integer.parseInt(args[0]) : 2;
    int n = 100000 * slices;
    List<Integer> l = new ArrayList<>(n);
    for (int i = 0; i < n; i++) {
      l.add(i);
    }
    JavaRDD<Integer> dataSet = jsc.parallelize(l, slices);
    int count = dataSet.map(integer -> {
      double x = Math.random() * 2 - 1;
      double y = Math.random() * 2 - 1;
      return (x * x + y * y <= 1) ? 1 : 0;
    }).reduce((integer, integer2) -> integer + integer2);
    System.out.println("Pi is roughly " + 4.0 * count / n);
    spark.stop();
  }
}

SparkExample-1.0-SNAPSHOT.jar をクリックして、テスト用の JAR パッケージをダウンロードします。

ステップ 2： JAR パッケージのアップロード

1. アーティファクトのアップロードページに移動します。

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

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

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

   4. EMR Serverless Spark ページの左側のナビゲーションウィンドウで、Artifacts をクリックします。

2. Artifacts ページで、Upload File をクリックします。

3. Upload File ダイアログボックスで、アップロードエリアをクリックしてローカルの JAR パッケージを選択するか、パッケージをアップロードエリアにドラッグします。

   この例では、SparkExample-1.0-SNAPSHOT.jar を使用します。

ステップ 3： ジョブの開発と実行

1. EMR Serverless Spark ページの左側のナビゲーションウィンドウで、Development をクリックします。

2. Development タブで、 アイコンをクリックします。

3. 名前を入力し、タイプとして Application > JAR を選択してから、OK をクリックします。

4. 右上隅で、リソースキューを選択します。

   キューの追加方法の詳細については、「リソースキューの管理」をご参照ください。

5. 新しいジョブエディターで、次のパラメーターを設定します。残りはデフォルト設定のままにして、実行 をクリックします。

   パラメーター	説明
   Main JAR Resource	アップロードした JAR パッケージを選択します。この例では、SparkExample-1.0-SNAPSHOT.jar を使用します。
   Main Class	Spark ジョブで実行するメインクラスです。 円周率の近似値を計算するには、org.example.JavaSparkPi と入力します。 Hive テーブルをクエリするには、org.example.HiveTableAccess と入力します。

6. ジョブが完了したら、下部にある Execution Records セクションに移動し、[操作] 列の [ログ] をクリックします。

   [ログ] パネルで、[ドライバーログ] > [Stdout] を選択します。出力 Pi is roughly 3.1403 は、Spark ジョブが正常に実行されたことを示します。

   テーブルクエリの出力例は、id=1, name=jay という 1 つのレコードを含むフォーマットされたテーブルです。

ステップ 4： ジョブの公開

1. ジョブが完了したら、右側の Publish をクリックします。

2. 表示されるダイアログボックスで、公開情報を入力し、OK をクリックします。

(オプション) ステップ 5： Spark UI の表示

ジョブの実行後、Spark UI でその実行詳細を表示できます。

1. 左側のナビゲーションウィンドウで、Job History をクリックします。

2. [アプリケーション] ページで、対象のジョブを見つけ、[操作] 列の Spark UI をクリックします。

3. ページには、ユーザー： root、スケジューリングモード： FIFO、完了したジョブ： 1 などの詳細が表示されます。ジョブエントリには、ID 0、説明 reduce at SparkPi.scala:38、持続時間 4 秒、ステージ (1/1) とタスク (2/2) の進捗が表示されます。これらの詳細から、Spark 円周率ジョブが正常に実行されたことが確認できます。