Ray 分散式運算：物件不可變性與序列化技術探討

Ray 框架的核心概念之一是物件的不可變性，這表示一旦 Ray 物件建立後，其內容便無法修改。開發者在操作 Ray 物件時，需注意對物件的任何修改操作僅作用於本地副本，而不會影響儲存在 Ray 物件儲存中的原始物件。文章以程式碼範例展示了此特性，並說明如何使用 ray.put 和 ray.get 來操作 Ray 物件。此外，文章也探討了 Ray 物件的儲存策略，建議在處理大型或頻繁使用的資料時，明確地將其儲存為 Ray 物件，以提高效能。Ray 的物件儲存機制採用參考計數垃圾回收策略，並支援將物件溢位到磁碟以應對記憶體壓力。開發者可以透過系統設定來調整物件儲存的行為，例如設定最小溢位大小和物件儲存的記憶體容量。

在分散式環境中，序列化和反序列化是不可或缺的技術。Ray 利用 Cloudpickle 擴充套件了 Python 內建的 pickle 模組，使其能夠處理更多種類別的物件，例如包含非序列化元件的類別。文章以程式碼範例示範瞭如何使用自定義序列化器來處理無法直接序列化的物件，例如包含執行緒池的類別。同時，文章也介紹瞭如何註冊自定義序列化器，以簡化程式碼並提高可讀性。Ray 使用 Apache Arrow 作為高效的資料格式，並利用 gRPC 作為高效能的通訊框架。Arrow 的強型別和跨語言特性使其成為 Ray 處理資料集的理想選擇，而 gRPC 則為 Ray 的內部通訊提供了穩固的基礎。

文章最後討論了 Ray 的資源管理和垂直擴充套件策略。Ray 允許開發者透過 @ray.remote 裝飾器來指定函式或 Actor 的資源需求，例如 CPU 和 GPU 的數量。Ray 的資源請求通常是軟請求，這意味著 Ray 會盡力滿足需求，但不保證一定能夠滿足。理解 Ray 的資源管理機制對於構建高效能的分散式應用程式至關重要。

Ray 物件的不可變性及其管理

在 Ray 系統中，Ray 物件是不可變的，這意味著一旦建立，就無法修改。這一特性對於理解和使用 Ray 非常重要。如果你從 Ray 中讀取了一個物件（例如使用 ray.get）或將其儲存在 Ray 中（例如使用 ray.put），那麼對這個物件所做的任何修改都不會反映到物件儲存中。以下是一個示範如何操作 Ray 物件的例子：

示範範例

remote_array = ray.put([1])
v = ray.get(remote_array)
v.append(2)
print(v)  # 輸出 [1, 2]
print(ray.get(remote_array))  # 輸出 [1]

在這段程式碼中，我們可以看到雖然可以修改一個值，但這些修改不會傳播到物件儲存中。

Ray 物件的儲存策略

當引數或傳回值比較大且被多次使用，或者中等大小但頻繁使用時，明確地將其儲存為物件是值得考慮的。這樣可以使用 ObjectRef 代替普通引數，Ray 會自動將 ObjectRef 轉換為 Python 型別。以下是一個示範如何使用 ray.put 的例子：

import numpy as np

@ray.remote
def sup(x):
    import random
    import sys
    return len(x)

p = ray.put(np.array(range(0, 1000)))
ray.get([sup.remote(p), sup.remote(p), sup.remote(p)])

內容解密：

import numpy as np

@ray.remote
def sup(x):
    import random
    import sys
    return len(x)

上述程式碼定義了一個遠端函式 sup，該函式接受一個引數 x，並傳回其長度。這個函式用於演示如何在 Ray 中處理大型資料。

p = ray.put(np.array(range(0, 1000)))

這行程式碼將一個包含 1000 個元素的 NumPy 陣列儲存在 Ray 的物件儲存中，並傳回一個 ObjectRef。

ray.get([sup.remote(p), sup.remote(p), sup.remote(p)])

這行程式碼將 sup.remote(p) 被呼叫三次，並使用 ray.get 來取得結果。這樣可以展示如何在多次呼叫中重複使用相同的物件。

