Lindorm:Python ジョブ開発: 実践ガイド

Spark Python ジョブ開発プロセス

1. Python ベースの Spark ジョブの定義

2. Python ベースの Spark ジョブのパッケージ化

3. Python ベースの Spark ジョブファイルのアップロード

4. Python ベースの Spark ジョブの送信

ステップ 1: Python ベースの Spark ジョブの定義

1. 「[サンプル Spark ジョブ]」をクリックして、サンプルパッケージをダウンロードします。

2. ダウンロードしたパッケージを展開します。展開されたフォルダの名前は lindorm-spark-examples です。「lindorm-spark-examples/python」ディレクトリに移動し、Python のディレクトリ構造を確認します。

3. このセクションでは、your_project をプロジェクトのルートディレクトリと仮定した場合のプロジェクトのディレクトリ構造について説明します。

   1. your_project ディレクトリに、__init__.py という名前の空のファイルを作成します。

   2. エントリファイルを変更します。

      1. your_project ディレクトリのパスを sys.path に追加します。詳細については、「lindorm-spark-examples/python/your_project/main.py」の Notice1 セクションをご参照ください。

         # Notice1: You need to do the following step to complete the code modification:
         # Step1: Please add a "__init__.py" to your project directory, your project will act as a module of launcher.py
         # Step2: Please add current dir to sys.path, you should add the following code to your main file
         current_dir = os.path.abspath(os.path.dirname(__file__))
         sys.path.append(current_dir)
         print("current dir in your_project: %s" % current_dir)
         print("sys.path: %s \n" % str(sys.path))

      2. エントリーロジックを main(argv) メソッドにカプセル化します。詳細については、「lindorm-spark-examples/python/your_project/main.py」の Notice2 セクションをご参照ください。

         # Notice2: Move the code in `if __name__ == "__main__":` branch to a new defined main(argv) function,
         # so that launcher.py in parent directory just call main(sys.argv)
         def main(argv):
             print("Receive arguments: %s \n" % str(argv))
         
             print("current dir in main: %s \n" % os.path.abspath(os.path.dirname(__file__)))
             # Write your code here
         
         
         if __name__ == "__main__":
             main(sys.argv)

   3. エントリーファイルを作成して、main(argv) メソッドを呼び出します。ルートディレクトリ your_project で、launcher.py という名前のファイルを作成します。lindorm-spark-examples/python/launcher.py からコードをコピーできます。

ステップ 2: Python ベースの Spark ジョブのパッケージ化

1. プロジェクトが依存する Python ランタイム環境とサードパーティライブラリをパッケージ化します。これらの依存関係を tar ファイルにパッケージ化するには、Conda または Virtualenv を使用することを推奨します。詳細については、「Python パッケージ管理」をご参照ください。

   重要

   • `spark.archives` パラメーターを使用して、Conda または Virtualenv によって作成された tar ファイルを渡します。`spark.archives` でサポートされているすべてのフォーマットが有効です。詳細については、「spark.archives」をご参照ください。

   • Lindorm コンピュートエンジンが Python バイナリファイルを認識できるように、このステップは Linux で実行してください。

2. プロジェクトファイルをパッケージ化します。your_project ディレクトリを .zip または .egg ファイルに圧縮します。

   • 次のコマンドを実行して、.zip ファイルを作成します。

     zip -r project.zip your_project

   • 「.egg」ファイルを作成するには、「Eggs のビルド」をご参照ください。

ステップ 3: Python ベースの Spark ジョブファイルのアップロード

ステップ 4: Python ベースの Spark ジョブの送信

Python ジョブ開発例

1. 「サンプル Spark ジョブ」をクリックして、ファイルをダウンロードして展開します。

2. 「your_project/main.py」ファイルを開き、エントリーポイントを変更します。

   1. your_project ディレクトリを sys.path に追加します。

      current_dir = os.path.abspath(os.path.dirname(__file__))
      sys.path.append(current_dir)
      print("current dir in your_project: %s" % current_dir)
      print("sys.path: %s \n" % str(sys.path))

   2. エントリーロジックを main.py ファイルに追加します。以下の例では、SparkSession を初期化します。

      from pyspark.sql import SparkSession
      spark = SparkSession \
          .builder \
          .appName("PythonImportTest") \
          .getOrCreate()
      print(spark.conf)
      spark.stop()

3. Python ディレクトリで、your_project ディレクトリを ZIP ファイルに圧縮します。

   zip -r your_project.zip your_project

4. Linux で、Conda を使用して Python ランタイム環境をパッケージ化します。

   conda create -y -n pyspark_conda_env -c conda-forge numpy conda-pack
   conda activate pyspark_conda_env
   conda pack -f -o pyspark_conda_env.tar.gz

5. 「your_project.zip」、「pyspark_conda_env.tar.gz」、および「launcher.py」を OSS にアップロードします。

6. 次のいずれかの方法でジョブを送信します。

   • Lindorm コンソールでジョブを送信します。詳細については、「コンソールでのジョブ管理」をご参照ください。

   • DMS を使用してジョブを送信します。詳細については、「DMS を使用したジョブ管理」をご参照ください。

ジョブ診断

Python ジョブを送信した後、Jobs ページでそのステータスと Spark UI アドレスを確認できます。 詳細については、「ジョブの表示」をご参照ください。 送信中に問題が発生した場合は、チケットを送信してください。 サポートスタッフにジョブ ID と Spark UI アドレスを提供してください。