E-MapReduce:Ray クイックスタート

前提条件

Ray クラスターが作成されて実行中であること、およびクラスターの [Endpoint Information] ページからエンドポイントとトークンを取得済みであることが必要です。詳細については、「Ray クラスターの管理」をご参照ください。

サブミッション方法の選択

Ray クラスターは、以下の 3 つのサブミッション方法をサポートしています。シナリオに基づいていずれかを選択してください。

インタラクティブな開発

クイックスタートやデバッグに最適です。EMR Serverless Spark コンソールのノートブック環境から、追加のネットワーク設定なしで Ray クラスターに直接接続できます。ローカルマシンから接続するには、マシンと Ray クラスター間のネットワーク接続があり、Python 3.12 環境が設定されていることを確認してください。

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

   pip install ray[client]==2.47.1

2. クラスターの [Endpoint Information] ページで、[gRPC address] と [token] を取得します。

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)

   Ray クラスターへの接続と計算タスクの実行が成功すると、ターミナルに次のような出力が表示されます：

   2025-12-22 17:32:04,514 INFO util.py:154 — Missing packages: ['ipywidgets']. Run `pip install -U ipywidgets`, then restart the notebook server for rich notebook output.
   2025-12-22 17:32:04,537 INFO client_builder.py:242 — Passing the following kwargs to ray.init() on the server: log_to_driver
   2025-12-22 17:32:05,842 WARNING utils.py:1565 — Python patch version mismatch: The cluster was started with:
       Ray: 2.47.1
       Python: 3.12.9
   This process on Ray Client was started with:
       Ray: 2.47.1
       Python: 3.12.12
   
   Square results: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

SDK サブミッション

アプリケーションコードに Ray ジョブのサブミッションを埋め込む場合に適しています。ローカル環境には Python 3.12 がインストールされている必要があります。

1. Ray ジョブサブミッションクライアントライブラリをインストールします：

   pip install "ray[default]==2.47.1"

2. クラスターの [Endpoint Information] ページで、[public endpoint] と [token] を取得します。

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) 

   スクリプトを実行すると、ターミナルに次のような出力が表示されます：

   (.venv) xxx % python '/Users/xxx/xxx.py'
   Submitted job with ID: raysubmit_9Q5JTEgG8tqLEJb4
   Job status: PENDING
   Job status: PENDING
   Job status: SUCCEEDED
   Job logs:
   2026-04-03 17:01:15,126 INFO job_manager.py:531 -- Runtime env is setting up.
   Hello from Ray Client!

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)

コマンドラインサブミッション

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

1. Ray ジョブサブミッションクライアントライブラリをインストールします：

   pip install "ray[default]==2.47.1"

2. クラスターの [Endpoint Information] ページで、[public endpoint] と [token] を取得します。

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 Core 分散ソート
   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 公式ドキュメント」をご参照ください。