Ray 物件的複製與遺失處理

當另一個節點需要一個物件時，它會向擁有該物件的節點請求並建立一個本地副本。因此，相同的物件可以存在於不同節點的物件儲存中。Ray 預設不主動複製物件，因此可能只有一份物件存在。

如果你嘗試取得一個遺失的物件，Ray 預設會引發 ObjectLostError。你可以透過設定 enable_object_reconstruction=True 或在啟動 Ray 時新增 --enable-object-reconstruction 來啟用重新計算。這種重新計算會在需要物件時發生（解析時延遲）。

物件可以以兩種方式遺失：如果擁有該物件的節點丟失，則該物件也會丟失；如果所有儲存該物件的節點都失敗了，則 Ray 也會遺失該物件。

Ray 的垃圾回收機制

Ray 的物件儲存使用參考計數垃圾回收來清理不再需要的物件。這包括追蹤直接和間接參考。

即使有垃圾回收機制，物件儲存仍可能被填滿。當物件儲存填滿時，Ray 會首先執行垃圾回收，移除沒有參考的物件。如果記憶體壓力仍然存在，物件儲存會嘗試將物件刪除到磁碟上。

調整物件儲存設定

你可能需要根據具體使用案例調整物件儲存設定。例如，你可能需要為物件儲存分配更多或更少的記憶體。你可以透過 _system_config 設定來組態物件儲存。兩個重要的組態選項包括最小刪除到磁碟的聚合大小 min_spilling_size 和分配給物件儲存的總記憶體 object_store_memory_mb。

以下是如何組態 Ray 的物件儲存設定的示範：

import json

ray.init(num_cpus=20,
         _system_config={
             "min_spilling_size": 1024 * 1024, # 最少刪除到磁碟大小為 1 MB
             "object_store_memory_mb": 500,
             "object_spilling_config": json.dumps(
                 {"type": "filesystem", "params": {"directory_path": "/tmp/fast"}},
             )
         })

內容解密：

import json

此行程式碼匯入了 JSON 模組，用於處理 JSON 資料格式。

ray.init(num_cpus=20,
         _system_config={
             "min_spilling_size": 1024 * 1024, # 最少刪除到磁碟大小為 1 MB
             "object_store_memory_mb": 500,
             "object_spilling_config": json.dumps(
                 {"type": "filesystem", "params": {"directory_path": "/tmp/fast"}},
             )
         })

此段程式碼初始化了 Ray 的系統組態設定。其中包括設定最小刪除到磁碟的聚合大小、分配給物件儲存的總記憶體以及指定刪除到磁碟的目錄路徑。

摘要

Ray 物件是不可變的，這意味著它們一旦建立就無法修改。Ray 提供了多種策略來管理和儲存這些物件，包括參考計數垃圾回收和刪除到磁碟。透過適當地組態系統設定，你可以根據具體需求調整 Ray 的行為。

問題與挑戰

由於 Ray 的分散式特性和高效運算需求，Ray 需要處理許多潛在問題和挑戰：

• 序列化：Ray 須依賴序列化技術來傳輸資料和函式。
• 容錯性：序列化/反序列化必須可靠且高效。
• 分散式管理：多節點下資源管理和組態調整更為複雜。
• 效能最佳化：如何平衡記憶體使用和 I/O 效能？當前實作上依然有許多可進步之處。
• 安全性：跨網路傳輸及資料分散管理帶來額外安全隱憂及挑戰。

玄貓建議開發者深入瞭解每項技術細節及其應用場景，以便根據實際需求進行最佳選擇及系統設計調整。

快速序列化與反序列化技術概述

Ray 作為高效分散式運算框架之一，須依賴快速序列化/反序列化技術來傳輸資料和函式。以下是常見序列化技術概述：

Python Pickle 和 Cloudpickle

Pickle 是 Python 自帶的一種序列化/反序列化模組（即 “pickling” 和 “unpickling”），能夠處理 Python 的大部分資料結構與自訂類別。然而 Pickle 不支援跨語言支援及無法應用於複雜狀態保留（如網路連線）。

