一個嵌套并行的簡單例子。一個應用運行兩個并行實驗（每個都是長時間運行任務），每個實驗運行一定數量的并行模擬（每一個同時也是一個任務）。
Ray 有兩種主要使用方法：通過低級 API 或高級庫。高級庫是構建在低級 API 之上的。目前它們包括 Ray RLlib，一個可擴展強化學習庫；和 Ray.tune，一個高效分布式超參數搜索庫。
開發 Ray API 的目的是讓我們能更自然地表達非常普遍的計算模式和應用，而不被限制為固定的模式，就像 MapReduce 那樣。
Ray 應用的基礎是動態任務圖。這和 TensorFlow 中的計算圖很不一樣。TensorFlow 的計算圖用于表征神經網絡，在單個應用中執行很多次，而 Ray 的任務圖用于表征整個應用，并僅執行一次。任務圖對于前臺是未知的，隨著應用的運行而動態地構建，且一個任務的執行可能創建更多的任務。
計算圖示例。白色橢圓表示任務，藍色方框表示對象。箭頭表示任務依賴于一個對象，或者任務創建了一個對象。
任意的 Python 函數都可以當成任務來執行，并且可以任意地依賴于其它任務的輸出。如下示例代碼所示：
# Define two remote functions. Invocations of these functions create tasks
# that are executed remotely.
@ray
.
remote
def
multiply
(
x
,
y
):
return
np
.
dot
(
x
,
y
)
@ray
.
remote
def
zeros
(
size
):
return
np
.
zeros
(
size
)
# Start two tasks in parallel. These immediately return futures and the
# tasks are executed in the background.
x_id
=
zeros
.
remote
((
100
,
100
))
y_id
=
zeros
.
remote
((
100
,
100
))
# Start a third task. This will not be scheduled until the first two
# tasks have completed.
z_id
=
multiply
.
remote
(
x_id
,
y_id
)
# Get the result. This will block until the third task completes.
z
=
ray
.
get
(
z_id
)僅用遠程函數和上述的任務所無法完成的一件事是在相同的共享可變狀態上執行多個任務。這在很多機器學習場景中都出現過，其中共享狀態可能是模擬器的狀態、神經網絡的權重或其它。Ray 使用 actor 抽象以封裝多個任務之間共享的可變狀態。以下是關于 Atari 模擬器的虛構示例：
import
gym
@ray
.
remote
class
Simulator
(
object
):
def
__init__
(
self
):
self
.
env
=
gym
.
make
(
"Pong-v0"
)
self
.
env
.
reset
()
def
step
(
self
,
action
):
return
self
.
env
.
step
(
action
)
# Create a simulator, this will start a remote process that will run
# all methods for this actor.
simulator
=
Simulator
.
remote
()
observations
=
[]
for
_
in
range
(
4
):
# Take action 0 in the simulator. This call does not block and
# it returns a future.
observations
.
append
(
simulator
.
step
.
remote
(
0
))Actor 可以很靈活地應用。例如，actor 可以封裝模擬器或神經網絡策略，并且可以用于分布式訓練（作為參數服務器），或者在實際應用中提供策略。
圖左：actor 為客戶端進程提供預測/操作。圖右：多個參數服務器 actor 使用多個工作進程執行分布式訓練。
參數服務器示例
一個參數服務器可以作為一個 Ray actor 按如下代碼實現：
@ray
.
remote
class
ParameterServer
(
object
):
def
__init__
(
self
,
keys
,
values
):
# These values will be mutated, so we must create a local copy.
values
=
[
value
.
copy
()
for
value
in
values
]
self
.
parameters
=
dict
(
zip
(
keys
,
values
))
def
get
(
self
,
keys
):
return
[
self
.
parameters
[
key
]
for
key
in
keys
]
def
update
(
self
,
keys
,
values
):
# This update function adds to the existing values, but the update
# function can be defined arbitrarily.
for
key
,
value
in
zip
(
keys
,
values
):
self
.
parameters
[
key
]
+=
value這里有更完整的示例：http://ray.readthedocs.io/en/latest/example-parameter-server.html
執行以下代碼初始化參數服務器：
parameter_server
=
ParameterServer
.
remote
(
initial_keys
,
initial_values
)執行以下代碼，創建 4 個長時間運行的持續恢復和更新參數的工作進程：
@ray
.
remote
def
worker_task
(
parameter_server
):
while
True
:
keys
=
[
'key1'
,
'key2'
,
'key3'
]
# Get the latest parameters.
values
=
ray
.
get
(
parameter_server
.
get
.
remote
(
keys
))
# Compute some parameter updates.
updates
=
…
# Update the parameters.
parameter_server
.
update
.
remote
(
keys
,
updates
)
# Start 4 long-running tasks.
for
_
in
range
(
4
):
worker_task
.
remote
(
parameter_server
)Ray 高級庫
Ray RLib 是一個可擴展的強化學習庫，其建立的目的是在多個機器上運行，可以通過示例訓練腳本或者 Python API 進行使用。目前它已有的實現為：
UC Berkeley 的開發者在未來將繼續添加更多的算法。同時，RLib 和 OpenAI gym 是完全兼容的。
Ray.tune 是一個高效的分布式超參數搜索庫。它提供了一個 Python API 以執行深度學習、強化學習和其它計算密集型任務。以下是一個虛構示例的代碼：
from
ray
.
tune
import
register_trainable
,
grid_search
,
run_experiments
# The function to optimize. The hyperparameters are in the config
# argument.
def
my_func
(
config
,
reporter
):
import
time
,
numpy
as
np
i
=
0
while
True
:
reporter
(
timesteps_total
=
i
,
mean_accuracy
=(
i
**
config
[
'alpha'
]))
i
+=
config
[
'beta'
]
time
.
sleep
(
0.01
)
register_trainable
(
'my_func'
,
my_func
)
run_experiments
({
'my_experiment'
:
{
'run'
:
'my_func'
,
'resources'
:
{
'cpu'
:
1
,
'gpu'
:
0
},
'stop'
:
{
'mean_accuracy'
:
100
},
'config'
:
{
'alpha'
:
grid_search
([
0.2
,
0.4
,
0.6
]),
'beta'
:
grid_search
([
1
,
2
]),
},
}
})可以使用 TensorBoard 和 rllab’s VisKit 等工具對運行狀態的結果進行實時可視化（或者直接閱讀 JSON 日志）。Ray.tune 支持網格搜索、隨機搜索和更復雜的早停算法，如 HyperBand。
本文章轉載微信公眾號@機器之心