Ray 框架：分散式計算與機器學習應用

– 15 分鐘閱讀

Ray 是一個高效能的分散式計算框架，專為處理大規模資料和機器學習任務而設計。它提供了一組簡潔的 API，讓開發者能輕鬆地將 Python 程式碼平行化，並在多個機器上執行。Ray 的核心概念包括任務和 actor，分別代表無狀態和有狀態的計算單元。透過 Ray Tune，開發者可以有效地執行超引數搜尋，而 Ray Serve 則簡化了模型部署和線上服務的流程。這些功能讓 Ray 成為一個功能強大且易於使用的分散式機器學習平臺。

超引數最佳化

超引數最佳化是機器學習中的一個重要步驟，涉及尋找最佳的模型引數以獲得最佳的效能。Ray 提供了一個名為 Ray Tune 的庫，支援超引數最佳化。

以下是一個使用 Ray Tune 進行超引數最佳化的例子：

import ray
from ray import tune

def training_function(config):
    # 定義模型和訓練過程
    x, y = config["x"], config["y"]
    # 模擬訓練過程
    time.sleep(10)
    # 計算目標函式
    score = objective(x, y)
    # 報告結果
    tune.report(score=score)

def objective(x, y):
    # 定義目標函式
    return math.sqrt((x**2 + y**2)/2)

# 定義超引數搜尋空間
config = {
    "x": tune.grid_search([-1, -0.5, 0, 0.5, 1]),
    "y": tune.grid_search([-1, -0.5, 0, 0.5, 1])
}

# 執行超引數最佳化
result = tune.run(
    training_function,
    config=config
)

# 獲取最佳超引陣列合
best_config = result.get_best_config(metric="score", mode="min")
print(best_config)

在這個例子中，我們定義了一個訓練函式 training_function，它接受一個超引數配置 config 作為輸入，並傳回一個目標函式值。然後，我們使用 Ray Tune 的 grid_search 方法定義了一個超引數搜尋空間，並執行超引數最佳化。最後，我們獲取最佳超引陣列合並打印出來。

模型服務

模型服務是指將訓練好的模型部署到線上，提供 API 供使用者訪問和使用。Ray 提供了一個名為 Ray Serve 的庫，支援模型服務。

以下是一個使用 Ray Serve 進行模型服務的例子：

import ray
from ray import serve
from transformers import pipeline

# 啟動 Ray Serve
serve.start()

# 載入和部署模型
@serve.deployment
def gpt2_model():
    # 載入 GPT-2 模型
    model = pipeline("text-generation", model="gpt2")
    # 定義模型預測函式
    def predict(text):
        # 執行模型預測
        output = model(text)
        # 傳回預測結果
        return output
    # 傳回模型預測函式
    return predict

# 啟動模型服務
gpt2_model.deploy()

# 測試模型服務
text = "What is the meaning of life?"
output = requests.post("http://localhost:8000/gpt2_model", json={"text": text})
print(output.json())

在這個例子中，我們首先啟動 Ray Serve，然後載入和部署一個 GPT-2 模型。我們定義了一個模型預測函式 predict，它接受一個輸入文字並傳回預測結果。然後，我們啟動模型服務並測試它使用一個示例輸入文字。

這些只是 Ray 提供的兩個例子，更多的功能和庫可以在 Ray 的文件中找到。

人工智慧框架 Ray 的概覽

Ray 是一個高效能的分散式計算框架，旨在提供一個統一的平臺 для資料科學和機器學習任務。它的設計目的是為了提供一個高效、可擴充套件和易用的平臺，讓開發者可以輕鬆地構建和部署分散式應用程式。

Ray 的核心概念

Ray 的核心概念是「任務」（Task）和「演算法」（Actor）。任務是指一個可以獨立執行的函式，演算法是指一個可以管理多個任務的實體。Ray 提供了一個簡單的 API，讓開發者可以輕鬆地定義和執行任務和演算法。

Ray 的高階庫

Ray 提供了一系列高階庫，包括 Ray Datasets、Ray RLlib 和 Ray Tune。這些庫提供了一個簡單的 API，讓開發者可以輕鬆地進行資料科學和機器學習任務。例如，Ray Datasets 提供了一個簡單的 API，讓開發者可以輕鬆地載入和處理大規模資料。

