Ray Summit是Ray社區一年一度的全球盛會，2024年于9月30日至10月2日在美國舊金山舉行，主題是"Where Builders Create the AI Future"，聚焦于構建人工智能的未來，吸引了全球眾多AI開發者和行業領袖。今年的Ray Summit不僅是一個技術交流的平臺，更是一個展示最新AI技術和趨勢的舞臺。包括OpenAI、Meta、LangChain、Google、Nvidia、ByteDance在內的多家頂尖科技公司和組織參與了此次盛會，共同探討和分享了他們在AI領域的最新進展和洞見。

來自ByteDance的Xiaohong Dong、Zhibei Ma、Liguang Xie分享了題為《How Bytedance Builds Large-Scale Data Processing Pipelines for Multimodal Models with Ray》的演講。

YouTube視頻鏈接：https://www.youtube.com/watch?v=f67SKoxR9H0&t=152s

我們來自字節跳動Seed（語音、視覺）和Data Infra團隊，致力于構建高性能、可擴展的分布式數據處理平臺，通過數據驅動的方法來提高多模態大模型能力。

當前數據處理面臨三大挑戰：1) 數據呈指數級增長，數據量達到PB量級；2) GPU和CPU資源有限；3) 數據處理任務越來越復雜，有多個步驟和復雜的算法。

我們的目標是在有限的資源下，提高數據處理效率。Ray就是答案，它可以處理大量數據，優化異構資源分配，并具有靈活的編排能力。

接下來將介紹在字節跳動，如何使用Ray/RayData構建Audio/Video數據處理Pipeline，以及在大規模不穩定資源上運行RayData所做的優化工作。

首先，讓我們深入探究 Audio 數據處理平臺的細節，了解 Ray 是怎樣解決上述提到的在數據處理中頗具挑戰性的三個問題的。如圖所示，數據處理平臺架構可分為三層：第一層為基礎設施層，它管理基礎的存儲資源與計算資源，以及任務調度，確保可用資源的高效利用以及各類任務的順暢執行。第二層是自主研發的數據處理Pipeline，專門用于處理各種Audio的數據，執行一系列的數據轉換與處理。頂層是應用層，處理后的數據被用于各種業務場景，比如音樂生成等。接下來重點介紹數據處理Pipeline。

數據處理Pipeline中的第一個概念是node，它通常代表一個特定的任務或算子，例如過濾算子、打標算子、去重算子。node可能需要 CPU資源也可能需要 GPU 資源。

第二個概念是flow，flow被定義為節點之間的數據或控制傳輸關系，是一種有向連接。一個node可能有多個輸入flow或輸出flow。最終數據處理Pipeline的DAG由node和flow定義，并通過 YAML 組裝。

基于上述設計，框架定義了很多常見的數據處理Pipeline，通過重新組合這些常見的Pipeline，可以滿足用戶模型訓練的數據處理要求。這種方法極大地提高了整體數據處理的工作效率，并為數據處理和模型訓練提供了更靈活的解決方案。

在構建數據處理Pipeline的初始階段，遇到了一些問題：1）可擴展性不夠，很難從單個節點拓展至多個節點，不能充分利用資源；2）任務調度與負載均衡問題，需要手動管理任務分配，復雜度很高且任務分配不合理；3）高可用性和容錯性，缺乏任務重試和節點故障檢測的自動化機制，在節點中斷時，可能會有任務失敗和數據丟失的情況；4）數據傳輸與共享，需要手動完成任務和節點之間高效的數據傳輸和同步。

在調研了Ray的基礎能力之后，開始嘗試使用RayCore構建Pipeline。因為Ray對Python生態非常友好，使得用戶能夠更高效地進行開發以及和現存方案進行集成，原有方案得以快速遷移到了Ray上。

RayCore 提供了強大的分布式計算能力，比如Actor、Task，使用RayCore可以方便的開發分布式應用程序，構建數據處理pipeline。但是RayCore提供的是low level的API，直接使用它進行開發需要自行處理很多問題，包括不限于：1) 數據切片和分片管理，需要手動管理數據分片和分布，這無疑增加了復雜性；2）數據讀取和加載效率低的問題，缺乏高效的自動化數據讀取和加載機制，會影響整體效率；3）缺乏高級數據操作功能，需要手動實現常見的數據操作，開發成本高。

所以我們開始在Pipeline中使用RayData，它提供了一系列開箱即用的算子，和豐富的多模態數據DataSource支持，自動管理數據分片能力，同時具有自動擴縮容的能力，極大的減少了開發成本。

