EMR Spark SQLノードの作成 - DataWorks - Alibaba Cloud ドキュメントセンター

E-MapReduce (EMR) Spark SQL ノードを作成して、分散 SQL クエリエンジンを使用して構造化データを処理します。これにより、タスクの実行効率が向上します。

前提条件

ノードを開発する前に、コンポーネント環境をカスタマイズする必要がある場合は、公式イメージ dataworks_emr_base_task_pod に基づいてカスタムイメージを作成し、DataStudio でカスタムイメージを使用できます。
たとえば、カスタムイメージを作成する際に、Spark JAR パッケージを置き換えたり、特定の libraries、files、または JAR packages を含めたりすることができます。
EMR クラスターが DataWorks に登録されています。詳細については、「レガシーデータ開発: EMR 計算リソースのアタッチ」をご参照ください。
タスクを開発するには、Resource Access Management (RAM) ユーザーが必要です。この RAM ユーザーはワークスペースに追加され、[開発者] または [ワークスペース管理者] ロールを付与されている必要があります。ワークスペース管理者ロールには広範な権限があるため、慎重に付与する必要があります。メンバーの追加方法の詳細については、「ワークスペースへのメンバーの追加」をご参照ください。
リソースグループが購入され、設定されています。これには、リソースグループのワークスペースへのアタッチとネットワークの設定が含まれます。詳細については、「サーバーレスリソースグループの追加と使用」をご参照ください。
ビジネスフローが作成されます。DataStudio では、さまざまなエンジンの開発操作がビジネスフロー内で整理されます。ノードを作成する前に、ビジネスフローを作成する必要があります。詳細については、「ビジネスフローの作成」をご参照ください。
特定の開発環境が必要な場合は、DataWorks のカスタムイメージ機能を使用して、タスク実行用のコンポーネントイメージをビルドできます。詳細については、「カスタムイメージ」をご参照ください。

制限事項

このタイプのノードは、サーバーレスリソースグループまたは専用スケジューリングリソースグループでのみ実行できます。サーバーレスリソースグループの使用を推奨します。DataStudio でイメージを使用する必要がある場合は、サーバーレスコンピューティングリソースグループを使用してください。
DataWorks で DataLake またはカスタムクラスターのメタデータを管理するには、クラスターに EMR-HOOK を設定する必要があります。EMR-HOOK が設定されていない場合、DataWorks でメタデータのリアルタイム表示、監査ログの生成、データリネージの表示、または EMR 関連の管理タスクを実行することはできません。EMR-HOOK の設定方法の詳細については、「Spark SQL 用に EMR-HOOK を設定する」をご参照ください。
EMR on ACK Spark クラスターではデータリネージは利用できません。データリネージは EMR Serverless Spark クラスターで利用できます。
可視化機能を使用して、DataLake クラスターおよびカスタムクラスターに関数を登録できますが、EMR on ACK Spark クラスターまたは EMR Serverless Spark クラスターには登録できません。

注意事項

現在のワークスペースにバインドされている EMR クラスターで Spark の Ranger アクセス制御を有効にした場合:

この機能は、デフォルトイメージを使用する Spark タスクを実行する場合、デフォルトで利用可能です。
カスタムイメージを使用する Spark タスクを実行するには、ヘルプデスクにチケットを送信して、この機能をサポートするようにイメージをアップグレードする必要があります。

1. EMR Spark SQL ノードの作成

DataStudio ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ開発と O&M] > [データ開発] を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ開発へ] をクリックします。
EMR Spark SQL ノードを作成します。
1. ターゲットのビジネスフローを右クリックし、[新しいノード] > [EMR] > [EMR Spark SQL] を選択します。
  説明
  または、[新規] にカーソルを合わせ、[新しいノード] > [EMR] > [EMR Spark SQL] を選択することもできます。
2. [新しいノード] ダイアログボックスで、[名前]、[エンジンインスタンス]、[ノードタイプ]、および [パス] を設定します。[確認] をクリックします。EMR Spark SQL ノードの設定タブが表示されます。
  説明
  ノード名には、大文字、小文字、漢字、数字、アンダースコア (_)、ピリオド (.) を使用できます。

2. EMR Spark SQL タスクの開発

