Python中的分布式框架Ray的安装与使用教程

目录

• 技术背景
• 安装
• 启动和连接服务
• 基础使用
• 查看和管理任务
• 总结概要
• 版权声明

$ python3 -m pip install ray[default]

$ ssh-keygen -t rsa
$ ssh-copy-id user_name@ip_address

$ ray start --head --dashboard-host='0.0.0.0' --dashboard-port=8265
Usage stats collection is enabled. To disable this, add `--disable-usage-stats` to the command that starts the cluster, or run the following command: `ray disable-usage-stats` before starting the cluster. See https://docs.ray.io/en/master/cluster/usage-stats.html for more details.
Local node IP: xxx.xxx.xxx.xxx
--------------------
Ray runtime started.
--------------------
Next steps
  To add another node to this Ray cluster, run
    ray start --address='xxx.xxx.xxx.xxx:6379'
  To connect to this Ray cluster:
    import ray
    ray.init()
  To submit a Ray job using the Ray Jobs CLI:
    RAY_ADDRESS='http://xxx.xxx.xxx.xxx:8265' ray job submit --working-dir . -- python my_script.py
  See https://docs.ray.io/en/latest/cluster/running-applications/job-submission/index.html
  for more information on submitting Ray jobs to the Ray cluster.
  To terminate the Ray runtime, run
    ray stop
  To view the status of the cluster, use
    ray status
  To monitor and debug Ray, view the dashboard at
    xxx.xxx.xxx.xxx:8265
  If connection to the dashboard fails, check your firewall settings and network configuration.

$ ray start --address='xxx.xxx.xxx.xxx:6379'

# test_ray.py 
import os
import ray
ray.init()
print('''This cluster consists of
    {} nodes in total
    {} CPU resources in total
'''.format(len(ray.nodes()), ray.cluster_resources()['CPU']))

$ RAY_ADDRESS='http://xxx.xxx.xxx.xxx:8265' ray job submit --working-dir . -- python test_ray.py 
Job submission server address: http://xxx.xxx.xxx.xxx:8265
2024-08-27 07:35:10,751 INFO dashboard_sdk.py:338 -- Uploading package gcs://_ray_pkg_4b79155b5de665ce.zip.
2024-08-27 07:35:10,751 INFO packaging.py:518 -- Creating a file package for local directory '.'.
-------------------------------------------------------
Job 'raysubmit_7Uqy8LjP4dxjZxGa' submitted successfully
-------------------------------------------------------
Next steps
  Query the logs of the job:
    ray job logs raysubmit_7Uqy8LjP4dxjZxGa
  Query the status of the job:
    ray job status raysubmit_7Uqy8LjP4dxjZxGa
  Request the job to be stopped:
    ray job stop raysubmit_7Uqy8LjP4dxjZxGa
Tailing logs until the job exits (disable with --no-wait):
2024-08-27 15:35:14,079 INFO worker.py:1330 -- Using address xxx.xxx.xxx.xxx:6379 set in the environment variable RAY_ADDRESS
2024-08-27 15:35:14,079 INFO worker.py:1458 -- Connecting to existing Ray cluster at address: xxx.xxx.xxx.xxx:6379...
2024-08-27 15:35:14,103 INFO worker.py:1639 -- Connected to Ray cluster. View the dashboard at http://xxx.xxx.xxx.xxx:8265 
This cluster consists of
    1 nodes in total
    48.0 CPU resources in total
------------------------------------------
Job 'raysubmit_7Uqy8LjP4dxjZxGa' succeeded
------------------------------------------

import ray
ray.init()
import numpy as np
print (np.__file__)

$ RAY_ADDRESS='http://xxx.xxx.xxx.xxx:8265' ray job submit --working-dir . -- python test_ray.py 
Job submission server address: http://xxx.xxx.xxx.xxx:8265
2024-08-27 07:46:10,645 INFO dashboard_sdk.py:338 -- Uploading package gcs://_ray_pkg_5bba1a7144beb522.zip.
2024-08-27 07:46:10,658 INFO packaging.py:518 -- Creating a file package for local directory '.'.
-------------------------------------------------------
Job 'raysubmit_kQ3XgE4Hxp3dkmuU' submitted successfully
-------------------------------------------------------
Next steps
  Query the logs of the job:
    ray job logs raysubmit_kQ3XgE4Hxp3dkmuU
  Query the status of the job:
    ray job status raysubmit_kQ3XgE4Hxp3dkmuU
  Request the job to be stopped:
    ray job stop raysubmit_kQ3XgE4Hxp3dkmuU
