如何建立EMR Spark節點 - DataWorks

前提條件

開始進行節點開發前，若您需要定製組件環境，此時即可基於官方鏡像dataworks_emr_base_task_pod建立自訂鏡像，並在資料開發中使用鏡像。

例如：在建立自訂鏡像時替換Spark Jar包或是依賴特定的庫、檔案或jar包。
登入EMR叢集至DataWorks。操作詳情請參見舊版資料開發：綁定EMR計算資源。
（可選，RAM帳號需要）進行任務開發的RAM帳號已被添加至對應工作空間中，並具有開發或空間管理員（許可權較大，謹慎添加）角色許可權，新增成員的操作詳情請參見為工作空間增加空間成員。
已購買資源群組完成資源群組配置。包括綁定工作空間、網路設定等。詳情請參見使用Serverless資源群組。
已建立商務程序。資料開發（DataStudio）基於商務程序對不同開發引擎進行具體開發操作，所以您建立節點前需要先建立商務程序，操作詳情請參見建立商務程序。
如果您在開發工作單位時，需要特定的開發環境支援，可使用DataWorks提供的自訂鏡像功能，定製化構建任務執行所需的組件鏡像。更多資訊，請參見自訂鏡像。

使用限制

僅支援使用Serverless資源群組（推薦）或獨享調度資源群組運行該類型任務，若在資料開發時需使用鏡像，則必須使用Serverless資源群組。
DataLake或自訂叢集若要在DataWorks管理中繼資料，需先在叢集側配置EMR-HOOK。若未配置，則無法在DataWorks中即時展示中繼資料、產生審計日誌、展示血緣關係、開展EMR相關治理任務。配置EMR-HOOK，詳情請參見配置Spark SQL的EMR-HOOK。
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中開發Spark任務代碼並完成任務代碼的編譯，產生編譯後的任務JAR包，EMR Spark任務的開發指導詳情請參見Spark概述。

說明

後續您需要將任務JAR包上傳至DataWorks，在DataWorks中周期性調度EMR Spark任務。

一、建立EMR Spark節點

進入資料開發頁面。
登入DataWorks控制台，切換至目標地區後，單擊左側導覽列的數據開發與營運 > 資料開發，在下拉框中選擇對應工作空間後單擊進入資料開發。
建立EMR Spark節點。
1. 按右鍵目標商務程序，選擇新建節點 > EMR > EMR Spark。
  
  說明
  您也可以滑鼠移至上方至新建，選擇新建節點 > EMR > EMR Spark。
2. 在新建節點對話方塊中，輸入名称，並選擇引擎實例、節點類型及路徑。單擊確認，進入EMR Spark節點編輯頁面。
  
  說明
  節點名稱支援大小寫字母、中文、數字、底線（_）和小數點（.）。

二、開發Spark任務

在EMR Spark節點編輯頁面雙擊已建立的節點，進入任務開發頁面，您可以根據不同情境需求選擇適合您的操作方案：

（推薦）先從本地上傳資源至DataStudio，再引用資源。詳情請參見方案一：先上傳資源後引用EMR JAR資源。
使用OSS REF方式引用OSS資源，詳情請參見方案二：直接引用OSS資源。

方案一：先上傳資源後引用EMR JAR資源

DataWorks也支援您從本地先上傳資源至DataStudio，再引用資源。EMR Spark任務編譯完成後，您需擷取編譯後的JAR包，建議根據JAR包大小選擇不同方式儲存JAR包資源。

上傳JAR包資源，建立為DataWorks的EMR資源並提交，或直接儲存在EMR的HDFS儲存中（EMR on ACK 類型的Spark叢集及EMR Serverless Spark叢集不支援上傳資源到HDFS）。

JAR包小於500MB時

建立EMR JAR資源。

JAR包小於500MB時，可將JAR包通過本地上傳的方式上傳為DataWorks的EMR JAR資源，便於後續在DataWorks控制台進行可視化管理，建立完成資源後需進行提交，操作詳情請參見建立和使用EMR資源。

在建立資源對話方塊中，設定引擎類型為EMR，選擇引擎執行個體，設定資源類型為EMR JAR，填寫路徑並勾選上傳為EMR資源，選擇儲存路徑類型（OSS一鍵授權或HDFS），選擇檔案來源為本地並單擊點擊上傳上傳檔案，填寫資源名稱（JAR類型檔案名稱需加尾碼名.jar），然後單擊建立。

說明
首次建立EMR資源時，如果您希望JAR包上傳後儲存在OSS中，您需要先參考介面提示進行授權操作。

引用EMR JAR資源。

雙擊建立的EMR Spark節點，開啟EMR Spark 節點的代碼編輯頁面。
在EMR > 資源節點下，找到上述步驟中已上傳的EMR JAR資源，右鍵選擇引用資源。
選擇引用資源後，當前開啟的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資源名稱。