EMR Spark SQL ノード設定タブで、作成したノードをダブルクリックします。タスク開発タブが表示されます。

SQL コードの開発

SQL 編集エリアで、タスクコードを開発します。コード内で ${variable_name} 形式を使用して変数を定義できます。ノード設定タブの右側のナビゲーションウィンドウにある [スケジューリング設定] > [スケジューリングパラメーター] セクションで変数に値を割り当てることができます。これにより、スケジューリングシナリオでパラメーターを動的に渡すことができます。スケジューリングパラメーターの使用方法の詳細については、「サポートされているスケジューリングパラメーターの形式」をご参照ください。以下に例を示します。

SHOW TABLES; 
-- ${var} を使用して var という名前の変数を定義します。この変数に値 ${yyyymmdd} を割り当てると、データタイムスタンプをサフィックスとして持つテーブルを作成できます。
CREATE TABLE IF NOT EXISTS userinfo_new_${var} (
ip STRING COMMENT'IP アドレス',
uid STRING COMMENT'ユーザー ID'
)PARTITIONED BY(
dt STRING
); -- これはスケジューリングパラメーターと共に使用できます。

説明

SQL 文は 130 KB を超えることはできません。
データ開発で複数の EMR 計算リソースがワークスペースにアタッチされている場合は、計算リソースを選択する必要があります。

(オプション) 詳細パラメーターの設定

ノードの [詳細設定] タブで、Spark 固有の属性パラメーターを設定します。Spark 属性パラメーターの詳細については、「Spark Configuration」をご参照ください。設定可能な詳細パラメーターは、次の表に示すように、EMR クラスターのタイプによって異なります。

DataLake クラスター/カスタムクラスター: EMR on ECS

詳細パラメーター	設定の説明
queue	ジョブが送信されるスケジューリングキュー。デフォルトのキューは `default` です。EMR YARN の詳細については、「基本的なキュー設定」をご参照ください。
priority	優先度。デフォルト値は 1 です。
FLOW_SKIP_SQL_ANALYZE	SQL 文の実行モード。有効な値: `true`: 一度に複数の SQL 文が実行されます。 `false` (デフォルト): 一度に 1 つの SQL 文が実行されます。説明このパラメーターは、データ開発環境でのフローのテストと実行にのみ使用できます。
ENABLE_SPARKSQL_JDBC	SQL コードの送信メソッド。有効な値: `true`: Java Database Connectivity (JDBC) を使用して SQL コードを送信します。EMR クラスターに Kyuubi サービスがない場合、SQL コードは Spark Thrift Server に送信されます。EMR クラスターに Kyuubi サービスがある場合、SQL コードは JDBC を使用して Kyuubi に送信され、カスタム Spark パラメーターがサポートされます。どちらのメソッドもメタデータリネージをサポートしますが、Thrift Server に送信されたタスクは、メタデータに対応するノードタスクの出力情報が欠落します。 `false` (デフォルト): `spark-submit cluster` メソッドを使用して SQL コードを送信します。この送信モードでは、Spark 2 と Spark 3 の両方がメタデータリネージと出力情報をサポートします。カスタム Spark パラメーターもサポートされます。説明 Spark-submit cluster 送信モードでは、デフォルトで EMR クラスターの HDFS 上の `/tmp` ディレクトリに一時ファイルとディレクトリが作成されます。このディレクトリに読み取りおよび書き込み権限があることを確認する必要があります。 Spark-submit cluster モードを選択して送信する場合、詳細設定でカスタム SparkConf パラメーターを直接追加できます。コードをコミットすると、DataWorks は新しいパラメーターをコマンドに自動的に追加します。例: `"spark.driver.memory" : "2g"`。
DATAWORKS_SESSION_DISABLE	開発環境でタスクを直接テストおよび実行するシナリオに適用されます。有効な値: `true`: SQL 文が実行されるたびに新しい JDBC 接続が作成されます。 `false` (デフォルト): ユーザーがノードで異なる SQL 文を実行するときに、同じ JDBC 接続が再利用されます。説明このパラメーターを `false` に設定すると、Hive `yarn applicationId` は出力されません。`yarn applicationId` を出力するには、このパラメーターを `true` に設定します。
その他	カスタム Spark Configuration パラメーター。Spark 固有の属性パラメーターを追加します。設定形式は次のとおりです: `"spark.eventLog.enabled":false` 。DataWorks は、次の形式で EMR クラスターに配信されるコードに設定を自動的に追加します: `--conf key=value`。パラメーター設定の詳細については、「グローバル Spark パラメーターの設定」をご参照ください。説明 DataWorks では、グローバル Spark パラメーターを設定できます。つまり、各 DataWorks モジュールが使用する Spark パラメーターをワークスペースレベルで指定できます。これらのグローバル Spark パラメーターの優先度が、特定のモジュール内の Spark パラメーターよりも高いかどうかを指定できます。 Ranger アクセス制御を有効にするには、グローバル Spark パラメーターの設定で `spark.hadoop.fs.oss.authorization.method=ranger` 設定を追加して、Ranger アクセス制御が有効になるようにします。