Ray 的生態系統

Ray 的生態系統包括了一系列第三方庫和工具。這些庫和工具提供了一個簡單的 API，讓開發者可以輕鬆地將 Ray 整合到現有的應用程式中。例如，Ray 可以與 Spark、Dask 和 Pandas 等庫整合，提供了一個簡單的 API，讓開發者可以輕鬆地進行資料科學和機器學習任務。

Ray 的優點

Ray 的優點包括：

  • 高效能：Ray 提供了一個高效能的分散式計算框架，讓開發者可以輕鬆地構建和部署分散式應用程式。
  • 可擴充套件：Ray 的設計目的是為了提供一個可擴充套件的平臺，讓開發者可以輕鬆地增加或減少計算資源。
  • 易用：Ray 提供了一個簡單的 API，讓開發者可以輕鬆地定義和執行任務和演算法。

上述內容簡要介紹了 Ray 的核心概念、 高階庫和生態系統。Ray 的核心概念包括任務和演算法，高階庫包括 Ray Datasets、Ray RLlib 和 Ray Tune。Ray 的生態系統包括了一系列第三方庫和工具，提供了一個簡單的 API，讓開發者可以輕鬆地將 Ray 整合到現有的應用程式中。

import ray

# 定義一個任務
@ray.remote
def my_task(x):
    return x * x

# 執行任務
result = ray.get(my_task.remote(4))
print(result)  # 輸出：16

圖表翻譯

以下是 Ray 的架構圖：

  graph LR
    A[Ray Core] --> B[Ray Datasets]
    A --> C[Ray RLlib]
    A --> D[Ray Tune]
    B --> E[資料載入和處理]
    C --> F[強化學習]
    D --> G[超引數最佳化]

這個圖表展示了 Ray 的核心架構，包括 Ray Core、Ray Datasets、Ray RLlib 和 Ray Tune。Ray Core 是 Ray 的核心，提供了一個簡單的 API，讓開發者可以輕鬆地定義和執行任務和演算法。Ray Datasets 提供了一個簡單的 API，讓開發者可以輕鬆地載入和處理大規模資料。Ray RLlib 提供了一個簡單的 API，讓開發者可以輕鬆地進行強化學習任務。Ray Tune 提供了一個簡單的 API，讓開發者可以輕鬆地進行超引數最佳化任務。

分散式計算框架 Ray 的概覽

Ray 是一個分散式計算框架，旨在提供一個通用且高效的方式來處理大規模的計算任務。它的核心是 Ray Core API，一個強大的分散式計算引擎，能夠讓開發者輕鬆地將計算任務分佈到多個機器上。

Ray 的架構可以分為三個層次：Ray Core API、Ray 高階庫和 Ray 生態系統。Ray Core API 提供了一個基礎的分散式計算介面，讓開發者可以輕鬆地將計算任務分佈到多個機器上。Ray 高階庫則提供了一系列的高階介面，讓開發者可以更容易地使用 Ray 的功能。Ray 生態系統則提供了一系列的第三方庫和工具，讓開發者可以擴充套件 Ray 的功能。

Ray Core API 的基礎

Ray Core API 是 Ray 的核心，它提供了一個基礎的分散式計算介面。它的主要功能包括：

  • 分散式任務（Task）：Ray Core API 提供了一個分散式任務介面，讓開發者可以輕鬆地將計算任務分佈到多個機器上。
  • 分散式演員（Actor）：Ray Core API 提供了一個分散式演員介面，讓開發者可以輕鬆地將計算任務分佈到多個機器上。
  • 物件儲存（Object Store）：Ray Core API 提供了一個物件儲存介面，讓開發者可以輕鬆地儲存和取回計算任務的結果。

Ray Core API 的使用

要使用 Ray Core API，開發者需要先啟動一個 Ray 叢集。Ray 叢集可以透過 ray.init() 函式啟動。啟動後，開發者可以使用 Ray Core API 提供的介面來分散式計算任務。

以下是一個簡單的例子，示範如何使用 Ray Core API 來分散式計算任務：

import ray

# 啟動 Ray 叢集
ray.init()

# 定義一個計算任務
@ray.remote
def calculate(data):
    # 進行計算
    result = data * 2
    return result

