Spark は、高性能、使いやすさ、汎用性に優れたビッグデータ向けの統合分析エンジンです。複雑なインメモリコンピューティングをサポートしており、大規模で低レイテンシーなデータ分析アプリケーションの構築に最適です。DataWorks では、EMR Spark ノードを使用して Spark タスクを開発し、定期的にスケジュールすることができます。このトピックでは、EMR Spark ノードの設定方法と使用方法について説明し、アプリケーション例を示します。
前提条件
ノードのコンポーネント環境をカスタマイズするには、公式の
dataworks_emr_base_task_podイメージに基づいてカスタムイメージを作成します。詳細については、「カスタムイメージ」および「Data Development でのイメージの使用」をご参照ください。たとえば、カスタムイメージを作成する際に、Spark JAR パッケージを置き換えたり、特定の
libraries、files、またはJAR packagesへの依存関係を追加したりできます。-
Alibaba Cloud E-MapReduce (EMR) クラスターを作成し、DataWorks に登録済みであること。詳細については、「Data Studio: EMR コンピューティングリソースの関連付け」をご参照ください。
(オプション、RAM ユーザーに必要) タスク開発を担当する Resource Access Management (RAM) ユーザーをワークスペースに追加し、[Developer] または [Workspace Administrator] ロールを割り当てます。Workspace Administrator ロールは広範な権限を持つため、慎重に付与してください。メンバーの追加の詳細については、「ワークスペースにメンバーを追加」をご参照ください。
Alibaba Cloud アカウントを使用している場合は、このステップをスキップできます。
タスクに特定の開発環境が必要な場合、カスタムイメージをビルドできます。詳細については、「カスタムイメージ」をご参照ください。
制限事項
このタスクタイプは、Serverless リソースグループ（推奨）または専用スケジューリングリソースグループでのみ実行されます。Data Development でイメージを使用する場合は、Serverless リソースグループを使用する必要があります。
-
DataWorks で DataLake クラスターまたはカスタムクラスターのメタデータを管理するには、まずクラスターに EMR-HOOK を設定する必要があります。詳細については、「Spark SQL 用の EMR-HOOK の設定」をご参照ください。
説明クラスターに EMR-HOOK が設定されていない場合、DataWorks はメタデータをリアルタイムで表示したり、監査ログを生成したり、データリネージを表示したり、EMR 関連のガバナンスタスクを実行したりすることはできません。
-
EMR on ACK Spark クラスターではデータリネージの表示はサポートされていませんが、EMR Serverless Spark クラスターではサポートされています。
-
EMR on ACK Spark クラスターと EMR Serverless Spark クラスターは、OSS REF 経由の OSS リソースの参照と OSS へのリソースのアップロードのみをサポートしており、HDFS へのリソースのアップロードはサポートされていません。
-
DataLake クラスターとカスタムクラスターは、OSS REF 経由の OSS リソースの参照、OSS へのリソースのアップロード、および HDFS へのリソースのアップロードをサポートしています。
注意事項
現在のワークスペースに関連付けられている EMR クラスターで、Spark の Ranger アクセス制御を有効にしている場合：
-
デフォルトイメージを使用する Spark タスクは、この機能を自動的にサポートします。
-
カスタムイメージで Spark タスクを実行するには、テクニカルサポートに チケットを送信 して、イメージのアップグレードをリクエストしてください。
Spark ジョブの開発と JAR パッケージの取得
DataWorks で EMR Spark ジョブをスケジュールするには、EMR でジョブを開発し、コードをコンパイルして JAR パッケージを作成する必要があります。EMR Spark ジョブの開発に関する詳細については、「Spark の概要」をご参照ください。
その後、定期的な EMR Spark ジョブをスケジュールするために、JAR パッケージを DataWorks にアップロードする必要があります。
手順
-
EMR Spark ノードの編集ページで、ジョブを開発します。
Spark ジョブの開発
シナリオに応じてオプションを選択します。
オプション1：アップロードと参照
DataWorks では、ローカルマシンから DataStudio にリソースをアップロードして参照できます。EMR Spark ジョブをコンパイルした後、コンパイル済みの JAR パッケージを取得する必要があります。JAR パッケージの推奨される保存方法は、そのサイズによって異なります。
JAR パッケージをアップロードして DataWorks EMR リソースとして作成し、送信します。EMR クラスターの HDFS に直接保存することもできます。EMR on ACK および EMR Serverless Spark クラスターは、HDFS へのリソースのアップロードをサポートしていません。
JAR < 500 MB
-
EMR JAR リソースの作成
JAR パッケージが 500 MB 未満の場合、ローカルマシンから DataWorks EMR JAR リソースとしてアップロードできます。これにより、DataWorks コンソールで簡単に視覚的に管理できます。リソースを作成してから、送信します。詳細については、「EMR リソースの作成と使用」をご参照ください。
-
[Local] アップロード方法を使用して、JAR リソースの保存ディレクトリーに JAR パッケージをアップロードします。詳細については、「リソース管理」をご参照ください。
-
Click Upload をクリックして JAR リソースをアップロードします。
-
Storage Path、Data Sources、Resource Group を選択します。
-
Save をクリックします。
-
-
EMR JAR リソースの参照
-
作成した [EMR Spark] ノードを開き、コード編集ページに移動します。
-
左側のナビゲーションペインの [リソース管理] セクションで、参照するリソースを見つけ、右クリックして Insert Resource Path を選択します。
-
参照を選択すると、[EMR Spark] ノードのコードエディターに参照ステートメントが自動的に追加されます。これは、参照が成功したことを示します。
##@resource_reference{"spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jar"} spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jarこのコードでは、
spark-examples_2.12-1.0.0-SNAPSHOT-shaded.jarは、アップロードした EMR JAR リソースの名前です。 -
spark-submit コマンドが含まれるように、EMR Spark ノードのコードを編集します。次のコードは例です。
説明EMR Spark ノードエディターはコメントをサポートしていません。次の例に示すように、コメントなしでジョブコードを記述してください。そうしないと、ノードの実行が失敗します。
##@resource_reference{"spark-examples_2.11-2.4.0.jar"} spark-submit --class org.apache.spark.examples.SparkPi --master yarn spark-examples_2.11-2.4.0.jar 100説明-
org.apache.spark.examples.SparkPi：コンパイル済み JAR パッケージのメインクラスです。 -
spark-examples_2.11-2.4.0.jar：アップロードした EMR JAR リソースの名前です。 -
その他のパラメーターには、サンプル値を使用できます。また、
spark-submit --helpコマンドを実行して使用方法を確認し、必要に応じて spark-submit コマンドを変更することもできます。 -
Spark ノードで spark-submit コマンドの簡略化されたパラメーターを使用するには、コードに追加する必要があります。例：
--executor-memory 2G。 -
Spark ノードは、クラスターモードでの YARN へのジョブ送信のみをサポートしています。
-
spark-submit でジョブを送信する場合、デプロイモードはクライアントモードではなくクラスターモードに設定してください。
-
-
JAR >= 500 MB
-
Create an EMR JAR resource.
If a jar package is 500 MB or larger, you cannot upload it from your local machine as a DataWorks resource. We recommend that you store the jar package directly in HDFS on the EMR cluster and record its storage path. You can then reference this path when you schedule the Spark job in DataWorks.
-
Upload the jar package to the storage directory for JAR resources by using the [Local] upload method. For more information, see Resource management.
-
Click [Click Upload] to upload the JAR resource.
-
Select a [Storage Path], [Data Sources], and [Resource Group].
-
Click [Save].
-
-
Reference the EMR JAR resource.
If the jar package is stored in HDFS, you can reference it by specifying its path in the code of the EMR Spark node.
-
Double-click the created [EMR Spark] node to open its code editor page.
-
Write the spark-submit command. The following is an example.
spark-submit --master yarn --deploy-mode cluster --name SparkPi --driver-memory 4G --driver-cores 1 --num-executors 5 --executor-memory 4G --executor-cores 1 --class org.apache.spark.examples.JavaSparkPi hdfs:///tmp/jars/spark-examples_2.11-2.4.8.jar 100説明-
hdfs:///tmp/jars/spark-examples_2.11-2.4.8.jar: The actual path of the jar package in HDFS.
-
org.apache.spark.examples.JavaSparkPi: The main class in the compiled jar package.
-
Configure the other parameters according to your EMR cluster's settings. You can also run the
spark-submit --helpcommand to view the usage information for spark-submit and modify the command as needed. -
To use simplified parameters for the spark-submit command in a Spark node, you must add them to your code, for example,
--executor-memory 2G. -
Spark nodes only support submitting jobs to YARN in cluster mode.
-
For jobs submitted with spark-submit, set the deploy mode to cluster mode, not client mode.
-
-
オプション2：OSS リソースの参照
OSS REF を使用して、現在のノードから OSS リソースを直接参照できます。EMR ノードを実行すると、DataWorks はコードで指定された OSS リソースを自動的にロードしてローカルで使用できるようにします。この方法は、EMR ジョブが JAR 依存関係を必要とする場合や、スクリプトに依存する場合のシナリオで一般的に使用されます。
-
JAR リソースの開発
-
コード依存関係の準備
EMR クラスターのマスターノードの
/usr/lib/emr/spark-current/jars/パスで、必要なコードの依存関係を確認できます。次の例では Spark 3.4.2 を使用しています。既存の IntelliJ IDEA プロジェクトを開き、プロジェクトオブジェクトモデル (POM) 依存関係を追加し、関連するプラグインを参照する必要があります。POM 依存関係
<dependencies> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>3.4.2</version> </dependency> <!-- Apache Spark SQL --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>3.4.2</version> </dependency> </dependencies>プラグイン
<build> <sourceDirectory>src/main/scala</sourceDirectory> <testSourceDirectory>src/test/scala</testSourceDirectory> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.7.0</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> <plugin> <artifactId>maven-assembly-plugin</artifactId> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> <configuration> <recompileMode>incremental</recompileMode> </configuration> <executions> <execution> <goals> <goal>compile</goal> <goal>testCompile</goal> </goals> <configuration> <args> <arg>-dependencyfile</arg> <arg>${project.build.directory}/.scala_dependencies</arg> </args> </configuration> </execution> </executions> </plugin> </plugins> </build> -
サンプルコード
package com.aliyun.emr.example.spark import org.apache.spark.sql.SparkSession object SparkMaxComputeDemo { def main(args: Array[String]): Unit = { // SparkSession を作成します。 val spark = SparkSession.builder() .appName("HelloDataWorks") .getOrCreate() // Spark バージョンを出力します。 println(s"Spark version: ${spark.version}") } } -
Scala コードを編集した後、JAR パッケージを生成します。
サンプル JAR パッケージの名前は
SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jarです。
-
-
JAR リソースのアップロード
-
コード開発が完了したら、OSS コンソール にログインします。左側のナビゲーションペインで、[Bucket List] をクリックします。
-
宛先バケットの名前をクリックして、File Management ページを開きます。
このトピックでは、
onaliyun-bucket-2バケットを例として使用します。 -
Create Directory をクリックして、JAR リソースを保存するディレクトリーを作成します。
ディレクトリ名 を
emr/jarsに設定して、JAR リソースの保存ディレクトリーを作成します。 -
JAR リソースを保存ディレクトリーにアップロードします。
ディレクトリーに移動し、Upload File をクリックします。アップロード待ちファイル エリアで [Scan for Files] をクリックし、
SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jarファイルをバケットに追加してから、Upload File をクリックします。
-
-
JAR リソースの参照
-
JAR リソースを参照するようにコードを編集します。
作成した EMR Spark ノードの編集ページで、JAR リソースを参照するようにコードを編集します。
spark-submit --class com.aliyun.emr.example.spark.SparkMaxComputeDemo --master yarn ossref://onaliyun-bucket-2/emr/jars/SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jarパラメーターの説明：
パラメーター
説明
class実行するメインクラスの完全修飾名です。
masterSpark アプリケーションの実行モードです。
ossrefファイルパス形式は
ossref://{endpoint}/{bucket}/{object}です。-
[Endpoint]：OSS サービスのエンドポイントです。このパラメーターが空の場合、OSS リソースは EMR クラスターと同じリージョンにある必要があります。
-
バケット：OSS 内のオブジェクトを保存するためのコンテナーです。各 バケット には一意の名前があります。OSS コンソール にログインすると、アカウント内のすべての バケット を表示できます。
-
オブジェクト：バケット に保存されている特定のオブジェクト (ファイル名やパスなど) です。
-
-
EMR Spark ノードジョブの実行
編集が完了したら、
アイコンをクリックし、作成したサーバーレスリソースグループを選択して EMR Spark ノードを実行します。ジョブが完了したら、コンソールから applicationIdsを記録します。例：application_1730367929285_xxxx。 -
結果の表示
EMR Shell ノードを作成し、そのノード上で
yarn logs -applicationId application_1730367929285_xxxxコマンドを実行して実行結果を表示します。yarn logs -applicationId application_xxx Process Output>>> LogLastModifiedTime:Fri Nov 01 16:27:46 +0800 2024 Process Output>>> LogLength:21 Process Output>>> LogContents: Process Output>>> Spark version: 3.4.2 Process Output>>>
-
(オプション) 詳細パラメーターの設定
ノードの右側にある Scheduling Settings ペインの で、以下の固有のプロパティパラメーターを設定できます。
説明-
使用可能な詳細パラメーターは、次の表に示すように EMR クラスタータイプによって異なります。
-
その他の オープンソースの Spark プロパティは、 の Scheduling Settings ペインで設定できます。
データレイククラスター / カスタムクラスター：EMR on ECS
詳細パラメーター
説明
queue
ジョブのスケジューリングキューです。デフォルト値は
defaultです。EMR クラスターを DataWorks ワークスペースに登録した際に、ワークスペースレベルの YARN リソースキュー を設定した場合：
-
グローバル設定が優先されるかどうか チェックボックスを 選択 し、Yes に設定した場合、Spark ジョブは EMR クラスターの登録時に設定されたキューを使用します。
-
チェックボックスを 選択しない 場合、Spark ジョブは EMR Spark ノードに設定されているキューを使用します。
EMR YARN の詳細については、「基本的なキュー設定」をご参照ください。EMR クラスターの登録時のキュー設定の詳細については、「グローバル YARN リソースキューの設定」をご参照ください。
priority
優先度です。デフォルト値は 1 です。
FLOW_SKIP_SQL_ANALYZE
SQL ステートメントの実行モードです。有効な値：
-
true: 複数の SQL文を一度に実行します。 -
false(デフォルト): 一度に 1 つの SQL ステートメントを実行します。
説明このパラメーターは、データ開発環境でのテスト実行でのみサポートされています。
その他
-
詳細設定でカスタム Spark パラメーターを追加できます。 たとえば、
spark.eventLog.enabled : falseを追加すると、DataWorks は、EMR クラスターに送信されるコードに、そのパラメーターを--conf key=value形式で自動的に追加します。 -
グローバル Spark パラメーターを設定することもできます。詳細については、「グローバル Spark パラメーターの設定」をご参照ください。
説明Ranger アクセス制御を有効にするには、グローバル Spark パラメーターを設定するで
spark.hadoop.fs.oss.authorization.method=ranger設定を追加します。
Spark クラスター：EMR on ACK
詳細パラメーター
説明
FLOW_SKIP_SQL_ANALYZE
SQL ステートメントの実行モードです。有効な値：
-
true: 一度に複数の SQL ステートメントを実行します。 -
false: SQL ステートメントを 1 つずつ実行します。
説明このパラメーターは、データ開発環境でのテスト実行でのみサポートされています。
その他
-
[詳細設定] でカスタム Spark パラメーターを追加できます。 たとえば、
spark.eventLog.enabled: falseです。 DataWorks は、EMR クラスターに送信されるコードに、--conf key=value形式でパラメーターを自動的に追加します。 -
グローバル Spark パラメーターを設定することもできます。詳細については、「グローバル Spark パラメーターの設定」をご参照ください。
Hadoop クラスター：EMR on ECS
詳細パラメーター
説明
queue
ジョブが送信されるスケジューリングキューです。デフォルト値：default。
EMR クラスターを DataWorks ワークスペースに登録する際に、ワークスペースレベルの [YARN リソースキュー] を設定した場合、次のルールが適用されます：
-
グローバル設定が優先されるかどうか チェックボックスを選択し、値を Yes に設定した場合、Spark ジョブの実行時に、EMR クラスターの登録時に指定したキューの設定が優先されます。
-
チェックボックスの選択を解除した場合、Spark ジョブの実行時に、EMR Spark ノードに指定したキューの設定が優先されます。
EMR YARN の詳細については、「キューの基本設定」をご参照ください。EMR クラスターの登録時にキューを設定する方法の詳細については、「グローバルYARNリソースキューの設定」をご参照ください。
priority
優先度です。デフォルト値は 1 です。
FLOW_SKIP_SQL_ANALYZE
SQL ステートメントの実行モードです。有効な値：
-
true: 一度に複数の SQL ステートメントを実行します。 -
false: SQL ステートメントを 1 つずつ実行します。
説明このパラメーターは、データ開発環境でのテスト実行でのみサポートされています。
USE_GATEWAY
このノードのジョブをゲートウェイクラスター経由で送信するかどうかを指定します。有効な値：
-
true: Gateway クラスター経由でジョブを送信します。 -
false: ゲートウェイクラスター経由でジョブを投入しません。デフォルトでは、ジョブはヘッダーノードに投入されます。
説明このパラメーターを
trueに設定しても、ノードのクラスターが Gateway クラスターに関連付けられていない場合、後続の EMR ジョブの送信は失敗します。その他
-
[詳細設定] で カスタム Spark パラメーターを直接追加できます。
spark.eventLog.enabled : falseのようなエントリは、DataWorks によって--conf key=value形式で EMR クラスターに送信されるコードに自動的に追加されます。 -
グローバル Spark パラメーターを設定することもできます。詳細については、「グローバル Spark パラメーターの設定」をご参照ください。
説明Ranger アクセス制御を有効にするには、グローバル Spark パラメーターの設定で
spark.hadoop.fs.oss.authorization.method=ranger設定を追加して、Ranger アクセス制御が有効になるようにします。
EMR Serverless Spark クラスター
パラメーター設定の詳細については、「Spark ジョブ送信用のパラメーターの設定」をご参照ください。
詳細パラメーター
説明
queue
ジョブのスケジューリングキューです。デフォルト値は
dev_queueです。priority
優先度です。デフォルト値は 1 です。
FLOW_SKIP_SQL_ANALYZE
SQL ステートメントの実行モードです。有効な値：
-
true: 複数の SQL ステートメントを一度に実行します。 -
false: SQL 文を 1 つずつ実行します。
説明このパラメーターは、データ開発環境でのテスト実行でのみサポートされています。
SERVERLESS_RELEASE_VERSION
Spark エンジンバージョン。デフォルトでは、ジョブは 管理センター の クラスター管理 でクラスター用に設定された デフォルトのエンジンバージョン を使用します。このパラメーターを設定して、ジョブごとに異なるエンジンバージョンを設定できます。
SERVERLESS_QUEUE_NAME
リソースキュー。デフォルトでは、ジョブは管理センターのクラスター管理でクラスター用に設定されたデフォルトのリソースキューを使用します。リソースの分離および管理に関する要件がある場合は、キューを追加できます。詳細については、「リソースキューの管理」をご参照ください。
その他
-
詳細設定で カスタム Spark パラメーター を追加できます。 例えば、
spark.eventLog.enabled : falseを追加すると、DataWorks は自動的にそのパラメーターを--conf key=value形式で EMR クラスターに送信されるコードに追加します。 -
グローバル Spark パラメーターを設定することもできます。詳細については、「グローバル Spark パラメーターの設定」をご参照ください。
Spark ジョブの実行
-
Run Configuration ペインで、Compute Resource セクションに移動し、Compute Resource と DataWorks Resource Group を設定します。
説明-
タスクに必要なリソースに基づいて、CUs for Scheduling を設定することもできます。デフォルトの CU は
0.25です。 -
パブリックインターネット経由または VPC 内のデータソースにアクセスするには、データソースへのネットワーク接続が確認済みのスケジューリングリソースグループを使用する必要があります。詳細については、「ネットワーク接続ソリューション」をご参照ください。
-
-
ツールバーのパラメーターダイアログボックスで、対応するデータソースを選択し、Run をクリックします。
-
-
ノードジョブを定期的に実行するには、スケジューリングプロパティを設定します。詳細については、「ノードのスケジューリングプロパティの設定」をご参照ください。
説明コンポーネント環境をカスタマイズするには、公式イメージに基づいてカスタム
dataworks_emr_base_task_podを作成し、「カスタムイメージ」および「Data Development でのイメージの使用」をご参照ください。たとえば、カスタムイメージを作成する際に、Spark JAR パッケージを置き換えたり、特定の
libraries、files、またはjar packagesへの依存関係を追加したりできます。 -
ノードジョブを設定した後、ノードをデプロイする必要があります。詳細については、「ノードまたはワークフローのデプロイ」をご参照ください。
-
ジョブのデプロイ後、Operation Center で定期ジョブの実行ステータスを確認できます。詳細については、「Operation Center の開始方法」をご参照ください。
よくある質問
-
Q：ノード接続タイムアウトはどのように解決しますか。
A: リソースグループとクラスター間のネットワーク接続を確認します。コンピューティングリソースページに移動し、リソースを見つけて [リソースの初期化] をクリックします。表示されるダイアログボックスで、[再初期化] をクリックし、初期化が成功したことを確認します。