E-MapReduce:spark-submit を使用した Spark ジョブの送信

ステップ 1：EMR Serverless spark-submit ツールのダウンロードとインストール

1. emr-serverless-spark-tool-1.16.0-bin.zip をクリックして、インストールパッケージをダウンロードします。

2. インストールパッケージをご利用の ECS インスタンスにアップロードします。詳細については、「ファイルのアップロードまたはダウンロード」をご参照ください。

3. 次のコマンドを実行して、EMR Serverless spark-submit ツールを解凍します。

   unzip emr-serverless-spark-tool-1.16.0-bin.zip

ステップ 2：パラメーターの設定

1. 次のコマンドを実行して、connection.properties ファイルの構成を変更します。

   vim emr-serverless-spark-tool-1.16.0/conf/connection.properties

2. 次の例に示すように、ファイルを key=value フォーマットのパラメーターで設定することをお勧めします。

   accessKeyId=<ALIBABA_CLOUD_ACCESS_KEY_ID>
   accessKeySecret=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
   regionId=cn-hangzhou
   endpoint=emr-serverless-spark.cn-hangzhou.aliyuncs.com
   workspaceId=w-xxxxxxxxxxxx

   重要

   この AccessKey に対応する RAM ユーザーまたはロールは、RAM で認可され、EMR Serverless Spark ワークスペースに追加されている必要があります。

   • RAM 認可については、「RAM ユーザーへの権限付与」をご参照ください。

   • EMR Serverless Spark ワークスペースでのユーザーとロールの管理については、「ユーザーとロールの管理」をご参照ください。

   次の表にパラメーターを示します。

   パラメーター	必須	説明
   accessKeyId	はい	Spark ジョブを実行する Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID と AccessKey Secret。 重要 accessKeyId および accessKeySecret パラメーターを設定する際は、AccessKey に関連付けられているユーザーが、ワークスペースに関連付けられている OSS バケットに対する読み取りおよび書き込み権限を持っていることを確認してください。 Spark ページで、ワークスペースの [操作] 列にある Details をクリックすると、そのワークスペースに関連付けられている OSS バケットを表示できます。
   accessKeySecret	はい
   regionId	はい	リージョン ID。この例では、中国 (杭州) リージョンを使用します。
   endpoint	はい	EMR Serverless Spark のエンドポイント。詳細については、「エンドポイント」をご参照ください。 この例では、中国 (杭州) リージョンのパブリックエンドポイントを使用します。パラメーターの値は emr-serverless-spark.cn-hangzhou.aliyuncs.com です。 説明 ご利用の ECS インスタンスがインターネットにアクセスできない場合は、VPC エンドポイントを使用する必要があります。
   workspaceId	はい	EMR Serverless Spark ワークスペースの ID。

ステップ 3：Spark ジョブの送信

1. 次のコマンドを実行して、EMR Serverless spark-submit ツールのディレクトリに移動します。

   cd emr-serverless-spark-tool-1.16.0