EMR Serverless Spark クラスター

パラメーター設定の詳細については、「Spark ジョブを送信するためのパラメーターの設定」をご参照ください。

詳細パラメーター	設定の説明
FLOW_SKIP_SQL_ANALYZE	SQL 文の実行モード。有効な値: `true`: 一度に複数の SQL 文が実行されます。 `false` (デフォルト): 一度に 1 つの SQL 文が実行されます。説明このパラメーターは、データ開発環境でのフローのテストと実行にのみ使用できます。
DATAWORKS_SESSION_DISABLE	ジョブの送信メソッド。データ開発でタスクを実行すると、タスクはデフォルトで SQL Compute に送信されて実行されます。このパラメーターを使用して、タスクを SQL Compute で実行するか、キューに送信して実行するかを指定できます。 `true`: タスクはキューに送信されて実行されます。デフォルトでは、計算リソースを関連付けるときに指定されたデフォルトのキューが使用されます。`DATAWORKS_SESSION_DISABLE` パラメーターが `true` に設定されている場合、`SERVERLESS_QUEUE_NAME` パラメーターを設定して、データ開発でタスクが実行のために送信されるキューを指定できます。 `false` (デフォルト): タスクは SQL Compute に送信されて実行されます。説明このパラメーターは、データ開発での実行中にのみ有効であり、スケジュールされたランタイム中には有効になりません。
SERVERLESS_RELEASE_VERSION	Spark DPI エンジンバージョン。デフォルトでは、[管理センター] > [クラスター管理] でクラスターに設定されている [デフォルト DPI エンジンバージョン] が使用されます。タスクごとに異なる DPI エンジンバージョンを設定するには、ここで設定できます。説明 `SERVERLESS_RELEASE_VERSION` パラメーターは、登録済みクラスターで指定された SQL Compute (セッション) が EMR Serverless Spark コンソールで開始されていない場合にのみ、詳細設定で有効になります。
SERVERLESS_QUEUE_NAME	タスクが送信されるリソースキューを指定します。タスクがキューに送信されて実行されるように指定されている場合、デフォルトでは [管理センター] > [クラスター管理] でクラスターに設定されている [デフォルトリソースキュー] が使用されます。リソースの隔離と管理の要件がある場合は、キューを追加できます。詳細については、「リソースキューの管理」をご参照ください。設定方法: ノードパラメーターを設定して、タスク送信用のリソースキューを指定します。グローバル Spark パラメーターを設定して、タスク送信用のリソースキューを指定します。説明 `SERVERLESS_QUEUE_NAME` パラメーターは、登録済みクラスターに指定された SQL Compute (セッション) が EMR Serverless Spark コンソールで開始されていない場合にのみ、詳細設定で有効になります。データ開発でジョブを実行する場合: まず `DATAWORKS_SESSION_DISABLE` を `true` に設定して、ジョブが実行のためにキューに送信されるようにする必要があります。そうして初めて、ジョブキューを指定するために設定した `SERVERLESS_QUEUE_NAME` パラメーターが有効になります。オペレーションセンターでの定期実行中: タスクは強制的にキューに送信されて実行され、SQL Compute に送信して実行することはできません。
SERVERLESS_SQL_COMPUTE	SQL Compute (SQL セッション) を指定します。デフォルトでは、[管理センター] > [クラスター管理] でクラスターに設定されている [デフォルト SQL Compute] が使用されます。タスクごとに異なる SQL セッションを設定するには、ここで設定できます。SQL セッションを作成および管理するには、「SQL セッションの管理」をご参照ください。
その他	カスタム Spark Configuration パラメーター。Spark 固有の属性パラメーターを追加します。設定形式は次のとおりです: `"spark.eventLog.enabled":"false"`。DataWorks は、最終的に EMR クラスターに送信されるコードで `--conf key=value` の形式でコードを自動的に補完します。説明 DataWorks では、グローバル Spark パラメーターを設定できます。つまり、各 DataWorks モジュールが使用する Spark パラメーターをワークスペースレベルで指定できます。これらのグローバル Spark パラメーターの優先度が、特定のモジュール内の Spark パラメーターよりも高いかどうかを指定できます。グローバル Spark パラメーターの設定方法の詳細については、「グローバル Spark パラメーターの設定」をご参照ください。