改寫EMR Spark節點代碼，補充spark submit命令，改寫後的樣本如下。

說明

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包中的任務主Class。
spark-examples_2.11-2.4.0.jar：為您實際上傳的EMR JAR資源名稱。

其他參數可參考以上樣本不做修改，您也可執行以下命令查看spark submit的使用協助，根據需要修改spark submit命令。

說明

若您需要在Spark節點中使用spark submit命令簡化的參數，您需要在代碼中自行添加，例如，--executor-memory 2G。
Spark節點僅支援使用YARN的Cluster模式提交作業。
spark submit方式提交的任務，deploy-mode推薦使用cluster模式，不建議使用client模式。

spark-submit --help

spark-submit --help
Process Output>>>
Process Output>>> Options:
Process Output>>>       --master MASTER_URL         spark://host:port, mesos://host:port, yarn,
Process Output>>>                                   k8s://https://host:port, or local (Default: local[*]).
Process Output>>>       --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or
Process Output>>>                                   on one of the worker machines inside the cluster ("cluster")
Process Output>>>                                   (Default: client).
Process Output>>>       --class CLASS_NAME          Your application's main class (for Java / Scala apps).
Process Output>>>       --name NAME                 A name of your application.
Process Output>>>       --jars JARS                 Comma-separated list of jars to include on the driver
Process Output>>>                                   and executor classpaths.
Process Output>>>       --packages                  Comma-separated list of maven coordinates of jars to include
Process Output>>>                                   on the driver and executor classpaths. Will search the local
Process Output>>>                                   maven repo, then maven central and any additional remote
Process Output>>>                                   repositories given by --repositories. The format for the
Process Output>>>                                   coordinates should be groupId:artifactId:version.
Process Output>>>       --exclude-packages          Comma-separated list of groupId:artifactId, to exclude while
Process Output>>>                                   resolving the dependencies provided in --packages to avoid
Process Output>>>                                   dependency conflicts.
Process Output>>>       --repositories              Comma-separated list of additional remote repositories to
Process Output>>>                                   search for the maven coordinates given with --packages.
Process Output>>>       --py-files PY_FILES         Comma-separated list of .zip, .egg, or .py files to place
Process Output>>>                                   on the PYTHONPATH for Python apps.
Process Output>>>       --files FILES               Comma-separated list of files to be placed in the working
Process Output>>>                                   directory of each executor. File paths of these files

JAR包大於等於500MB時

建立EMR JAR資源。

JAR包大於等於500MB時，無法通過本地上傳的方式直接上傳為DataWorks的資源，建議直接將JAR包儲存在EMR的HDFS中，並記錄下JAR包的儲存路徑，以便後續在DataWorks調度Spark任務時引用該路徑。
引用EMR JAR資源。

JAR包儲存在HDFS時，您可以直接在EMR Spark節點中通過代碼指定JAR包路徑的方式引用JAR包。
1. 雙擊建立的EMR Spark節點，開啟EMR Spark 節點的代碼編輯頁面。
2. 編寫spark submit命令，樣本如下。
```
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：為JAR包實際在HDFS中的路徑。
  - org.apache.spark.examples.JavaSparkPi：為您實際編譯的JAR包中的任務主class。
  - 其他參數為實際EMR叢集的參數，需根據實際情況進行修改配置。您也可以執行以下命令查看spark submit的使用協助，根據需要修改spark submit命令。
    重要
    
    若您需要在Spark節點中使用Spark-submit命令簡化的參數，您需要在代碼中自行添加，例如，--executor-memory 2G。
    
    Spark節點僅支援使用Yarn的Cluster模式提交作業。
    
    spark-submit方式提交的任務，deploy-mode推薦使用cluster模式，不建議使用client模式。
```
spark-submit --help
```

方案二：直接引用OSS資源

當前節點可直接通過OSS REF的方式引用OSS資源，在運行EMR節點時，DataWorks會自動載入代碼中的OSS資源至本地使用。該方式常用於“需要在EMR任務中運行JAR依賴”、“EMR任務需依賴指令碼”等情境。

開發JAR資源。

代碼依賴準備。

您可前往EMR叢集，在叢集Master節點的/usr/lib/emr/spark-current/jars/路徑下查看您所需的代碼依賴。下面以Spark3.4.2版本為例，您需在開啟已建立的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}")
  }
}

將代碼打包成JAR檔案。

編輯儲存Scala代碼後，將Scala代碼打包成JAR檔案。樣本產生的JAR包為SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar。

上傳JAR資源。
1. 完成代碼開發後，您需登入OSS管理主控台，單擊所在地區左側導覽列的Bucket列表。
2. 單擊目標Bucket名稱，進入文件管理頁面。
  
  本文樣本使用的Bucket為onaliyun-bucket-2。
3. 單擊新建目錄，建立JAR資源的存放目錄。
  
  配置目錄名為emr/jars，建立JAR資源的存放目錄。
4. 上傳JAR資源至JAR資源的存放目錄。
  
  進入存放目錄，單擊上傳文件，在待上傳檔案地區單擊掃描檔案，添加SparkWorkOSS-1.0-SNAPSHOT-jar-with-dependencies.jar檔案至Bucket，單擊上傳文件。

引用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`	啟動並執行主類全名稱。
`master`	Spark應用程式的運行模式。
`ossref`檔案路徑	格式為`ossref://{endpoint}/{bucket}/{object}` Endpoint：OSS對外服務的訪問網域名稱。Endpoint為空白時，僅支援使用與當前訪問的EMR叢集同地區的OSS，即OSS的Bucket需要與EMR叢集所在地區相同。 Bucket：OSS用於儲存物件的容器，每一個Bucket有唯一的名稱，登入OSS管理主控台，可查看當前登入帳號下所有Bucket。 object：儲存在Bucket中的一個具體的對象（檔案名稱或路徑）。