2. ジョブの種類に基づいて送信メソッドを選択します。

   ジョブを送信する際には、JAR パッケージや Python スクリプトなど、ジョブが依存するファイルリソースを指定する必要があります。これらのリソースは OSS またはローカルに保存できます。このトピックでは、すべての例で OSS リソースを使用します。

   spark-submit

   spark-submit は、Spark が提供する、Java/Scala および PySpark タスク用の汎用タスク送信ツールです。

   Java/Scala ジョブ

   この例では、spark-examples_2.12-3.5.2.jar を使用します。spark-examples_2.12-3.5.2.jar をクリックしてテスト用の JAR パッケージをダウンロードし、その JAR パッケージを OSS にアップロードできます。この JAR パッケージは、円周率 (π) の値を計算する、Spark に組み込まれた簡単な例です。

   説明

   spark-examples_2.12-3.5.2.jar ファイルは、ジョブを送信するために esr-4.x エンジンバージョンで使用する必要があります。esr-5.x エンジンバージョンを使用している場合は、この例のために spark-examples_2.13-4.0.1.jar をダウンロードしてください。

   ./bin/spark-submit  --name SparkPi \
   --queue dev_queue  \
   --num-executors 5 \
   --driver-memory 1g \
   --executor-cores 2 \
   --executor-memory 2g \
   --class org.apache.spark.examples.SparkPi \
    oss://<yourBucket>/path/to/spark-examples_2.12-3.5.2.jar \
   10000

   PySpark ジョブ

   この例では、DataFrame.py と employee.csv を使用します。DataFrame.py と employee.csv をクリックしてテストファイルをダウンロードし、それらを OSS にアップロードできます。

   説明

   • DataFrame.py は、Apache Spark フレームワークを使用して OSS のデータを処理するコードスニペットです。

   • employee.csv は、従業員の名前、部署、給与を含むデータファイルです。

   ./bin/spark-submit --name PySpark \
   --queue dev_queue  \
   --num-executors 5 \
   --driver-memory 1g \
   --executor-cores 2 \
   --executor-memory 2g \
   --conf spark.tags.key=value \
   oss://<yourBucket>/path/to/DataFrame.py \
   oss://<yourBucket>/path/to/employee.csv

   パラメーターは以下で説明します。

   • サポートされているオープンソースパラメーター

     パラメーター	例	説明
     --name	SparkPi	Spark ジョブのアプリケーション名。
     --class	org.apache.spark.examples.SparkPi	Spark ジョブのエントリクラス。このパラメーターは Java または Scala アプリケーションでは必須ですが、Python アプリケーションでは必須ではありません。
     --num-executors	5	Spark ジョブのエグゼキュータ数。
     --driver-cores	1	Spark ジョブのドライバーコア数。
     --driver-memory	1g	Spark ジョブのドライバーメモリ量。
     --executor-cores	2	Spark ジョブのエグゼキュータコア数。
     --executor-memory	2g	Spark ジョブのエグゼキュータメモリ量。
     --files	oss://<yourBucket>/file1,oss://<yourBucket>/file2	Spark ジョブで必要なリソースファイル。ファイルは OSS またはローカルに保存できます。複数のファイルはカンマ (,) で区切ります。
     --py-files	oss://<yourBucket>/file1.py,oss://<yourBucket>/file2.py	Spark ジョブで必要な Python スクリプト。スクリプトは OSS またはローカルに保存できます。複数のファイルはカンマ (,) で区切ります。このパラメーターは PySpark アプリケーション専用です。
     --jars	oss://<yourBucket>/file1.jar,oss://<yourBucket>/file2.jar	Spark ジョブで必要な JAR パッケージ。リソースは OSS またはローカルに保存できます。複数のファイルはカンマ (,) で区切ります。
     --archives	oss://<yourBucket>/archive.tar.gz#env,oss://<yourBucket>/archive2.zip	Spark ジョブで必要なアーカイブファイル。リソースは OSS またはローカルに保存できます。複数のファイルはカンマ (,) で区切ります。
     --queue	root_queue	Spark ジョブが実行されるキューの名前。この名前は、EMR Serverless Spark ワークスペースのキュー管理セクションのキュー名と一致する必要があります。
     --proxy-user	test	設定値は HADOOP_USER_NAME 環境変数をオーバーライドし、その動作はオープンソース版と同様です。
     --conf	spark.tags.key=value	Spark ジョブのカスタムパラメーター。
     --status	jr-8598aa9f459d****	Spark ジョブのステータスを確認します。
     --kill	jr-8598aa9f459d****	Spark ジョブを終了します。

   • 拡張パラメーター

     パラメーター	例	説明
     --detach	値は不要	ジョブのステータスを待たずに、ジョブを送信した直後に spark-submit を終了させます。
     --detail	jr-8598aa9f459d****	Spark ジョブの詳細を表示します。
     --release-version	esr-4.1.1 (Spark 3.5.2, Scala 2.12)	Spark のバージョンを指定します。コンソールに表示されるエンジンバージョン番号を入力します。
     --enable-template	値は不要	テンプレート機能を有効にします。ジョブはワークスペースのデフォルト設定テンプレートを使用します。 [設定管理] で Configurations を作成した場合、--conf パラメーターに spark.emr.serverless.templateId を追加してその ID を指定できます。ジョブは指定されたテンプレートを直接適用します。テンプレートの作成に関する詳細については、「設定管理」をご参照ください。 --enable-template のみを指定した場合、タスクはワークスペースのデフォルトの構成テンプレートを自動的に適用します。 --conf のみを使用してテンプレートIDを指定すると、タスクは指定されたテンプレートIDを直接適用します。 --enable-template と --conf の両方を指定： --enable-template と --conf spark.emr.serverless.templateId の両方を指定した場合、--conf のテンプレートIDがデフォルトテンプレートをオーバーライドします。 パラメーター未指定: --enable-template を使用せず、--conf spark.emr.serverless.templateId も指定しない場合、ジョブにはテンプレート構成が適用されません。
     --timeout	60	ジョブのタイムアウト期間 (秒単位)。
     --workspace-id	w-4b4d7925a797****	ジョブレベルでワークスペース ID を指定します。この設定は、connection.properties ファイル内の workspaceId パラメーターをオーバーライドします。

   • サポートされていないオープンソースパラメーター

     • --deploy-mode

     • --master

     • --repositories

     • --keytab

     • --principal

     • --total-executor-cores

     • --driver-library-path

     • --driver-class-path

     • --supervise

     • --verbose

   spark-sql

   spark-sql は、SQL クエリまたはスクリプトを直接実行するためのツールです。

   • 例 1：SQL ステートメントを直接実行する

     spark-sql -e "SHOW TABLES"

     このコマンドは、現在のデータベース内のすべてのテーブルを一覧表示します。

   • 例 2：SQL スクリプトファイルを実行する

     spark-sql -f oss://<yourBucketname>/path/to/your/example.sql

     この例では、example.sql を使用します。example.sql をクリックしてテストファイルをダウンロードし、それを OSS にアップロードできます。

     example.sql ファイルのコンテンツ例

     CREATE TABLE IF NOT EXISTS employees (
         id INT,
         name STRING,
         age INT,
         department STRING
     );
     INSERT INTO employees VALUES
     (1, 'Alice', 30, 'Engineering'),
     (2, 'Bob', 25, 'Marketing'),
     (3, 'Charlie', 35, 'Sales');
     SELECT * FROM employees;
     

   次の表にパラメーターを示します。

   パラメーター	例	説明
   -e "<sql>"	-e "SELECT * FROM table"	コマンドラインからインライン SQL ステートメントを実行します。
   -f <パス>	-f oss://path/script.sql	指定されたパスから SQL スクリプトファイルを実行します。

ステップ 4：Spark ジョブの照会

cd emr-serverless-spark-tool-1.16.0
./bin/spark-submit --status <jr-8598aa9f459d****>

cd emr-serverless-spark-tool-1.16.0
./bin/spark-submit --detail <jr-8598aa9f459d****>

(オプション) ステップ 5：Spark ジョブの終了

cd emr-serverless-spark-tool-1.16.0
./bin/spark-submit --kill <jr-8598aa9f459d****>