# 分散式計算任務
result_ids = [calculate.remote(i) for i in range(10)]

# 取回計算結果
results = ray.get(result_ids)

# 關閉 Ray 叢集
ray.shutdown()

這個例子示範如何使用 Ray Core API 來分散式計算任務。開發者可以定義一個計算任務，然後使用 @ray.remote 裝飾器來將其轉換為一個分散式任務。然後，開發者可以使用 ray.get() 函式來取回計算結果。

平行處理的重要性

在現代計算中，能夠高效地處理大量資料是一個非常重要的議題。傳統的序列處理方式往往會遇到效能瓶頸，尤其是在面對龐大資料集時。為瞭解決這個問題，程式設計師們開始探索平行處理的可能性。

什麼是平行處理？

平行處理是指在同一時間內，多個任務或程式可以同時執行的技術。這種技術可以大大提高計算效率，尤其是在多核心處理器或分散式計算系統中。

Ray Core 的介紹

Ray Core 是一個開源的平行處理框架，提供了一個簡單且高效的方式來實作平行計算。Ray Core 的設計目標是提供一個易於使用且高效的框架，讓開發者可以輕鬆地實作平行處理。

範例程式碼

以下是一個簡單的範例程式碼，示範如何使用 Ray Core 來實作平行處理：

import time
import ray

# 定義一個資料庫
database = {
    0: 'Learning',
    1: 'Ray',
    2: 'Flexible',
    3: 'Distributed',
    4: 'Python',
    5: 'for',
    6: 'Machine',
    7: 'Learning'
}

# 定義一個函式，模擬資料處理
@ray.remote
def retrieve(item):
    time.sleep(item / 10.)
    return item, database[item]

# 啟動 Ray Core
ray.init()

# 定義一個函式，印出執行時間
def print_runtime(input_data, start_time):
    print(f'Runtime: {time.time() - start_time:.2f} seconds, data:')
    print(*input_data, sep="\n")

# 執行平行處理
start = time.time()
data = ray.get([retrieve.remote(item) for item in range(8)])
print_runtime(data, start)

# 關閉 Ray Core
ray.shutdown()

執行結果

執行上述程式碼，將會輸出以下結果：

Runtime: 0.82 seconds, data:
(0, 'Learning')
(1, 'Ray')
(2, 'Flexible')
(3, 'Distributed')
(4, 'Python')
(5, 'for')
(6, 'Machine')
(7, 'Learning')

內容解密

上述程式碼使用了 Ray Core 來實作平行處理。首先，定義了一個資料庫和一個函式 retrieve，模擬資料處理。然後，使用 @ray.remote 來標記 retrieve 函式，表示它可以被平行執行。接下來，啟動 Ray Core，定義一個函式 print_runtime，印出執行時間。最後，執行平行處理，使用 ray.get 來取得結果，然後印出執行時間和結果。

圖表翻譯

  flowchart TD
    A[啟動 Ray Core] --> B[定義資料庫]
    B --> C[定義 retrieve 函式]
    C --> D[標記 retrieve 函式為平行執行]
    D --> E[啟動平行處理]
    E --> F[取得結果]
    F --> G[印出執行時間和結果]
    G --> H[關閉 Ray Core]

上述圖表展示了程式碼的執行流程。首先，啟動 Ray Core，然後定義資料庫和 retrieve 函式。接下來，標記 retrieve 函式為平行執行，然後啟動平行處理。最後，取得結果，印出執行時間和結果，然後關閉 Ray Core。

平行計算與 Ray Core

平行計算是指在多個 CPU 核心或甚至多臺機器上同時執行多個任務，以達到更快的計算速度。Python 的 Global Interpreter Lock (GIL) 限制了平行計算的能力，但有方法可以繞過這個限制。

Python 的 GIL

GIL 是 Python 的一項機制，確保在多執行緒環境中，只有一個執行緒可以執行 Python 程式碼。這使得記憶體管理更容易，但也限制了平行計算的能力。GIL 是 Python 的一項特性，對於 I/O 繫結的程式來說，GIL 可能不是一個問題，但對於 CPU 繫結的程式，GIL 可能會導致效能下降。

Ray Core

Ray Core 是一個開源的平行計算框架，允許使用者將 Python 程式碼分佈在多個機器上執行。Ray Core 提供了一個簡單的 API，允許使用者定義任務並將其分佈在叢集上執行。

