Interactive Ray development with DMS Notebook

Step 1: Install dependencies

1. Check the Python version of your notebook environment.

   !python --version

2. Uninstall any existing versions of Ray.

   !pip uninstall -y ray

3. Install the required libraries.

   !pip install "ray[client]==2.49.2" xgboost_ray pandas scikit-learn ipywidgets tqdm

Step 2: Get Ray connection information

1. On the details page of the target AnalyticDB for MySQL cluster, in the navigation pane on the left, click Cluster Management > Resource Management.

2. Click the Resource Groups tab. In the Actions column for the target resource group, click the icon and select Details.

3. Get the Ray Cluster Endpoint. The format is http://[RAY_INTERNAL_HOST]:8265.

Step 3: Connect and check versions

1. Initialize the Ray connection.

   Modify the protocol and port of the Ray cluster connection address obtained in Step 2 to ray://[RAY_INTERNAL_HOST]:10001. Replace remote_url in the following code sample with this address.

   import ray
   import os
   
   # Define runtime environment dependencies
   runtime_env = { "pip": ["xgboost_ray", "pandas", "scikit-learn"] }
   
   remote_url = "ray://<ray_host>:10001"
   ray.init(address=remote_url, runtime_env=runtime_env)
   
   print(f"Successfully connected to the cluster! Ray cluster node count: {len(ray.nodes())}")

2. Check version consistency between the client and server.

   import ray
   import sys
   
   def print_ray_versions():
       local_version = ray.__version__
       local_python = sys.version.split()[0]
       @ray.remote
       def get_server_info():
           import ray
           import sys
           return {
               "version": ray.__version__,
               "python": sys.version.split()[0]
           }
   
       try:
           server_info = ray.get(get_server_info.remote())
           
           print(f"{'Component':<10} | {'Ray Version':<15} | {'Python Version'}")
           print("-" * 45)
           print(f"{'Local(Client)':<10} | {local_version:<15} | {local_python}")
           print(f"{'Remote(Server)':<10} | {server_info['version']:<15} | {server_info['python']}")
           
           # Check for version compatibility
           if local_version != server_info['version']:
               print(f"\n Warning: Version mismatch! This may cause serialization errors or AttributeError.")
           else:
               print(f"\n Versions are fully matched.")
               
       except Exception as e:
           print(f" Failed to connect to cluster or execute task: {e}")
   
   # Run the version check
   print_ray_versions()

Step 4: Interactive development

from ray.util.multiprocessing import Pool

def square(x):
    return x * x

# The usage is identical to Python's native multiprocessing. The @ray.remote decorator is not needed.
with Pool() as pool:
    # This line automatically distributes the execution to the remote cluster.
    results = pool.map(square, range(10))
    print(f"Pool results: {results}")

@ray.remote
def add(x, y):
    return x + y

# asynchronous call - immediately returns an ObjectRef, which is similar to a Promise or Future.
ref1 = add.remote(1, 2)
ref2 = add.remote(3, 4)

# Block only when the actual result is needed.
final_result = ray.get(add.remote(ref1, ref2)) 
print(f"Task result: {final_result}")

@ray.remote
class Counter:
    def __init__(self):
        self.value = 0
    def increment(self):
        self.value += 1
        return self.value

# Create a persistent "counter" object on the cluster.
counter_actor = Counter.remote()

# Call the same Actor instance multiple times. The state is preserved.
results = [counter_actor.increment.remote() for _ in range(3)]
print(f"Actor counter results: {ray.get(results)}") # Output: [1, 2, 3]