Cloudpickle 則是根據 Pickle 提升而成之延伸函式庫（fork），適用於分散式計算環境下強需求序列化方法。它能夠序列化更多型別資料（特別是無法透過 Pickle 序列化之類別）。

Arrow 與 Parquet

Apache Arrow 是一種高效且跨語言資料格式與運算框架（in-memory data format），專注於資料運算最佳化並提供高速轉換與 I/O 效能最佳化能力。

Parquet 則是適用於大型資料集之資料格式（columnar storage file format），具有高壓縮率及快速查詢能力特色。透過 Arrow 和 Parquet 預先處理資料後才進行傳輸能顯著提升效能並降低資源消耗量。

Fast Serialization 和 MessagePack

Fast Serialization 是專為快速序列化而設計之 Java 函式庫（library）。它擁有極高效能之特性並且適用於大型 Java 應用場景下之需求。

MessagePack 則是另一種輕量級且高效能之二進位制編碼協定（binary encoding scheme）。它兼具 JSON 的易讀性及 Protobuf 高壓縮率優點，常用於 Web 應用、IoT 等高需求低延遲情境之下之應用場景。

問題與挑戰

選擇合適之序列化技術取決於具體應用需求和執行環境：

• 效能：不同技術間在處理速度、記憶體消耗方面皆有差異。
• 相容性：不同語言間如何互通？
• 安全性：如何確保跨網路傳輸安全？
• 維護成本：新技術引入後維護成本增加？

玄貓建議開發者根據具體情境進行權衡考量並選擇最適合之序列化技術來達到最高效能及最佳穩定性平衡。

  graph TD;
    A[資料產生] --> B[轉換至 Arrow]
    B --> C[壓縮至 Parquet]
    C --> D[透過 Fast Serialization / MessagePack 傳輸]
    D --> E[接收端解壓縮]

內容解密：

此圖示展示了從資料產生到最終傳輸接收端的一系列轉換與壓縮過程：

  graph TD;

圖示初始部分定義了此圖示為流程圖型態（graph type definition）。

  A[資料產生] --> B[轉換至 Arrow]

此行表示從 A 資料產生節點指向 B 轉換至 Arrow 節點（即初步轉換處理）。

  B --> C[壓縮至 Parquet]

此行表示從 B 轉換至 Arrow 節點指向 C 壓縮至 Parquet 節點（即進行壓縮以提升傳輸效率）。

  C --> D[透過 Fast Serialization / MessagePack 傳輸]

此行表示從 C 壓縮至 Parquet 節點指向 D 傳輸節點（即進行快速傳輸）。

  D --> E[接收端解壓縮]

此行表示從 D 傳輸節點指向 E 接收端解壓縮節點（即最終接收端完成解壓縮並提取原始資料）。

最後分享一些實務建議：
1. **模組選擇**：根據實際需求選擇最適合之序列化/反序列模組。
2. **測試評估**：進行全面測試評估確保其穩定性與效能達標。
3. **持續最佳化**：根據實際執行狀況持續最佳化相關設定以達最佳平衡。
4. **安全措施**：強烈建議針對跨網路傳輸加強安全措施以降低潛在風險！

## 使用Ray進行序列化：探討

在現代分散式計算中，資料序列化與反序列化是至關重要的技術。Ray是一個強大的分散式計算框架，其設計旨在高效處理這些任務。Ray的序列化機制比Python內建的pickle更為強大，能夠處理更多的函式和型別。當你遇到一些資料無法序列化時，可以使用Ray提供的工具來檢查和解決問題。

### 雲端pickle（CloudPickle）與自定義序列化

雲端pickle（CloudPickle）是Ray中的一個強大工具，它能夠序列化和反序列化更多的函式和型別。當你需要序列化一些複雜的物件，例如包含非序列化元件（如資料函式庫連線）時，可以使用自定義序列化器。

#### 自定義序列化器

以下是一個示例，展示瞭如何為包含非序列化元件的類別建立自定義序列化器。這個示例中，我們將序列化一個包含執行緒池的類別。