定義 Ray 任務

要定義一個 Ray 任務，需要使用 @ray.remote 裝飾器。這個裝飾器將一個 Python 函式轉換為一個 Ray 任務，允許它在叢集上執行。例如：

@ray.remote
def retrieve_task(item):
    return retrieve(item)

這個任務可以使用 .remote() 方法呼叫，傳入引數 item。Ray 會將這個任務分佈在叢集上執行，並傳回一個物件參考。

執行 Ray 任務

要執行 Ray 任務，需要使用 ray.get() 方法，傳入物件參考。這個方法會傳回任務的結果。例如：

object_references = [retrieve_task.remote(item) for item in range(8)]
data = ray.get(object_references)

這個範例展示瞭如何定義一個 Ray 任務，執行它，並傳回結果。

使用物件儲存

Ray 提供了一個物件儲存機制，允許使用者在叢集上共享物件。物件儲存可以使用 ray.put() 方法儲存物件，然後使用 ray.get() 方法取回物件。例如：

db_object_ref = ray.put(database)
@ray.remote
def retrieve_task(item, db):
    time.sleep(item / 10.)
    return retrieve(item, db)

這個範例展示瞭如何使用物件儲存機制共享物件。

非阻塞式呼叫的使用

在 Ray 中，使用 ray.get() 來存取結果是一種阻塞式呼叫，意味著驅動程式必須等待所有結果都可用後才會繼續執行。然而，在某些情況下，我們可能需要讓驅動程式在等待結果的同時也能執行其他任務，或者我們希望在結果出現時就立即處理它們。

為瞭解決這個問題，Ray 提供了一種非阻塞式呼叫的方法，稱為 ray.wait()。這個方法允許我們指定一個 timeout 時間，如果在這個時間內有結果出現，就會傳回這些結果，否則就會傳回 None。

以下是使用 ray.wait() 的範例：

import time
import ray

# 定義一個遠端任務
@ray.remote
def retrieve_task(item, db_object_ref):
    # 模擬資料庫查詢
    time.sleep(1)
    return item, db_object_ref

# 啟動 Ray
ray.init()

# 建立一個物件儲存
db_object_ref = ray.put({"key": "value"})

# 啟動遠端任務
object_references = [
    retrieve_task.remote(item, db_object_ref) for item in range(8)
]

# 初始化變數
all_data = []
start = time.time()

# 非阻塞式呼叫
while len(object_references) > 0:
    # 等待結果出現
    finished, object_references = ray.wait(
        object_references, num_returns=2, timeout=7.0
    )

    # 取得結果
    data = ray.get(finished)

    # 處理結果
    print_runtime(data, start)

    # 將結果新增到 all_data 中
    all_data.extend(data)

# 關閉 Ray
ray.shutdown()

在這個範例中，我們使用 ray.wait() 來等待結果出現，並指定了 num_returns=2，這意味著當有兩個結果出現時，就會傳回這兩個結果。同時，我們也指定了 timeout=7.0，這意味著如果在 7 秒內沒有結果出現，就會傳回 None。

當結果出現時，我們會使用 ray.get() 來取得結果，並將其新增到 all_data 中。最後，我們會關閉 Ray。

這種非阻塞式呼叫的方法可以讓我們在等待結果的同時也能執行其他任務，或者我們可以在結果出現時就立即處理它們。這是一種非常有用的方法，尤其是在需要處理大量資料的場合。

依賴任務的處理

到目前為止，我們的範例程式相對簡單，僅包含一個步驟：從資料庫中取回一批資料。現在，假設我們想要在資料載入後執行一個後續的處理任務。為了具體化，假設我們想要使用第一個取回任務的結果來查詢其他相關的資料（假設我們正在查詢相同資料庫中的不同表格的資料）。範例 2-2 設定了這樣一個任務，並連續執行了我們的 retrieve_task 和 follow_up_task。

import ray

# 定義一個遠端任務
@ray.remote
def follow_up_task(retrieve_result):
    # 從 retrieve_result 中解析原始專案
    original_item, _ = retrieve_result
    
    # 使用原始專案查詢更多資料
    follow_up_result = retrieve(original_item + 1)
    
    # 傳回原始結果和後續結果
    return retrieve_result, follow_up_result

