MaxCompute:MaxCompute Sparkタスクの開発

制限事項

Spark 3.Xバージョンを使用するODPS Sparkノードをコミットしたときにエラーが報告された場合、 チケットを起票

し、テクニカルサポートに連絡して、ノードの実行に使用されるスケジューリング専用リソースグループのバージョンを更新します。

準備

ODPS Sparkノードを使用すると、Java、Scala、またはPythonを使用して、MaxComputeタスクでオフラインSparkを開発および実行できます。 オフライン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タスクのコードの開発を完了する必要があります。 MaxComputeでSparkが提供するサンプルプロジェクトテンプレートを使用することを推奨します。

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

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

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

プログラミング言語: Python (カスタムPython環境を使用する)

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

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

   ビジネス要件に基づいてPython環境を構成するには、PySpark Pythonバージョンとサポートされている依存関係を参照してください。

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

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

说明のパラメータ

DataWorksのクラスターモードで、MaxComputeタスクでオフラインSparkを実行できます。 このモードでは、カスタムアプリケーションのエントリポイントとしてMainメソッドを指定する必要があります。 MainがSuccessまたはFail状態になると、Sparkタスクは終了します。 spark-defaults.confファイルの設定項目をODPS Sparkノードの設定に追加する必要があります。 たとえば、executorsの数、メモリサイズ、spark.hadoop.odps.ru ntime.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:
          # 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 on MaxComputeタスクのパラメーターとスケジューリングプロパティを設定し、ノードを保存してコミットします。

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

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

   説明

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

4. 結果を表示します。

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

   False
   True
   True
   True
   True

高度なコード編集の例

他のシナリオでのMaxComputeタスクでのSparkの開発の詳細については、以下のトピックを参照してください。

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

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

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

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

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