DataWorks:ODPS Spark ジョブの開発

制限事項

Spark 3.x を使用する ODPS Spark ノード の送信が失敗した場合、サーバーレスリソースグループ を購入して使用する必要があります。詳細については、「サーバーレスリソースグループの使用」をご参照ください。

前提条件

ODPS Spark ノードは、Java/Scala または Python を使用して MaxCompute オフラインジョブで Spark を実行することをサポートしています。開発プロセスおよび構成オプションは、言語によって異なります。

パラメーター

DataWorks は、Cluster Mode で MaxCompute 上の Spark Offline Jobs を実行します。このモードでは、カスタムプログラムのエントリポイントとして、main を指定する必要があります。main メソッドが完了すると Spark ジョブは終了し、Success または Fail ステータスを返します。さらに、spark-defaults.conf の構成を、ODPS Spark ノードの 構成アイテムセクションに 1 つずつ追加する必要があります。例として、エグゼキュータ インスタンスの数、メモリ、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 バージョン	Spark 2.x
   言語	Python
   メイン Python リソースの選択	ドロップダウンリストから、作成した Python リソース（spark_is_number.py）を選択します。

3. 開発環境の オペレーションセンター に移動し、データバックフィル ジョブ を実行します。詳細な手順については、「データバックフィルインスタンスの運用管理」をご参照ください。

   説明

   ODPS Spark ノード は データ開発 で直接実行できないため、開発環境の オペレーションセンター から ジョブ を実行する必要があります。

4. 結果を確認します。

   データバックフィル インスタンス の実行が成功した後、実行ログ 内の トラッキング URL にアクセスして結果を確認します：

   False
   True
   True
   True
   True

高度な例

さまざまなユースケース向けに MaxCompute 上の Spark ジョブ を開発するためのその他の例については、以下のトピックをご参照ください：

• Java/Scala の例：Spark 1.x

• Java/Scala の例：Spark 2.x

• Python の例：PySpark 開発

• ユースケース：Spark から VPC インスタンスへのアクセス

• ユースケース：Spark から OSS へのアクセス