Tailing logs until the job exits (disable with --no-wait):
2024-08-27 15:46:12,456 INFO worker.py:1330 -- Using address xxx.xxx.xxx.xxx:6379 set in the environment variable RAY_ADDRESS
2024-08-27 15:46:12,457 INFO worker.py:1458 -- Connecting to existing Ray cluster at address: xxx.xxx.xxx.xxx:6379...
2024-08-27 15:46:12,470 INFO worker.py:1639 -- Connected to Ray cluster. View the dashboard at http://xxx.xxx.xxx.xxx:8265 
/home/dechin/anaconda3/envs/mindspore-latest/lib/python3.7/site-packages/numpy/__init__.py
------------------------------------------
Job 'raysubmit_kQ3XgE4Hxp3dkmuU' succeeded
------------------------------------------
$ python3 -m pip show numpy
Name: numpy
Version: 1.21.6
Summary: NumPy is the fundamental package for array computing with Python.
Home-page: https://www.numpy.org
Author: Travis E. Oliphant et al.
Author-email: 
License: BSD
Location: /usr/local/python-3.7.5/lib/python3.7/site-packages
Requires: 
Required-by: CyFES, h5py, hadder, matplotlib, mindinsight, mindspore, mindspore-serving, pandas, ray, scikit-learn, scipy

import ray
ray.init()
@ray.remote(num_returns=1)
def cpu_task():
    import time
    time.sleep(2)
    import numpy as np
    nums = 100000
    arr = np.random.random((null, nums))
    arr2 = arr[1]**2 + arr[0]**2
    pi = np.where(arr2<=1, 1, 0).sum() * 4 / nums
    return pi
num_conc = 10
res = ray.get([cpu_task.remote() for _ in range(num_conc)])
print (sum(res) / num_conc)

$ time RAY_ADDRESS='http://xxx.xxx.xxx.xxx:8265' ray job submit --working-dir . --entrypoint-num-cpus 10 -- python te
st_ray.py 
Job submission server address: http://xxx.xxx.xxx.xxx:8265
2024-08-27 08:30:13,315 INFO dashboard_sdk.py:385 -- Package gcs://_ray_pkg_d66b052eb6944465.zip already exists, skipping upload.
-------------------------------------------------------
Job 'raysubmit_Ur6MAvP7DYiCT6Uz' submitted successfully
-------------------------------------------------------
Next steps
  Query the logs of the job:
    ray job logs raysubmit_Ur6MAvP7DYiCT6Uz
  Query the status of the job:
    ray job status raysubmit_Ur6MAvP7DYiCT6Uz
  Request the job to be stopped:
    ray job stop raysubmit_Ur6MAvP7DYiCT6Uz
Tailing logs until the job exits (disable with --no-wait):
2024-08-27 16:30:15,032 INFO worker.py:1330 -- Using address xxx.xxx.xxx.xxx:6379 set in the environment variable RAY_ADDRESS
2024-08-27 16:30:15,033 INFO worker.py:1458 -- Connecting to existing Ray cluster at address: xxx.xxx.xxx.xxx:6379...
2024-08-27 16:30:15,058 INFO worker.py:1639 -- Connected to Ray cluster. View the dashboard at http://xxx.xxx.xxx.xxx:8265 
3.141656
------------------------------------------
Job 'raysubmit_Ur6MAvP7DYiCT6Uz' succeeded
------------------------------------------
real    0m7.656s
user    0m0.414s
sys     0m0.010s

import ray
ray.init()
@ray.remote(num_returns=1, num_gpus=1)
def test_ms():
    import os
    os.environ['GLOG_v']='4'
    os.environ['CUDA_VISIBLE_DEVICE']='0'
    import mindspore as ms
    ms.set_context(device_target="GPU", device_id=0)
    a = ms.Tensor([1, 2, 3], ms.float32)
    return a.asnumpy().sum()
res = ray.get(test_ms.remote())
ray.shutdown()
print (res)

$ ray job status raysubmit_RUvkEqnkjNitKmnJ

$ ray job logs raysubmit_RUvkEqnkjNitKmnJ

$ ray job stop raysubmit_RUvkEqnkjNitKmnJ