# 執行 retrieve_task 並取得物件參考
retrieve_refs = [retrieve_task.remote(item, db_object_ref) for item in [0, 2, 4, 6]]

# 使用 retrieve_refs 中的物件參考執行 follow_up_task
follow_up_refs = [follow_up_task.remote(ref) for ref in retrieve_refs]

# 取回 follow_up_refs 中的結果
result = [print(data) for data in ray.get(follow_up_refs)]

這段程式碼使用 follow_up_task 來處理 retrieve_task 的結果，展示瞭如何在 Ray 中處理任務之間的依賴關係。Ray 會自動建立一個依賴圖，確保任務按照正確的順序執行，不需要手動指定等待前一個任務完成的時機。

Ray 的依賴圖和物件儲存

Ray 的依賴圖是其強大的功能之一，它允許使用者定義任務之間的依賴關係，而不需要手動管理這些依賴關係。當一個任務需要另一個任務的結果時，Ray 會自動等待前一個任務完成，並在必要時使用 ray.get 來解析物件參考。

此外，Ray 的物件儲存允許使用者在不同任務之間共享大型中間值，而不需要將這些值複製回驅動程式。這對於需要處理大量資料的應用程式尤其有用，因為它可以避免不必要的資料複製和傳輸。

關於 Ray Core 的進一步探索

在前面的例子中，我們已經瞭解瞭如何使用 Ray 來實作遠端函式和任務的執行。現在，讓我們進一步探索 Ray Core 的另一個重要概念：演員（Actors）。

演員的概念

演員是 Ray 中的一種特殊物件，允許你在叢集中執行有狀態的計算。它們可以相互通訊，並且可以用來追蹤某些狀態的變化。演員是透過使用 @ray.remote 裝飾器來定義的，就像我們之前使用它來定義遠端函式一樣。

實作一個簡單的計數器

讓我們實作一個簡單的計數器來追蹤我們的資料庫呼叫次數。為此，我們可以定義一個 DataTracker 類別，並使用 @ray.remote 裝飾器來使它成為一個演員：

@ray.remote
class DataTracker:
    def __init__(self):
        self._counts = 0

    def increment(self):
        self._counts += 1

    def counts(self):
        return self._counts

這個 DataTracker 類別已經是一個演員，因為我們使用了 @ray.remote 裝飾器。它可以追蹤狀態，即一個簡單的計數器，並且它的方法是 Ray 任務，可以使用 .remote() 方法來呼叫。

修改現有的任務

現在，讓我們修改現有的 retrieve_task 來使用這個新的演員。為此，我們需要將演員的例項作為一個引數傳遞給任務：

@ray.remote
def retrieve_tracker_task(item, tracker, db):
    time.sleep(item / 10.)
    tracker.increment.remote()
    return item, db[item]

tracker = DataTracker.remote()

object_references = [
    retrieve_tracker_task.remote(item, tracker, db_object_ref)
    for item in range(8)
]

data = ray.get(object_references)

print(data)

在這個例子中，我們建立了一個 DataTracker 演員的例項，並將它傳遞給 retrieve_tracker_task 任務。任務會在每次呼叫時遞增計數器。

分散式計算基礎：Ray Core

在分散式計算的世界中，Ray Core 是一個強大的工具，允許使用者輕鬆地在多個機器上執行任務和管理狀態。要開始使用 Ray Core，首先需要了解其基本概念和 API。

Ray API 概覽

Ray API 提供了一組簡單易用的方法，讓使用者可以輕鬆地在分散式環境中執行任務和管理狀態。以下是 Ray API 的六個主要方法：

  • ray.init(): 初始化 Ray Cluster。
  • @ray.remote: 將函式轉換為任務，將類別轉換為 actor。
  • ray.put(): 將值放入 Ray 的物件儲存中。
  • ray.get(): 從 Ray 的物件儲存中取出值。
  • .remote(): 在 Ray Cluster 上執行 actor 方法或任務。
  • ray.wait(): 等待任務完成並傳回結果。

Ray 系統元件

Ray 系統由多個元件組成，包括：

  • Raylet: 每個工作節點上的智慧元件，負責管理工作程序和物件儲存。
  • 物件儲存: 每個節點上的共享記憶體池，負責管理物件的儲存和取用。
  • 任務排程器: 負責管理資源和依賴關係，確保任務可以正確地在工作節點上執行。