同時在Pipeline中也使用到了RayServe進行高效的模型部署和服務，RayServe 提供了易于使用的 API，使模型能夠快速轉化為可訪問的服務。另一方面，在高可用性和容錯性方面。RayServe 具有內置的高可用性和容錯機制，可以自動檢測和從故障節點恢復，以確保服務的穩定性。

在構建Audio數據處理Pipeline中，我們看到Ray的優勢有：

1. 良好的可擴展性，從單機到大型集群的輕松無縫擴展
2. 靈活的 API，易于編寫和管理復雜的數據處理任務
3. 完善的數據生態，為各種數據處理需求提供全面的解決方案
4. 高性能，高效地分布式計算、數據傳輸
5. 兼容性好，RayData 與現有的數據處理庫和框架（如 Pandas 和 Spark）兼容，可以輕松集成到現有工作流程中

接下來介紹如何使用Ray來增強Video數據處理Pipeline。大量高質量的視頻數據是視頻生成基礎模型訓練的關鍵，然而與文本圖像和音頻相比，視頻數據量龐大，而且與其他數據格式相比，處理視頻數據需要更多的計算資源和時間。比如，對于視頻數據經常需要使用 ffmpeg 進行視頻編碼和解碼，這需要很多計算資源和計算時間。因此，高效處理大量視頻數據并實現高吞吐量和可擴展性是一個關鍵挑戰。

視頻數據處理流程一般有如下步驟：流程需要處理一系列的原始視頻，時長可以從幾秒到幾小時不等。使用視頻分割算法，將視頻分割成不同的片段，每個片段的時長可以從幾秒到 1 分鐘左右。然后，下一個步驟是視頻處理算法。首先裁剪視頻，以便只選擇需要的視頻，同時還會對視頻進行評分。在這個過程中，需要將元數據存儲到數據庫中，并將片段上傳到對象存儲中以備將來使用。下一步稱為打包，我們使用Ray構建了打包過程，選擇需要的視頻片段，將來自對象存儲和元數據的所有片段放入一個 Parquet 文件中，打包后的Parquet文件將被用于接下來的訓練。

那么什么是視頻數據打包呢？正如剛才提到的，視頻數據打包就是將一組視頻剪輯存儲在 Parquet 文件中，以方便高效的數據管理和訪問。這樣做的目的是避免在訓練階段加載大量小文件，預先將多個小的視頻文件放入一個 Parquet 文件中，然后在訓練階段訓練進程直接加載parquet以提高加載效率。

首先調研直接使用RayData構建打包流程。第一步是利用 RayData 從數據庫中讀取數據，依據不同條件進行過濾操作篩選視頻，隨后把數據集重新劃分成不同的partition，以利于后續進行并行處理。接著是視頻處理，從對象存儲中下載數據，借助 ffmpeg 等框架處理視頻。再將數據打包到 Parquet 并上傳。在實驗中我們發現這個方案有兩個問題，1) 二進制對象的序列化和反序列化，尤其對于大對象會非常耗時；2) 一旦 ObjectStore 滿，Ray就會把 Object Spill 到磁盤上，從而影響整體性能。

所以，打包過程中將視頻數據像存儲在 objectstore中，特別是在大容量的情況下，效率不高。嘗試使用另一個解決方案，將所有操作融合到單個 actor 中，以避免 actor 之間的數據傳輸。如上圖所示，actor內部啟動多個線程一起運行，在每個線程中下載視頻并運行視頻處理操作，然后寫入 Parquet 文件并上傳到外部存儲。這個解決方案效果很好，可以實現高吞吐量，并且具有良好的線性擴展性，增加更多的 CPU 資源，帶寬也會相應地增加。

接下來分享一些使用Ray的經驗以及Ray的優勢：

1）Ray具有可擴展性和靈活性，可以輕松地從本地 Python 腳本擴展到大規模集群，在數千個工作節點的規模下也能運行良好；

2）對Python友好，ML場景中大量使用 Python，Ray對python非常友好，非常方便進行開發調試，與現有ML生態也結合的比較好；

3）Ray Dashboard提供作業相關的Restful API，可以非常方便地將這些API集成到業務平臺中，包括提交和監控Ray作業的運行狀態；