SQL タスクの保存と実行

ツールバーで、アイコンをクリックして SQL 文を保存します。アイコンをクリックして SQL 文を実行します。

[実行] ダイアログボックスで、ネットワーク接続テストに合格したリソースグループを選択して、DataWorks が Spark サービスにアクセスできることを確認します。ノードコードで変数を使用する場合は、テスト目的で変数に定数を割り当てます。スケジューリング用のスケジューリングパラメーターとリソースグループの設定方法の詳細については、「ノードスケジューリングの設定」をご参照ください。タスクのデバッグの詳細については、「タスクのデバッグプロセス」をご参照ください。

説明

コード内のパラメーター割り当てを変更するには、ツールバーの [詳細実行] をクリックします。パラメーター割り当てロジックの詳細については、「開発環境での実行、詳細実行、スモークテストの割り当てロジックの違いは何ですか？」をご参照ください。

Spark クラスター: EMR on ACK

詳細パラメーター

設定の説明

FLOW_SKIP_SQL_ANALYZE

SQL 文の実行モード。有効な値:

true: 一度に複数の SQL 文が実行されます。
false (デフォルト): 一度に 1 つの SQL 文が実行されます。

説明

このパラメーターは、データ開発環境でのフローのテストと実行にのみ使用できます。

その他

カスタム Spark Configuration パラメーター。Spark 固有の属性パラメーターを追加します。

設定形式は次のとおりです: "spark.eventLog.enabled":false。DataWorks は、最終的に EMR クラスターに送信されるコードで --conf key=value の形式でコードを自動的に補完します。

説明

DataWorks では、グローバル Spark パラメーターを設定できます。つまり、各 DataWorks モジュールが使用する Spark パラメーターをワークスペースレベルで指定できます。これらのグローバル Spark パラメーターの優先度が、特定のモジュール内の Spark パラメーターよりも高いかどうかを指定できます。グローバル Spark パラメーターの設定方法の詳細については、「グローバル Spark パラメーターの設定」をご参照ください。

Hadoop クラスター: EMR on ECS

詳細パラメーター	設定の説明
queue	ジョブが送信されるスケジューリングキュー。デフォルトのキューは `default` です。EMR YARN の詳細については、「基本的なキュー設定」をご参照ください。
priority	優先度。デフォルト値は 1 です。
FLOW_SKIP_SQL_ANALYZE	SQL 文の実行モード。有効な値: `true`: 一度に複数の SQL 文が実行されます。 `false` (デフォルト): 一度に 1 つの SQL 文が実行されます。説明このパラメーターは、データ開発環境でのフローのテストと実行にのみ使用できます。
USE_GATEWAY	このノードのジョブを送信する際に、ゲートウェイクラスター経由でジョブを送信するかどうかを設定します。有効な値: `true`: ゲートウェイクラスター経由で送信します。 `false` (デフォルト): ゲートウェイクラスター経由で送信しません。ジョブはデフォルトでヘッダーノードに送信されます。説明ゲートウェイクラスターに関連付けられていないクラスター内のノードに対してこのパラメーターを手動で `true` に設定すると、後続の EMR ジョブの送信は失敗します。
その他	カスタム Spark Configuration パラメーター。Spark 固有の属性パラメーターを追加します。設定形式は次のとおりです: `"spark.eventLog.enabled":false`。DataWorks は、この設定を EMR クラスターに送信されるコードで `--conf key=value` 形式に自動的に変換します。パラメーター設定の詳細については、「グローバル Spark パラメーターの設定」をご参照ください。説明 DataWorks では、グローバル Spark パラメーターを設定できます。つまり、各 DataWorks モジュールが使用する Spark パラメーターをワークスペースレベルで指定できます。これらのグローバル Spark パラメーターの優先度が、特定のモジュール内の Spark パラメーターよりも高いかどうかを指定できます。 Ranger アクセス制御を有効にするには、グローバル Spark パラメーターの設定で `spark.hadoop.fs.oss.authorization.method=ranger` 設定を追加して、Ranger アクセス制御が有効になるようにします。