任務排程和執行

任務排程是 Ray 中的一個關鍵元件，負責將任務分配給工作節點上的工作程序。任務排程器需要考慮多個因素，包括：

  • 資源管理: 確保工作程序有足夠的資源（如 CPU、GPU、記憶體）來執行任務。
  • 依賴關係: 確保工作程序有所有需要的物件和依賴關係，以執行任務。
圖表翻譯

此圖表示 Ray 的系統元件和任務執行流程。首先，ray.init() 初始化 Ray Cluster，然後 Raylet 管理工作程序和物件儲存。任務排程器負責管理資源和依賴關係，確保任務可以正確地在工作節點上執行。最後，工作程序執行任務並傳回結果。

import ray

# 初始化 Ray Cluster
ray.init()

# 定義一個任務
@ray.remote
def my_task(x):
    return x * x

# 執行任務
result = my_task.remote(4)

# 等待任務完成
ray.wait([result])

# 取出結果
result = ray.get(result)
print(result)  # 輸出：16

內容解密

此程式碼示範瞭如何使用 Ray API 定義和執行一個任務。首先，初始化 Ray Cluster，然後定義一個任務 my_task，此任務接收一個引數 x 並傳回其平方。然後，使用 @ray.remote 將此任務轉換為一個遠端任務，然後使用 .remote() 執行此任務。最後，使用 ray.wait() 等待任務完成，然後使用 ray.get() 取出結果。

分散式計算中的擁有權和依賴關係

在分散式計算中，瞭解擁有權和依賴關係對於設計和實作高效的系統至關重要。讓我們透過一個具體的例子來闡述這兩個概念。在這個例子中，我們有一個程式，它啟動了一個簡單的任務，並在內部呼叫另一個任務。

@ray.remote
def task_owned():
    return

@ray.remote
def task(dependency):
    res_owned = task_owned.remote()
    return

val = ray.put("value")
res = task.remote(dependency=val)

在這個例子中，我們定義了兩個任務：task和task_owned。我們還有三個變數：val、res和res_owned。主程式定義了val和res，並呼叫了task。根據Ray的定義，主程式擁有task、val和res。相反，res依賴於task，但之間沒有擁有權關係。

當task被呼叫時，它以val作為依賴項。然後，它呼叫task_owned並分配res_owned，因此它擁有它們兩者。最後，task_owned本身不擁有任何東西，但res_owned確實依賴於它。

Ray的系統元件

Ray是一個分散式計算框架，它由多個系統元件組成。這些元件包括：

  • Head Node：每個Ray叢集都有一個特殊的節點，稱為Head Node。Head Node上執行著一個驅動程式（driver process），該程式負責提交任務和管理叢集。
  • Worker Node：Worker Node是叢集中的其他節點，它們執行著工作者程式（worker process），該程式負責執行任務。
  • Global Control Service (GCS)：GCS是一個儲存全域性資訊的後設資料儲存。它儲存著系統級別的後設資料，例如每個Raylet的心跳訊號，以確保它們仍然可達。
  • Raylet：Raylet是每個節點上執行的Ray代理程式。它負責管理節點上的記憶體和資源。

綜觀分散式計算框架的發展趨勢，Ray 以其簡潔的 API 和高效的效能，正逐漸成為機器學習和資料科學領域的熱門選擇。深入剖析 Ray Core 的核心架構，可以發現其任務、actor 和物件儲存機制巧妙地解決了分散式計算中的複雜性問題。然而，Ray 也並非完美無缺，目前仍存在一些挑戰，例如在處理極其複雜的依賴關係時，排程效率仍有提升空間，同時也需要更多針對不同硬體環境的最佳化策略。對於追求極致效能的應用場景，深入理解 Ray 的底層機制和最佳實踐至關重要。展望未來，隨著 Ray 生態系統的持續發展和社群的積極貢獻，我們預見 Ray 將在更多領域扮演關鍵角色，並推動分散式計算技術的進一步革新。對於有意採用分散式計算方案的團隊，建議深入評估 Ray 的適用性，並著重關注其與現有系統的整合以及相關技術人才的培養。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。