運行EMR Spark節點任務。

編輯完成後您可單擊表徵圖，選擇您所建立的Serverless資源群組運行EMR Spark節點。待任務執行完成後，記錄控制台列印的applicationIds，例如application_1730367929285_xxxx。

結果查看。

建立EMR Shell節點並在EMR Shell節點執行yarn logs -applicationId application_1730367929285_xxxx命令查看運行結果：

yarn logs -applicationId application_xxx
Process Output>>> LogLastModifiedTime:星期五 十一月 01 16:27:46 +0800 2024
Process Output>>> LogLength:21
Process Output>>> LogContents:
Process Output>>> Spark version: 3.4.2
Process Output>>>

（可選）配置進階參數

您可在節點進階設定處配置Spark特有屬性參數。更多Spark屬性參數設定，請參考Spark Configuration。不同類型EMR叢集可配置的進階參數存在部分差異，具體如下表。

DataLake叢集/自訂叢集：EMR on ECS

進階參數	配置說明
queue	提交作業的調度隊列，預設為default隊列。如果您在註冊EMR叢集至DataWorks工作空間時，配置了工作空間級的YARN資源隊列：如果勾選了全局配置是否優先為是，則實際Spark任務運行時，隊列以註冊EMR叢集時的配置結果為準。如果未勾選，則實際Spark任務運行時，隊列以EMR Spark節點的配置結果為準。關於EMR YARN說明，詳情請參見隊列基礎配置，註冊EMR叢集時的隊列配置詳情請參見設定全域YARN資源隊列。
priority	優先順序，預設為1。
FLOW_SKIP_SQL_ANALYZE	SQL語句執行方式。取值如下： `true`：表示每次執行多條SQL語句。 `false`（預設值）：表示每次執行一條SQL語句。說明該參數僅支援用於資料開發環境測試回合流程。
其他	您可以直接在進階配置裡追加自訂SPARK參數。例如，`"spark.eventLog.enabled":false`，DataWorks會自動在最終下發EMR叢集的代碼中進行補全，格式為：`--conf key=value`。還支援配置全域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語句。說明該參數僅支援用於資料開發環境測試回合流程。
SERVERLESS_RELEASE_VERSION	Spark引擎版本，預設使用管理中心的叢集管理中叢集配置的默認引擎版本。如需為不同任務設定不同的引擎版本，您可在此進行設定。
SERVERLESS_QUEUE_NAME	指定資源隊列，預設使用管理中心的叢集管理中叢集配置的默認資源隊列。如有資源隔離和管理需求，可通過添加隊列實現。詳情請參見管理資源隊列。
其他	您可以直接在進階配置裡追加自訂SPARK參數。例如，`"spark.eventLog.enabled":false`，DataWorks會自動在最終下發EMR叢集的代碼中進行補全，格式為：`--conf key=value`。還支援配置全域Spark參數，詳情請參見設定全域Spark參數。

Spark叢集：EMR ON ACK

進階參數	配置說明
queue	不支援。
priority	不支援。
FLOW_SKIP_SQL_ANALYZE	SQL語句執行方式。取值如下： `true`：表示每次執行多條SQL語句。 `false`：表示每次執行一條SQL語句。說明該參數僅支援用於資料開發環境測試回合流程。
其他	您可以直接在進階配置裡追加自訂SPARK參數。例如，`"spark.eventLog.enabled":false`，DataWorks會自動在最終下發EMR叢集的代碼中進行補全，格式為：`--conf key=value`。還支援配置全域Spark參數，詳情請參見設定全域Spark參數。

