Vertex AI 上の Ray クラスタでアプリケーションを開発する

• Ray Client の機能を含む Vertex AI SDK for Python のバージョンを使用して、Ray Client 経由で Vertex AI の Ray クラスタに接続します。インタラクティブな Python 開発環境が必要な場合は、このオプションを使用します。

  • Google Cloud コンソールの Colab Enterprise ノートブック内で Vertex AI SDK for Python を使用します。

  • Python セッション、シェル、Jupyter ノートブック内で Vertex AI SDK for Python を使用します。

• Python スクリプトを作成し、Ray Jobs API を使用して Vertex AI の Ray クラスタにそのスクリプトを送信します。プログラムでジョブを送信する場合は、このオプションを使用します。

Ray Client を介して Ray クラスタに接続する

インタラクティブな Ray クライアントを使用するには、Vertex AI の Ray クラスタに接続します。接続環境のネットワークは、クラスタのネットワーク構成によって異なります。クラスタが公共のインターネットにアクセスできる限り、接続環境に制限はありません。つまり、クラスタの作成時に VPC ネットワークが指定されていません。ただし、クラスタが Vertex AI とピアリングされているプライベート VPC ネットワーク上にある場合は、接続環境がクラスタと同じ VPC ネットワーク上にある必要があります。

クライアント側の Ray バージョンは、クラスタの Ray バージョンと一致している必要があります。pip install "google-cloud-aiplatform[ray]" は、デフォルトでクライアントサイドに Ray バージョン 2.47 をインストールします。クラスタの Ray バージョンが 2.42 より古い場合は、pip install ray==2.42.0 を使用して、クライアント側の Ray バージョンをクラスタの Ray バージョンと一致させます。

import ray

# Necessary even if aiplatform.* symbol is not directly used in your program.
from google.cloud import aiplatform
import vertex_ray

import vertexai
vertexai.init()
# The CLUSTER_RESOURCE_NAME is the one returned from vertex_ray.create_ray_cluster.
CLUSTER_RESOURCE_NAME='projects/{}/locations/{}/persistentResources/{}'.format(PROJECT_ID, LOCATION, CLUSTER_NAME)

ray.init('vertex_ray://{}'.format(CLUSTER_RESOURCE_NAME))

Python version:  3.10.12
Ray version: 2.47
Vertex SDK version: 1.46.0
Dashboard: xxxx-dot-us-central1.aiplatform-training.googleusercontent.com

@ray.remote
def square(x):
  print(x)
  return x * x

# Launch four parallel square tasks.
futures = [square.remote(i) for i in range(4)]

print(ray.get(futures))
# Returns [0, 1, 4, 9]

Ray Jobs API を使用してアプリケーションを開発する

このセクションでは、Ray Jobs API を使用して、Python プログラムを Vertex AI の Ray クラスタに送信する方法について説明します。

Python スクリプトを作成する

任意のテキスト エディタで、アプリケーションを Python スクリプトとして開発します。たとえば、次のスクリプトを my_script.py ファイルに配置します。

import ray
import time

@ray.remote
def hello_world():
    return "hello world"

@ray.remote
def square(x):
    print(x)
    time.sleep(100)
    return x * x

ray.init()  # No need to specify address="vertex_ray://...."
print(ray.get(hello_world.remote()))
print(ray.get([square.remote(i) for i in range(4)]))

Ray Jobs API を使用して Ray ジョブを送信する

Python、Ray Jobs CLI、または Ray ダッシュボードの公開アドレスを使用して、Ray ジョブを送信します。

import ray
import vertex_ray
from ray.job_submission import JobSubmissionClient
from google.cloud import aiplatform  # Necessary even if aiplatform.* symbol is not directly used in your program.

CLUSTER_RESOURCE_NAME='projects/{}/locations/REGION/persistentResources/{}'.format(PROJECT_ID, CLUSTER_NAME)

client = JobSubmissionClient("vertex_ray://{}".format(CLUSTER_RESOURCE_NAME))

job_id = client.submit_job(
  # Entrypoint shell command to execute
  entrypoint="python my_script.py",
  # Path to the local directory that contains the my_script.py file.
  runtime_env={
    "working_dir": "./directory-containing-my-script",
    "pip": ["numpy",
            "setuptools<70.0.0",
            "xgboost",
            "ray==CLUSTER_RAY_VERSION", # pin the Ray version to the same version as the cluster
           ]
  }
)

# Ensure that the Ray job has been created.
print(job_id)

from ray.job_submission import JobSubmissionClient
import vertex_ray

DASHBOARD_ADDRESS=DASHBOARD_ADDRESS

client = JobSubmissionClient(
  "vertex_ray://{}".format(DASHBOARD_ADDRESS),
)

job_id = client.submit_job(
  # Entrypoint shell command to execute
  entrypoint="python my_script.py",
  # Path to the local directory that contains the my_script.py file
  runtime_env={
    "working_dir": "./directory-containing-my-script",
    "pip": ["numpy",
            "setuptools<70.0.0",
            "xgboost",
            "ray==CLUSTER_RAY_VERSION", # pin the Ray version to the same version as the cluster
           ]
  }
)
print(job_id)

$ ray job submit --working-dir ./ --address vertex_ray://{CLUSTER_RESOURCE_NAME} -- python my_script.py

長時間実行される Ray ジョブを送信した後、client.get_job_status(job_id) を使用してジョブのステータスをモニタリングする場合は、JobSubmissionClient(client = JobSubmissionClient("vertex_ray://{}".format(CLUSTER_RESOURCE_NAME)) ) を再インスタンス化して認証トークンを更新します。

VPC ピアリングとカスタム サービス アカウントのサポート

Ray on Vertex AI は、デフォルトのサービス エージェントとカスタム サービス アカウントのパブリック ネットワークで Ray Client と Ray Jobs API（JobSubmissionClient）をサポートしています。

次の表に、VPC ネットワークで Ray クラスタを作成する場合の Ray on Vertex AI の VPC ピアリングのサポートを示します。

VPC Service Controls（VPC-SC）では追加の構成が必要です。詳細については、プライベート接続とパブリック接続をご覧ください。

Ray コードでネットワーク ファイル システム（NFS）を使用する

Ray クラスタの作成時に NFS マウントを設定した場合は、アプリケーション コードでこれらの NFS ボリュームを読み書きします。

RayClient

このセクションでは、Ray コードでネットワーク ファイル システム（NFS）を使用する方法について説明します。

1. Python 環境で RayClient を初期化する

   import ray
   from google.cloud import aiplatform
   import vertex_ray
   aiplatform.init(project=PROJECT_ID, location=REGION)
   ray.init(address='vertex_ray://projects/{}/locations/us-central1/persistentResources/{}'.format(PROJECT_NUMBER, PERSISTENT_RESOURCE_ID))

2. ジョブ スクリプトを実行する

   import ray
   import logging
   import os
   import sys
   
   @ray.remote
   def main():
   logging.info("list all files in mounted folder")
   return os.listdir("/mnt/nfs/test")
   
   print(''.join(ray.get(main.remote())))

Python、Ray Jobs CLI、または Ray ダッシュボードの公開アドレスを使用して、Ray ジョブを送信します。詳細については、Vertex AI 上の Ray クラスタでアプリケーションを開発するをご覧ください。

次のステップ

• BigQuery と Ray on Vertex AI を使用する

• Vertex AI にモデルをデプロイして推論を取得する

• Vertex AI 上の Ray クラスタのログを表示する

• Ray クラスタを削除する