DataWorks:EMR Spark SQL ノード

前提条件

• ノードのコンポーネント環境をカスタマイズするには、公式の 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 関連のガバナンスタスクを実行したりすることはできません。

• Spark on ACK クラスターはデータリネージの表示をサポートしていません。EMR Serverless Spark クラスターはデータリネージの表示をサポートしています。

• UI 上での関数登録は、DataLake およびカスタムクラスターではサポートされていますが、Spark on ACK または EMR Serverless Spark クラスターではサポートされていません。

注意事項

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

• この機能は、デフォルトイメージを使用する Spark タスクを実行する場合、デフォルトで利用可能です。

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

操作手順

1. EMR Spark SQL ノードの設定タブで、以下の開発操作を実行します。

   SQL コードの開発

   SQL エディターで、タスクコードを記述します。 ${variable_name} 形式を使用して変数を定義し、Scheduling Settings ペインの Scheduling Parameters セクションでその値を割り当て、スケジュールされたタスクにパラメーターを動的に渡します。 スケジューリングパラメーターの使用方法の詳細については、「スケジューリングパラメーターのソースと式」をご参照ください。 次のコードは一例です。

   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
   ); -- スケジューリングパラメーターと併用できます。

   (オプション) EMR ノードパラメーターの設定

   Run Configuration ペインの EMR Node Parameters セクションで、以下のパラメーターを設定します。

   • [Spark parameter]：Spark のネイティブプロパティです。詳細については、「Spark 設定」をご参照ください。Serverless Spark から直接 Spark 設定テンプレートをロードできます。これにより、設定が簡素化され、一貫性が確保されます。

   • [DataWorks parameters]：使用可能な詳細パラメーターは、EMR クラスターのタイプによって異なります。次の表で各パラメーターについて説明します。

   DataLake またはカスタム on ECS

   パラメーター	説明
   queue	タスクが送信されるスケジューリングキューです。デフォルトのキューは default です。EMR YARN の詳細については、「基本的なキュー設定」をご参照ください。
   priority	タスクの優先度です。デフォルト値：1。
   FLOW_SKIP_SQL_ANALYZE	SQL ステートメントの実行モードです。有効な値： true: 複数の SQL 文を一度に実行します。 false (デフォルト)：一度に 1 つの SQL ステートメントを実行します。 説明 このパラメーターは、DataStudio 環境でのテスト実行にのみ適用されます。
   ENABLE_SPARKSQL_JDBC	SQL コードの送信方法です。有効な値： true: SQL コードを JDBC 経由で送信します。コードは、Kyuubi サービスがない場合は Spark Thrift サーバーに、ある場合は Kyuubi に送信されます。このモードでは、カスタム Spark パラメーターがサポートされています。 どちらの方法もメタデータのデータリネージをサポートします。ただし、Thrift サーバーに送信されたタスクの場合、メタデータに対応するノードタスクの出力情報は利用できません。 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	このパラメーターは、DataStudio でのテスト実行にのみ適用されます。有効な値： true: SQL 文が実行されるたびに、新しい JDBC 接続が作成されます。 false (デフォルト): 単一ノードで異なる SQL ステートメントが実行される場合、同じ JDBC 接続が再利用されます。 説明 このパラメーターを false に設定した場合、Hive の yarn applicationId はログに出力されません。yarn applicationId を出力するには、このパラメーターを true に設定します。
   その他	カスタムの Spark 設定パラメーターです。Spark 固有のプロパティを追加できます。 形式: "spark.eventLog.enabled" : false、"spark.eventLog.memory" : "12g"。 DataWorks は、--conf key=value 形式で、EMR クラスターに送信されるコードにパラメーターを自動的に追加します。 パラメーター設定の詳細については、「グローバル Spark パラメーターを設定する」をご参照ください。 説明 DataWorks は、ワークスペースレベルでグローバル Spark パラメーターをサポートしています。このパラメーターは、異なるモジュール間で共有されます。これらのグローバルパラメーターがモジュール固有の Spark パラメーターを上書きするかどうかを設定できます。 Ranger アクセス制御を有効にするには、グローバル Spark パラメーターを設定するで spark.hadoop.fs.oss.authorization.method=ranger 設定を追加します。

   EMR Serverless Spark

   パラメーター設定の詳細については、「Spark タスクのパラメーター設定」をご参照ください。

   パラメーター	説明
   FLOW_SKIP_SQL_ANALYZE	SQL ステートメントの実行モードです。有効な値： true：複数の SQL ステートメントを一度に実行します。 false (デフォルト): 一度に 1 つの SQL 文を実行します。 説明 このパラメーターは、DataStudio 環境でのテスト実行にのみ適用されます。
   DATAWORKS_SESSION_DISABLE	タスクの送信方法を指定します。デフォルトでは、DataStudio で実行されるタスクは SQL セッションに送信されます。このパラメーターを使用すると、代わりにタスクをキューに送信できます。 true: タスクは実行のためにキューに送信されます。デフォルトでは、コンピューティングリソースをバインドする際に指定したデフォルトのキューが使用されます。DATAWORKS_SESSION_DISABLE パラメーターが true に設定されている場合、SERVERLESS_QUEUE_NAME パラメーターを設定することで、Data Studio でタスクを実行するために送信するキューを指定できます。 false (デフォルト): タスクを SQL セッションに送信して実行します。 説明 このパラメーターは、DataStudio でタスクを実行する場合にのみ有効です。スケジュールされたタスクには影響しません。
   SERVERLESS_RELEASE_VERSION	Spark エンジンバージョン。デフォルトでは、タスクは管理センターのCompute Resourceでクラスターに設定されているデフォルトのエンジンバージョンを使用します。特定のタスクで異なるエンジンバージョンを使用する場合に、このパラメーターを設定します。 説明 SERVERLESS_RELEASE_VERSION パラメーターは、登録済みクラスターに指定された SQL セッションが EMR Serverless Spark コンソールで停止状態である場合にのみ有効になります。
   SERVERLESS_QUEUE_NAME	タスク送信用のリソースキューを指定します。 デフォルトでは、タスクは 管理センター 配下の クラスター管理 でクラスター用に設定された デフォルトのリソースキュー を使用します。 リソースを分離して管理するために、キューを追加できます。 詳細については、「リソースキューを管理する」をご参照ください。 設定方法： ノードパラメーターを設定して、タスクの送信先リソースキューを指定します。 「グローバル Spark パラメーターの設定」でタスクの送信先リソースキューを指定します。 説明 SERVERLESS_QUEUE_NAME パラメーターは、登録済みクラスターに指定された SQL セッションが EMR Serverless Spark コンソールで停止状態である場合にのみ有効になります。 Data Studio での実行時: タスクを実行キューに送信するには、まず DATAWORKS_SESSION_DISABLE を true に設定する必要があります。 タスクキューを指定するために設定する SERVERLESS_QUEUE_NAME パラメーターは、この設定を有効にした場合にのみ有効になります。 オペレーションセンターのスケジュールされたタスクの場合：タスクは常にキューに送信されて実行され、SQL セッションには送信できません。
   SERVERLESS_SQL_COMPUTE	使用する SQL セッション。デフォルトでは、管理センター の Compute Resource でクラスター用に設定された デフォルトのSQL Compute が使用されます。異なるタスクに異なる SQL セッションを使用するには、このパラメーターを設定します。SQL セッションを作成および管理するには、「SQL セッションの管理」をご参照ください。
   その他	カスタムの Spark 設定パラメーターです。Spark 固有のプロパティを追加できます。 形式: "spark.eventLog.enabled": false。 DataWorks は、EMR クラスターに送信されたコードに、 --conf key=value 形式でパラメーターを自動的に追加します。 説明 DataWorks では、ワークスペースレベルでグローバル Spark パラメーターを設定できます。このパラメーターは、DataWorks の様々なモジュールで利用されます。グローバル Spark パラメーターが、モジュール固有の Spark パラメーターより優先されるかどうかを指定できます。グローバル Spark パラメーターの設定方法の詳細については、「グローバル Spark パラメーターの設定」をご参照ください。

   Spark on ACK

   パラメーター	説明
   FLOW_SKIP_SQL_ANALYZE	SQL ステートメントの実行モードです。有効な値： true: 複数の SQL文を一度に実行します。 false (デフォルト): SQL ステートメントを 1 つずつ実行します。 説明 このパラメーターは、DataStudio 環境でのテスト実行にのみ適用されます。
   その他	カスタムの Spark 設定パラメーターです。Spark 固有のプロパティを追加できます。 形式: "spark.eventLog.enabled": false。 DataWorks は、EMR クラスターに送信されたコードに、パラメーターを --conf key=value 形式で自動的に追加します。 説明 DataWorks では、ワークスペースレベルでグローバル Spark パラメーターを設定できます。このパラメーターは、DataWorks の様々なモジュールで利用されます。グローバル Spark パラメーターが、モジュール固有の Spark パラメーターより優先されるかどうかを指定できます。グローバル Spark パラメーターの設定方法の詳細については、「グローバル Spark パラメーターの設定」をご参照ください。

   Hadoop on ECS

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

   SQL タスクの実行

   1. Run Configuration ペインで、Compute Resource と Resource Group を設定します。

      説明

      • タスクの実行に必要なリソースに基づいて、CUs for Scheduling を設定することもできます。デフォルトの CU は 0.25 です。

      • インターネット経由または仮想プライベートクラウド (VPC) 内のデータソースにアクセスするには、データソースに接続できるスケジューリングリソースグループを使用する必要があります。詳細については、「ネットワーク接続ソリューション」をご参照ください。

   2. ツールバーのパラメーターダイアログボックスで、データソースを選択し、Run をクリックして SQL タスクを実行します。

   (オプション) 出力パラメーターの設定

   現在のノードのクエリ結果を後続のノードに渡すには、Scheduling Settings ペインの Node Context Parameters セクションに移動します。Add Assignment Parameter をクリックします。これにより、outputs という名前の出力パラメーターが自動的に追加され、その値はノードのコード内の最後のクエリの結果になります。詳細については、「ノードコンテキストパラメーターの設定と使用」をご参照ください。