SQL タスクの実行

ツールバーでアイコンをクリックします。[パラメーター] ダイアログボックスで、作成したスケジューリングリソースグループを選択し、[実行] をクリックします。
説明
- パブリックネットワークまたは VPC ネットワーク環境の計算リソースにアクセスするには、計算リソースとの接続性テストに合格したスケジューリングリソースグループを使用する必要があります。詳細については、「ネットワーク接続ソリューション」をご参照ください。
- 後続のタスク実行のためにリソースグループを変更するには、[パラメーターを指定して実行] アイコンをクリックし、切り替えたいスケジューリングリソースグループを選択します。
- EMR Spark SQL ノードを使用してデータをクエリする場合、クエリは最大 10,000 レコードを返すことができ、合計データサイズは 10 MB を超えることはできません。
アイコンをクリックして SQL 文を保存します。
(オプション) スモークテストを実行します。
開発環境でスモークテストを実行したい場合は、ノードを送信するとき、またはノードが送信された後に実行できます。詳細については、「スモークテストの実行」をご参照ください。

3. ノードスケジューリングの設定

システムにノード上のタスクを定期的に実行させたい場合は、ノードの設定タブの右側のナビゲーションウィンドウで [プロパティ] をクリックして、ビジネス要件に基づいてタスクスケジューリングプロパティを設定できます。詳細については、「概要」をご参照ください。

説明

4. ノードタスクの公開

ノード上のタスクが設定された後、タスクをコミットしてデプロイする必要があります。タスクをコミットしてデプロイすると、システムはスケジューリング設定に基づいてタスクを定期的に実行します。

上部のツールバーにあるアイコンをクリックしてタスクを保存します。
上部のツールバーにあるアイコンをクリックしてタスクをコミットします。
[送信] ダイアログボックスで、[変更の説明] パラメーターを設定します。次に、ビジネス要件に基づいてタスクをコミットした後にタスクコードをレビューするかどうかを決定します。
説明
- タスクをコミットする前に、[プロパティ] タブで [再実行] および [親ノード] パラメーターを設定する必要があります。
- コードレビュー機能を使用すると、タスクのコード品質を確保し、無効なタスクコードによるタスク実行エラーを防ぐことができます。コードレビュー機能を有効にすると、コミットされたタスクコードは、コードレビューに合格した後にのみデプロイできます。詳細については、「コードレビュー」をご参照ください。

標準モードのワークスペースを使用する場合、タスクをコミットした後、本番環境にタスクをデプロイする必要があります。ノードにタスクをデプロイするには、ノードの設定タブの右上隅にある [デプロイ] をクリックします。詳細については、「ノードのデプロイ」をご参照ください。

次のステップ

タスクをコミットしてデプロイすると、タスクはスケジューリング設定に基づいて定期的に実行されます。対応するノードの設定タブの右上隅にある [オペレーションセンター] をクリックしてオペレーションセンターに移動し、タスクのスケジューリングステータスを表示できます。詳細については、「定期タスクの表示と管理」をご参照ください。

よくある質問

EMR クラスターで Kerberos を有効にした後、EMR Spark ノードで YARN-Cluster モードを使用して spark-submit を実行すると、DlfMetaStoreClientFactory not found エラーが発生するのはなぜですか？
データ開発で EMR Spark SQL ノードタスクを実行するときに、タスクを SQL Compute に送信して実行する必要がある場合は、SQL Compute サービスのステータスが [実行中] であることを確認してください。そうでない場合、タスクは失敗します。SQL Compute のステータスを表示する方法の詳細については、「SQL セッションの管理」をご参照ください。