```python
import ray.cloudpickle as pickle
from multiprocessing import Pool

class BadClass:
    def __init__(self, threadCount, friends):
        self.friends = friends
        self.p = Pool(threadCount)  # 不可序列化

i = BadClass(5, ["boo", "boris"])
# 這將會失敗，因為"pool objects cannot be passed between processes or pickled"
# pickle.dumps(i)

class LessBadClass:
    def __init__(self, threadCount, friends):
        self.friends = friends
        self.p = Pool(threadCount)

    def __getstate__(self):
        state_dict = self.__dict__.copy()
        # 我們無法移動執行緒，但可以移動用於建立相同大小執行緒池的資訊
        state_dict["p"] = len(self.p._pool)
        return state_dict

    def __setsate__(self):
        self.__dict__.update(state)
        self.p = Pool(self.p)

k = LessBadClass(5, ["boo", "boris"])
pickle.loads(pickle.dumps(k))

內容解密：

在這段程式碼中，我們首先嘗試直接對包含執行緒池的BadClass進行序列化。然而，這會導致錯誤，因為執行緒池物件不能被傳遞或序列化。

於是我們引入了LessBadClass類別，這個類別重寫了__getstate__和__setstate__方法來自定義序列化行為。在__getstate__方法中，我們將執行緒池物件替換為其大小資訊，這樣就可以安全地進行序列化和反序列化。

註冊自定義序列器

除了在類別中重寫方法外，Ray還允許我們為外部類別註冊自定義序列器。這樣可以避免繼承和擴充套件外部函式庫中的類別，使程式碼更加簡潔易讀。

def custom_serializer(bad):
    return {"threads": len(bad.p._pool), "friends": bad.friends}

def custom_deserializer(params):
    return BadClass(params["threads"], params["friends"])

# 註冊自定義序列器和反序列器
ray.util.register_serializer(
    BadClass, serializer=custom_serializer, deserializer=custom_deserializer)
ray.get(ray.put(i))

內容解密：

在這段程式碼中，我們為BadClass註冊了一個自定義的序列器和反序列器。這樣做的好處是不需要繼承和擴充套件外部函式庫中的類別，使程式碼更加簡潔易讀。

Apache Arrow：高效的資料格式

Ray利用Apache Arrow來進行資料集的序列化。Arrow是一種強型別、導向行的格式，具有高效的空間利用率。它不僅能夠在不同版本的Python之間使用，還能在不同程式語言之間使用，例如Rust、C、Java、Python及CUDA。

然而，並不是所有使用Arrow的工具都支援相同的資料型別。例如，Arrow支援巢狀欄位，而pandas則不支援。

gRPC：高效能通訊框架

gRPC是一個現代、開源、高效能的遠端過程呼叫框架，它構成了Ray內部通訊的基礎。gRPC使用Protocol Buffers進行快速小物件的序列化，而較大物件則使用Arrow或cloudpickle進行序列化並儲存在Ray物件儲存中。

資源管理與垂直擴充套件

Ray預設假設所有函式和演員具有相同的資源需求（如一個CPU）。但如果某些演員或函式有不同的資源需求時，可以明確指定所需資源。排程器將嘗試找到具有可用資源的節點來分配任務。

@ray.remote(num_cpus=4, num_gpus=2)
def my_function():
    # 函式內容

內容解密：

在這段程式碼中，我們使用了@ray.remote裝飾器來指定函式所需的資源。例如，num_cpus=4, num_gpus=2表示該函式需要四個CPU和兩個GPU來執行。

資源請求與排程

大多數資源請求在Ray中都是軟請求，意味著Ray不保證資源限制，但會盡力滿足需求。當沒有指定CPU需求時，遠端函式將要求一個CPU進行分配和執行；而演員則需要一個CPU進行排程但不需要執行時CPU。

總結來說，Ray透過其強大的序列化機制和靈活的資源管理能力，為分散式計算提供了高效且靈活的解決方案。無論是自定義序列器還是資源管理策略，都展示了Ray在處理複雜任務時的卓越表現。