Hadoop叢集：EMR on ECS

進階參數	配置說明
queue	提交作業的調度隊列，預設為default隊列。如果您在註冊EMR叢集至DataWorks工作空間時，配置了工作空間級的YARN資源隊列：如果勾選了全局配置是否優先為是，則實際Spark任務運行時，隊列以註冊EMR叢集時的配置結果為準。如果未勾選，則實際Spark任務運行時，隊列以EMR Spark節點的配置結果為準。關於EMR YARN說明，詳情請參見隊列基礎配置，註冊EMR叢集時的隊列配置詳情請參見設定全域YARN資源隊列。
priority	優先順序，預設為1。
FLOW_SKIP_SQL_ANALYZE	SQL語句執行方式。取值如下： `true`：表示每次執行多條SQL語句。 `false`：表示每次執行一條SQL語句。說明該參數僅支援用於資料開發環境測試回合流程。
USE_GATEWAY	設定本節點提交作業時，是否通過Gateway叢集提交。取值如下： `true`：通過Gateway叢集提交。 `false`：不通過Gateway叢集提交，預設提交到header節點。說明如果本節點所在的叢集未關聯Gateway叢集，此處手動設定參數取值為`true`時，後續提交EMR作業時會失敗。
其他	您可以直接在進階配置裡追加自訂SPARK參數。例如，`"spark.eventLog.enabled":false`，DataWorks會自動在最終下發EMR叢集的代碼中進行補全，格式為：`--conf key=value`。還支援配置全域Spark參數，詳情請參見設定全域Spark參數。說明如需啟用Ranger許可權控制，請在設定全域Spark參數中添加配置`spark.hadoop.fs.oss.authorization.method=ranger`，以確保Ranger許可權控制生效。

執行SQL任務

在工具列單擊表徵圖，在參數對話方塊選擇已建立的調度資源群組，單擊運行。
說明
- 訪問公用網路或VPC網路環境的計算資源需要使用與計算資源測試連通性成功的調度資源群組。詳情請參見網路連通方案。
- 如果您後續執行任務需要修改使用的資源群組，您可單擊帶參運行表徵圖，選擇需要更換的調度資源群組。
- 使用EMR Spark節點查詢資料時，返回的查詢結果最大支援10000條資料，並且資料總量不能超過10MB。
單擊表徵圖，儲存編寫的SQL語句。
（可選）煙霧測試 (Smoke Test)。

如果您希望在開發環境進行煙霧測試 (Smoke Test)，可在執行節點提交，或節點提交後執行，煙霧測試 (Smoke Test)，操作詳情請參見執行煙霧測試 (Smoke Test)。

三、配置節點調度

如您需要周期性執行建立的節點任務，可單擊節點編輯頁面右側的調度配置，根據業務需求配置該節點任務的調度資訊。配置詳情請參見任務調度屬性配置概述。

說明

您需要設定節點的重跑屬性和依賴的上游節點，才可以提交節點。
若您需要定製組件環境，可基於官方鏡像dataworks_emr_base_task_pod建立自訂鏡像，並在資料開發中使用鏡像。

例如：在建立自訂鏡像時替換Spark Jar包或是依賴特定的庫、檔案或jar包。

四、發布節點任務

節點任務配置完成後，需執行提交發佈動作，提交發布後節點即會根據調度配置內容進行周期性運行。

單擊工具列中的表徵圖，儲存節點。
單擊工具列中的表徵圖，提交節點任務。
提交時需在提交對話方塊中輸入變更描述，並根據需要選擇是否在節點提交後執行程式碼檢閱。
說明
- 您需設定節點的重跑屬性和依賴的上遊節點，才可提交節點。
- 程式碼檢閱可對任務的代碼品質進行把控，防止由於任務代碼有誤，未經審核直接發布上線後出現任務報錯。如進行程式碼檢閱，則提交的節點代碼必須通過評審人員的審核才可發布，詳情請參見程式碼檢閱。

如您使用的是標準模式的工作空間，任務提交成功後，需單擊節點編輯頁面右上方的發布，將該任務發布至生產環境執行，操作請參見發布任務。

後續步驟

任務提交發布後，會基於節點的配置周期性運行，您可單擊節點編輯介面右上方的營運，進入營運中心查看周期任務的調度運行情況。詳情請參見管理周期任務。

常見問題

為什麼在EMR叢集中啟用了Kerberos後，在EMR Spark節點開發中使用YARN-Cluster模式執行spark-submit時會報錯DlfMetaStoreClientFactory not found？
Q：節點運行出現連線逾時？

A：請確保資源群組和叢集的網路聯通。請進入計算資源的列表頁，單擊資源初始化，在彈窗中單擊再次初始化，確保成功初始化。

DataWorks：建立EMR Spark節點