DataWorks:MaxCompute Spark タスクの開発

制限

Spark 3.X バージョンを使用する ODPS Spark ノードをコミットする際にエラーが発生した場合は、サーバーレスリソースグループを購入してください。詳細については、「サーバーレスリソースグループの作成と使用」をご参照ください。

事前準備

MaxCompute Spark ノードを使用して、Java/Scala または Python で Spark on MaxCompute のオフラインタスクを実行できます。開発手順と設定プロセスは言語ごとに異なります。ビジネス要件に基づいて言語を選択してください。

パラメーターの説明

DataWorks は、Spark on MaxCompute のオフラインタスクをクラスターモードで実行します。このモードでは、カスタムアプリケーションのエントリポイントを main メソッドで指定する必要があります。Spark タスクは main メソッドの実行が完了すると、Success または Fail のステータスで終了します。さらに、spark-defaults.conf ファイルの設定を、MaxCompute Spark ノードの設定項目に個別に追加する必要があります。例としては、エグゼキューターインスタンス数、メモリサイズ、spark.hadoop.odps.runtime.end.point 設定などがあります。

簡単なコード編集の例

このセクションでは、ODPS Spark ノードを使用して Spark on MaxCompute タスクを開発する方法を簡単な例で説明します。この例では、文字列が数字に変換できるかどうかを判断する Spark on MaxCompute タスクを開発します。

1. リソースの作成

   1. [DataStudio] ページで、新しい Python リソースを作成し、spark_is_number.py という名前を付けます。詳細については、「MaxCompute リソースの作成と使用」をご参照ください。次のコードを使用します：

      # -*- coding: utf-8 -*-
      import sys
      from pyspark.sql import SparkSession
      try:
          # Python 2の場合
          reload(sys)
          sys.setdefaultencoding('utf8')
      except:
          # Python 3では不要
          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 ノードで、このトピックの「パラメーターの説明」セクションを参照して、MaxCompute Spark タスクのパラメーターとスケジューリングプロパティを設定し、ノードを保存してコミットします。

   パラメーター	説明
   Spark バージョン	Spark2.x を選択します。
   言語	Python を選択します。
   メイン Python リソース	作成した Python リソース spark_is_number.py。

3. 開発環境のオペレーションセンターに移動して、ODPS Spark ノードのデータをバックフィルします。詳細については、「データバックフィルとデータバックフィルインスタンスの表示 (新バージョン)」をご参照ください。

   説明

   DataWorks は、DataStudio で ODPS Spark ノードを実行するためのエントリポイントを提供していません。ODPS Spark ノードは、開発環境のオペレーションセンターで実行する必要があります。

4. 結果の表示

   データバックフィルインスタンスが正常に実行された後、生成された実行ログの [tracking URL] をクリックして結果を表示します。次の情報が返されます：

   False
   True
   True
   True
   True

高度なコード編集の例

他のシナリオでの Spark on MaxCompute タスクの開発に関する詳細については、次のトピックをご参照ください：

• Java または Scala の Spark 1.x の例

• Java または Scala の Spark 2.x の例

• PySpark を使用した Spark on MaxCompute アプリケーションの開発

• Spark on MaxCompute から VPC 内のインスタンスへのアクセス

• Spark on MaxCompute から OSS へのアクセス