E-MapReduce:Ray クラスターへのジョブの送信

方法1：インタラクティブな開発

この方法は、クイックスタートとデバッグに最適です。EMR Serverless Spark コンソールの Notebook 環境から、追加のネットワーク設定なしで Ray クラスターに直接接続できます。ローカルマシンから接続する場合は、ローカルマシンから Ray クラスターへネットワーク接続できること、および Python 3.12 環境を使用していることを確認してください。

1. Ray クライアントをインストールします。

   pip install ray[client]==2.47.1

2. クラスターの [呼び出し情報] ページで、[gRPC アドレス] と [トークン] を取得します。

3. 次のサンプルコードを使用して、Ray クラスターに接続し、インタラクティブな開発を行います。

   import ray
   import os
   
   def get_metadata():
       headers = {"ray-token": "<your_token>"}
       return [(key.lower(), value) for key, value in headers.items()]
   
   ray.init(address="<your_gRPC_address>", _metadata=get_metadata())
   
   import time
   @ray.remote
   def square(x):
       time.sleep(0.1)
       return x * x
   
   futures = [square.remote(i) for i in range(10)]
   
   results = ray.get(futures)
   print("Square results:", results)

   

方法2：SDKによる送信

この方法は、アプリケーションコードに Ray ジョブの送信を組み込む場合に適しています。ローカル環境に Python 3.12 がインストールされている必要があります。

1. Ray ジョブ送信クライアントライブラリをインストールします。

   pip install "ray[default]==2.47.1"

2. クラスターの[呼び出し情報] ページで、[パブリックエンドポイント] と [トークン] を取得します。

3. 次のサンプルコードを使用して、Ray クラスターに接続し、ジョブを送信します。

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # または、内部ネットワークを使用することもできます。この場合、クライアントがクラスターと同じリージョンおよび VPC にあることが必要です。
   # 内部ネットワーク経由での送信は、安定性向上のため推奨します。
   # client = JobSubmissionClient("http://emr-spark-ray-gateway-cn-beijing-internal.spark.emr.aliyuncs.com", headers=custom_headers)
   
   job_id = client.submit_job(
       entrypoint="python -c 'print(\"Hello from Ray Client!\")'"
   )
   
   print(f"Submitted job with ID: {job_id}")
   
   while True:
       status = client.get_job_status(job_id)
       print(f"Job status: {status}")
       if status.is_terminal():
           break
       time.sleep(1)
   
   logs = client.get_job_logs(job_id)
   print("Job logs:")
   print(logs) 

   

4. (オプション) ジョブからマウントされたディレクトリ内のファイルにアクセスするには、Ray クラスターの作成時にマネージドファイルディレクトリをマウントします。次に、ジョブの送信時にマウントパスを使用してファイルを参照します。次の例では、OSS ディレクトリを /mnt/myoss パスにマウントしていることを前提としています。

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # マウントパスを使用して、OSS内のスクリプトファイルをエントリポイントとして直接参照します。
   job_id = client.submit_job(
       entrypoint="python /mnt/myoss/main.py"
   )
   
   print(f"Submitted job with ID: {job_id}")
   
   while True:
       status = client.get_job_status(job_id)
       print(f"Job status: {status}")
       if status.is_terminal():
           break
       time.sleep(1)
   
   logs = client.get_job_logs(job_id)
   print("Job logs:")
   print(logs)

   説明

   Ray ジョブコード内では、マウントパスを使用してファイルの読み書きを直接行うこともできます。たとえば、/mnt/myoss/test.txt から読み取ったり、/mnt/myoss/ ディレクトリにファイルを書き込んだりできます。このディレクトリに書き込まれたファイルは、対応する OSS パスに同期されます。

   次の例は、Ray ジョブ内でマウントされたディレクトリのファイルの読み書きを行う方法を示しています。

   # 次のコードはRay クラスターに送信されます。
   
   import ray
   import time
   
   ray.init()
   
   @ray.remote
   def read_from_mount_path(i):
       with open('/mnt/myoss/test.txt', 'r', encoding='utf-8') as f:
           content = f.read()
       return content
   
   @ray.remote
   def write_to_mount_path(i):
       content = "Hello world " + str(i)
       with open('/mnt/myoss/output' + str(i) + '.txt', 'w', encoding='utf-8') as f:
           f.write(content)
       return 'Write succeeded'
   
   print("--- Starting remote tasks ---")
   start_time = time.time()
   obj_refs = [read_from_mount_path.remote(i) for i in range(4)]
   obj_refs2 = [write_to_mount_path.remote(i) for i in range(4)]
   results = ray.get(obj_refs)
   results2 = ray.get(obj_refs2)

方法3：コマンドラインによる送信

この方法は、外部システムとの統合に最適です。ローカル環境に Python 3.12 がインストールされている必要があります。

1. Ray ジョブ送信クライアントライブラリをインストールします。

   pip install "ray[default]==2.47.1"

2. クラスターの[呼び出し情報] ページで、[パブリックエンドポイント] と [トークン] を取得します。

3. 次のコマンドを実行して、Ray ジョブを送信します。

   # `ray job submit` を実行する前に、RAY_ADDRESS および RAY_JOB_HEADERS 環境変数を設定する必要があります。
   export RAY_ADDRESS='<your_public_endpoint>'               
   export RAY_JOB_HEADERS='{"ray-token": "<your_token>"}'
   
   # ジョブを送信します
   # 例: 分散ソートを実行するジョブを送信します。
   ray job submit --working-dir "." -- python  test-ray-core-sort.py
   
   # OSSへの読み書きをサポートします。
   ray job submit --working-dir "." -- python test-saving-data-test.py
   
   # OSS-HDFSへの読み書きをサポートします。
   ray job submit --working-dir "." -- python test-saving-data-oss-hdfs-test.py
   
   # コードはマウントされたディレクトリへの読み書きをサポートします。
   ray job submit --working-dir "." -- python  test-saving-data-test-mount.py

その他のコマンドラインパラメーターについては、「Ray ジョブ送信 CLI 公式ドキュメント」をご参照ください。