DataWorks:MaxCompute Sparkタスクの開発

制限事項

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

準備

ODPS Sparkノードでは、Java、Scala、または Python を使用して、オフラインのSpark on MaxComputeタスクを開発および実行できます。オフラインのSpark on MaxComputeタスクの開発に必要な操作とパラメータは、使用するプログラミング言語によって異なります。ビジネス要件に基づいてプログラミング言語を選択できます。

Java/Scala

ODPS SparkノードでJavaまたはScalaコードを実行する前に、オンプレミスマシンでSpark on MaxComputeタスクのコードの開発を完了し、コードをMaxComputeリソースとしてDataWorksにアップロードする必要があります。次の手順を実行する必要があります。

1. 開発環境を準備します。

   使用するオペレーティングシステムに基づいて、Spark on MaxComputeタスクを実行する開発環境を準備する必要があります。詳細については、「Linux開発環境のセットアップ」または「Windows開発環境のセットアップ」をご参照ください。

2. JavaまたはScalaコードを開発します。

   ODPS SparkノードでJavaまたはScalaコードを実行する前に、オンプレミスマシンまたは準備された開発環境でSpark on MaxComputeタスクのコードの開発を完了する必要があります。Spark on MaxComputeによって提供される サンプルプロジェクトテンプレート を使用することをお勧めします。

3. 開発したコードをパッケージ化し、DataWorksにアップロードします。

   コードが開発された後、コードをパッケージ化し、パッケージをMaxComputeリソースとしてDataWorksにアップロードする必要があります。詳細については、「MaxComputeリソースの作成と使用」をご参照ください。

プログラミング言語: Python（デフォルトのPython環境を使用）

プログラミング言語: Python（カスタムPython環境を使用）

デフォルトのPython環境がビジネス要件を満たせない場合は、次の手順を実行して、Spark on MaxComputeタスクを実行するためのカスタムPython環境を準備できます。

1. オンプレミスマシンでPython環境を準備します。

   PySpark Pythonバージョンとサポートされている依存関係 を参照して、ビジネス要件に基づいてPython環境を設定できます。

2. Python環境のコードをパッケージ化し、パッケージをDataWorksにアップロードします。

   Python環境のコードをZIP形式でパッケージ化し、パッケージをMaxComputeリソースとしてDataWorksにアップロードする必要があります。これにより、環境内でSpark on MaxComputeタスクを実行できます。詳細については、「MaxComputeリソースの作成と使用」をご参照ください。

パラメータの説明

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

簡単なコード編集例

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

1. リソースを作成します。

   1. DataWorksコンソールの [datastudio] ページで、spark_is_number.pyという名前のPythonリソースを作成します。詳細については、「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バージョン	Spark 2.xを選択します。
   言語	Pythonを選択します。
   メイン Python リソース	作成したPythonリソース spark_is_number.py。

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

   説明

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

4. 結果を表示します。

   データバックフィルインスタンスが正常に実行された後、生成された実行ログの [トラッキング 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にアクセスする