DataWorks:EMR Sparkノードの作成

現在のノードは、OSS REF メソッドを使用してOSSリソースを参照できます。 ノードでタスクを実行すると、DataWorksはノードコードで指定されたOSSリソースを自動的にロードします。 このメソッドは、EMRタスクでJARの依存関係が必要なシナリオ、またはEMRタスクがスクリプトに依存する必要があるシナリオでよく使用されます。

1. JARパッケージを開発します。

   1. コードの依存関係を準備します。

      EMRクラスタにアクセスし、マスターノードの /usr/lib/emr/spark-current/jars/ パスで必要なコードの依存関係を表示できます。 以下の情報は、Spark 3.4.2を例として使用しています。 既存のIntelliJ IDEAプロジェクトを開き、Project Object Model (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>

   2. コードを記述します。 サンプルコード:

      package com.aliyun.emr.example.spark
      
      import org.apache.spark.sql.SparkSession
      
      object SparkMaxComputeDemo {
        def main(args: Array[String]): Unit = {
          // Sparkセッションを作成します。
          val spark = SparkSession.builder()
            .appName("HelloDataWorks")
            .getOrCreate()
      
          // Sparkバージョンを表示します。
          println(s"Spark version: ${spark.version}")
        }
      }

   3. コードをJARファイルにパッケージ化します。

      上記のコードを記述して保存したら、コードをJARファイルにパッケージ化します。 この例では、SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar という名前のファイルが生成されます。

2. JARファイルをアップロードします。

   1. OSSコンソールにログインします。 上部のナビゲーションバーで、目的のリージョンを選択します。 次に、左側のナビゲーションウィンドウで、[バケット] をクリックします。

   2. [バケット] ページで、目的のバケットを見つけ、バケット名をクリックして [オブジェクト] ページに移動します。

      この例では、onaliyun-bucket-2 バケットが使用されます。

   3. [オブジェクト] ページで、[ディレクトリの作成] をクリックして、JARファイルを保存するために使用されるディレクトリを作成します。

      [ディレクトリの作成] パネルで、[ディレクトリ名] を emr/jars に設定し、[OK] をクリックします。

   4. JARファイルを、作成したディレクトリにアップロードします。

      作成したディレクトリに移動します。 [オブジェクトのアップロード] をクリックします。 [アップロードするファイル] セクションで、[ファイルの選択] をクリックし、SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar ファイルを追加します。 次に、[オブジェクトのアップロード] をクリックします。

3. JARファイルを参照します。

   1. 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	実行されるメインクラスの完全名。
      master	Sparkアプリケーションの実行モード。
      ossref ファイルパス	形式: ossref://{endpoint}/{bucket}/{object} endpoint: OSSのエンドポイント。 endpointパラメータを空のままにすると、現在のEMRクラスタと同じリージョンにあるOSSバケット内のリソースのみを参照できます。 bucket: OSSでオブジェクトを保存するために使用されるコンテナ。 各 bucket には一意の名前があります。 OSSコンソールにログインして、現在のログインアカウント内のすべての Bucket を表示できます。 object: bucket に保存されているファイル名またはパス。

   2. EMR Sparkノードでタスクを実行します。

      コードを記述したら、 アイコンをクリックし、作成済みのサーバーレスリソースグループを選択して、EMR Sparkノードでタスクを実行します。 タスクの実行が完了したら、コンソールに表示される アプリケーション ID (例: application_1730367929285_xxxx) を記録します。

   3. 結果を表示します。

      EMR Shellノードを作成し、EMR Shellノードで yarn logs -applicationId application_1730367929285_xxxx コマンドを実行して、実行結果を表示します。