2. ノードタスクをスケジュールするには、ビジネス要件に基づいてスケジューリングプロパティを設定します。詳細については、「ノードのスケジューリング設定」をご参照ください。

   説明

   コンポーネント環境をカスタマイズするには、公式イメージに基づいてカスタム dataworks_emr_base_task_pod を作成し、「カスタムイメージ」および「Data Development でのイメージの使用」をご参照ください。

   たとえば、カスタムイメージを作成する際に、Spark JAR パッケージを置き換えたり、特定の libraries、files、または jar packages への依存関係を追加したりできます。

3. ノードタスクを設定した後、デプロイします。詳細については、「ノードまたはワークフローのデプロイ」をご参照ください。

4. デプロイ後、オペレーションセンターでスケジュールされたタスクのステータスを表示します。詳細については、「オペレーションセンターの概要」をご参照ください。

よくある質問

• EMR クラスターで Kerberos を有効にした後、EMR Spark ノードで YARN-Cluster モードで spark-submit を実行すると、「DlfMetaStoreClientFactory not found」というエラーが表示されるのはなぜですか？

• Data Studio で EMR Spark SQL ノードタスクを実行する際に、タスクを SQL Compute に送信する必要がある場合は、SQL Compute のステータスが Running であることを確認してください。そうでない場合、タスクは失敗します。SQL Compute のステータスを確認するには、「SQL セッションの管理」をご参照ください。