接下來介紹Ray相關的底層基礎設施。在字節跳動，Ray被應用在很多業務場景中，包括但不限于Audio/Video數據處理、RLHF等。Ray支持非常靈活的編排和異構資源（CPU/GPU）的調度能力，幫助用戶進行靈活的多角色 DAG 編排和異構計算，構建大規模高性能的 ML 基礎設施。但是，像許多其他大規模分布式系統一樣，生產環境中使用Ray也面臨著巨大的挑戰。

LLMs的數據處理任務通常需要巨大的資源需求和相對較長的處理時間，一般做為離線處理任務。為了降低成本，會使用大量不穩定的Kubernetes Pod來運行這些數據處理任務，這些Pod可以隨時被搶占，也可能隨時重新添加進來。我們希望Worker節點被搶占不會導致數據處理任務失敗或中斷。比如，如果一個Ray任務在 100 個 GPU 上運行，其中 40 個GPU被搶占，期望剩余的 60 個 GPU 能夠繼續高效運行，這是一個非常大的挑戰。雖然 RayCore 提供了強大的 Actor 和 Task 恢復機制，但在當前的 RayData 設計中，當一個 Actor 異常退出，必須等待 Actor 重新啟動才能繼續執行 task，如果資源不足，Actor將處于Pending狀態，因此RayData任務也會hang住，直到資源恢復。為了解決這些問題，我們設計并開發了RayData增強方案。

第一個增強是在RayData調度器重中進行任務重新分配。簡單來說就是將RayData Actor Pool中失敗的task分派給Actor Pool中其他的actor運行。具體做法是，將Actor Pool中actor的max_restarts設置為0，也就是完全由RayData掌控actor的生命周期，這樣在actor掛掉后RayCore不會再重啟它。當RayData調度器檢測到actor異常退出時，原先分配給它的未完成的task，會被重新分配給其他actor。RayData調度器重新創建一個map actor，如果沒有資源，actor處在pending狀態，但是這不會阻礙RayData任務的正常運行。一旦新的資源加入，actor就會變為running，重新加入到actor pool中。

任務重新分配的策略非常簡單，但會引入一個新問題。由于任務重新分配需要將 actor 的 max_restarts 參數設置為 0，那么當一個object丟失時，就無法再依賴 RayCore 的血緣重建來重建object。例如，在右側的圖中，map算子1的輸出作為 map算子2 的輸入。如果運行 map算子1的節點異常退出，它輸出對象就會丟失，當 map算子 2 嘗試讀取該對象時，會出現object丟失的情況從而無法繼續處理。

為了解決這個問題，進一步提出了第二個優化方案——由RayData 管理算子之間輸入輸出數據的血緣關系，而不是依賴 RayCore 的對象血緣關系。RayData 調度器引入了一個與血緣相關的數據結構。每個算子都有一個表，用于記錄輸入和輸出數據之間的關系，其中key是輸出引用value是輸入引用，方便從輸出引用查找到輸入引用。當下游算子（例如 Map-B）的 actor 遇到object丟失時，意味著其上游（Map-A）的輸出丟失，RayData 調度器通過血緣表查找到input，并重新計算其輸出。當一個block流經所有算子產生輸出后，可以刪除其所有上游對應的血緣表。目前該解決方案僅支持OneToOne算子，但是理論上也完全支持 AllToAll 算子。

對于大多數RayData用戶來說，一般使用的是穩定資源，不會遇到前面介紹的問題。但是我們認為“RayData不穩定資源下的問題”其實是一個通用的問題，這里介紹的解決方案具有一定的通用性。

隨著模型的逐漸增大，actor故障后恢復時間會比較長，下載+加載模型會達到分鐘級，上游算子actor的故障對下游的影響會逐步放大。如上圖中，如果map op1的某個actor異常退出，短時間內沒有恢復，該actor的輸出已經被調度給了map op2的兩個actor，map-op2-1和map-op2-2。map-op2-1和map-op2-2因為輸入數據沒有ready，會處于waiting空閑狀態，等待上游恢復，此時也無法繼續處理后續數據，利用率降低。map op2給到下游map op3的輸出也減少，利用率也會降低。map op2和map op3變得空閑，也會觸發raydata的autoscale，這可能是不必要的，會有額外的抖動。而任務重新分配和RayData血緣方案可以有效減小 actor 的異常退出對整個作業的影響。

總結一下今天分享的內容。我們深入探討了如何使用Ray構建可擴展的Audio/Video數據處理Pipeline，分享了在不穩定的 Kubernetes 節點上運行 RayData 的經驗，并提出了對 RayData的改進方案。