DataWorks：開發ODPS Spark任務

使用限制

當ODPS Spark節點選擇Spark3.x版本時，若提交節點報錯，請購買使用Serverless資源群組，詳情請參見新增和使用Serverless資源群組。

準備工作

ODPS Spark節點支援使用Java/Scala和Python語言運行Spark on MaxCompute離線作業，不同語言開發步驟及配置介面存在差異，您可根據業務需要選擇使用。

Java/Scala

在ODPS Spark節點執行Java或Scala語言類型代碼前，您需先在本地開發好Spark on MaxCompute作業代碼，再通過DataWorks上傳為MaxCompute的資源。步驟如下：

1. 準備開發環境。

   根據所使用系統類別型，準備運行Spark on MaxCompute任務的開發環境，詳情請參見搭建Linux開發環境、搭建Windows開發環境。

2. 開發Java/Scala代碼。

   在ODPS Spark節點執行Java或Scala語言類型代碼前，需先在本地或已有環境開發好Spark on MaxCompute代碼，建議使用Spark on MaxCompute提供的專案樣本工程模板進行開發。

3. 打包代碼並上傳至DataWorks。

   代碼開發完成後，需將其打包，並通過DataWorks上傳為MaxCompute資源，詳情請參見建立並使用MaxCompute資源。

Python（使用預設Python環境實現）

Python（使用自訂Python環境實現）

若平台提供的預設Python環境無法滿足您的業務需求，則可根據如下步驟自訂Python環境，執行Spark on MaxCompute任務。

1. 本地準備Python環境。

   您可參考PySpark Python版本和依賴支援，根據業務需要配置可用的Python環境。

2. 打包環境並上傳至DataWorks。

   將Python環境壓縮為一個Zip包，並通過DataWorks上傳為MaxCompute資源，作為後續運行Spark on MaxCompute任務的執行環境。詳情請參見建立並使用MaxCompute資源。

配置項說明

DataWorks運行Spark on MaxCompute離線作業採用Cluster模式，在Cluster模式中，您需指定自訂程式入口main。main運行結束（即狀態為Success或Fail）時，對應的Spark作業便會結束。此外，spark-defaults.conf中的配置需逐條加到ODPS Spark節點配置項中。例如，Executor的數量、記憶體大小和spark.hadoop.odps.runtime.end.point的配置。

編輯代碼：簡單樣本

以下以一個簡單樣本為您介紹ODPS Spark節點的使用：判斷一個字串是否可以轉換為數字。

1. 建立資源。

   1. 在資料開發頁面建立Python類型的資源，並命名為spark_is_number.py，詳情請參見建立並使用MaxCompute資源。代碼如下：

      # -*- coding: utf-8 -*-
      import sys
      from pyspark.sql import SparkSession
      
      try:
          # for python 2
          reload(sys)
          sys.setdefaultencoding('utf8')
      except:
          # python 3 not needed
          pass
      
      if __name__ == '__main__':
          spark = SparkSession.builder\
              .appName("spark sql")\
              .config("spark.sql.broadcastTimeout", 20 * 60)\
              .config("spark.sql.crossJoin.enabled", True)\
              .config("odps.exec.dynamic.partition.mode", "nonstrict")\
              .config("spark.sql.catalogImplementation", "odps")\
              .getOrCreate()
      
      def is_number(s):
          try:
              float(s)
              return True
          except ValueError:
              pass
      
          try:
              import unicodedata
              unicodedata.numeric(s)
              return True
          except (TypeError, ValueError):
              pass
      
          return False
      
      print(is_number('foo'))
      print(is_number('1'))
      print(is_number('1.3'))
      print(is_number('-1.37'))
      print(is_number('1e3'))

   2. 儲存並提交資源。

2. 在已建立的ODPS Spark節點中，根據配置項說明配置節點參數和調度配置參數，並儲存提交節點。

   配置項	說明
   spark版本	Spark2.x
   語言	Python
   選擇主python資源	在下拉式清單中選擇上述已建立的python資源spark_is_number.py

3. 進入開發環境的營運中心，執行補資料，具體操作請參見執行補資料並查看補資料執行個體（新版）。

   說明

   由於資料開發中的ODPS Spark節點沒有運行入口，因此您需要在開發環境的營運中心執行Spark任務。

4. 查看返回結果。

   待補資料執行個體運行成功後，進入其作業記錄的tracking URL中查看運行結果，如下：

   False
   True
   True
   True
   True

編輯代碼：進階樣本

更多情境的Spark on MaxCompute任務開發，請參考：

• java/scala樣本：Spark-1.x樣本

• java/scala樣本：Spark-2.x樣本

• Python樣本：PySpark開發樣本

• 情境：Spark訪問VPC執行個體

• 情境：Spark訪問OSS