單機每秒最多可處理10億條數據！eBay開源數據處理框架Accelerator

策劃編輯 | Natalie

編譯 | 無明

編輯 | Emily

AI 前線導讀：近日，eBay 宣佈正式開源 Accelerator，一款久經考驗的數據處理框架，提供快速的數據訪問，並行執行以及自動組織源碼、輸入數據和結果。它能夠用於平常數據分析，也能夠用在包含數十萬大型數據文件的實時推薦系統上。

Accelerator 可運行在筆記本電腦或機架式服務器上，輕鬆處理數十億行數據，井井有理地處理成千上萬的輸入文件、計算和結果。

Accelerator 的數據吞吐量一般在每秒數百萬行。若是運行在高速計算機上，每秒最多可處理幾十億行數據。

更多幹貨內容請關注微信公衆號「AI 前線」，（ID：ai-front）

Accelerator 最初由瑞典人工智能公司 Expertmaker 開發，於 2012 年正式發佈，從那之後，它一直是衆多研究項目和實時推薦系統的核心工具。 2016 年，Expertmaker 被 eBay 收購，而 eBay 目前正在基於 Apache 許可協議第 2 版開源 Expertmaker Accelerator。

設計目標

Accelerator 的主要設計目標以下：

• 簡化在多個 CPU 上並行處理數據。

• 數據吞吐量應儘量快，即便一臺小型筆記本電腦也能輕鬆處理數百萬行數據。

• 若是可能，儘可能重用計算結果，而不是從新計算。一樣，在多個用戶之間共享結果應該是絕不費力的。

• 數據科學項目可能會有不少（數十萬）輸入文件和大量的源碼和中間結果。

• Accelerator 應該避免手動管理和記錄數據文件、計算、結果以及它們之間的關係。

主要功能

Accelerator 主要的原子操做是建立做業。 建立做業是用輸入數據和參數執行一些程序並將結果（即輸出）以及計算所需的全部信息存儲到磁盤上的過程。 做業目錄將包含計算結果和計算結果所需的全部信息。

做業能夠是簡單或複雜的計算，也能夠是大型數據集的容器。 做業之間能夠彼此連接，新做業能夠依賴於一個或多箇舊做業。

關鍵特性

Accelerator 提供了兩個關鍵功能，結果重用和數據流。

結果重用

在建立新做業以前，Accelerator 會檢查以前是否已經跑過相同的做業。若是已經存在，Accelerator 不會建立這個做業，而是將現有做業的連接返回。這樣不只節省了執行時間，並且有助於在用戶之間共享結果。更重要的是，它提供了可見性和肯定性。

Accelerator 提供了一種機制，將會話中的做業信息保存到數據庫中，這樣有助於管理做業和它們相互之間的關係。

數據流

將連續的數據流從磁盤傳輸到 CPU 比在數據庫中執行隨機查詢效率更高。流式傳輸是實現從磁盤到 CPU 高帶寬的最佳途徑。它不須要緩存，能夠很好地利用操做系統的基於 RAM 的磁盤緩衝區。

總體架構

如今讓咱們來看看 Accelerator 的總體架構。

是一個基於客戶端 / 服務器的應用程序。在左側有一個 runner 客戶端，在右邊有兩臺服務器，稱爲 daemon 和 urd，其中 urd 是可選的。runner 經過執行腳本（構建腳本）在 daemon 服務器上啓動做業。此服務器將加載和存儲使用基於 workdirs 文件系統的數據庫執行的全部做業的信息和結果。同時，構建腳本中全部有關做業的信息將由 urd 服務器存儲到做業日誌文件系統的數據庫中。 urd 負責管理做業，包括存儲和檢索以前執行過的相關做業的會話或清單。

做業

做業是經過執行稱爲 method 的小程序來建立的。method 用 Python 2 或 Python 3 編寫，有時也用 C 語言。

最簡單的做業：「Hello, World」

咱們經過一個簡單的「Hello World」程序來講明如何建立一個做業（method）：

def synthesis():
  return "hello world"複製代碼

這個程序不須要任何輸入參數，只是返回一個字符串並退出。要執行它，咱們還須要建立一個構建腳本，以下所示：

def main(urd):
  jid = urd.build('hello_world')複製代碼

當執行完這個方法以後，用戶會獲得一個叫做 jobid 的連接。jobid 指向存儲執行結果的目錄，以及運行做業所需的全部信息。

若是咱們嘗試再次執行這個做業，它將不會被執行，而是返回指向上一次執行做業的 jobid，由於 Accelerator 記得以前已經執行過與此相似的做業。 要再次執行做業，咱們必須更改源代碼或輸入參數。

連接做業

咱們假設剛剛建立的 hello_world 做業很是耗費計算資源，並已經返回了咱們想要的結果。爲了簡單起見，咱們經過建立一個名爲 print_result 的方法來演示其中的原理，該方法只讀取前一個做業的結果並將結果打印到 stdout。

import blob

jobids = ('hello_world_job',)

def synthesis(): 
  x = blob.load(jobid=jobids.hello_world_job)
  print(x)複製代碼

要建立這個做業，咱們須要擴展構建腳本：

def main(urd):
  jid = urd.build('hello_world') 
  urd.build('print_result', jobids=dict(hello_world_job=jid))複製代碼

在執行構建腳本時，只會建立 print_result 做業，由於 hello_world 做業以前已經建立過了。

做業執行流程和結果傳遞

到目前爲止，咱們已經知道如何建立、連接和執行簡單的做業。如今咱們將重點轉向 method。在執行 method 時，Accelerator 會調用三個函數，它們分別是 prepare()、analysis() 和 synthesis()。一個 method 能夠同時調用這三個函數，或者至少調用一個。

三個函數的返回值均可以存儲在做業的目錄中，並被用在其餘做業上。

數據集

數據集是 Accelerator 默認的存儲類型，專爲並行處理和高性能而設計。數據集創建在做業之上，所以數據集經過 method 來建立，並存儲在做業目錄中。單個做業能夠包含任意數量的數據集。

在內部，數據集中的數據以行列格式存儲。全部列均可以被獨立訪問，避免讀取到沒必要要的數據。數據也被分紅固定數量的片斷，提供並行訪問能力。數據集可能會被散列，散列函數將具備相同散列值的數據行組合到同一個片斷中。

導入數據

讓咱們來看看導入文件（建立數據集）的常見操做。csvimport 方法可用於導入許多不一樣的文件類型，它能夠解析大量的 CSV 格式的文件，並將數據存儲爲數據集。建立的數據集存儲在結果做業中，數據集的名稱默認爲 jobid 加上字符串 default，也可使用自定義字符串。

連接數據集

就像做業同樣，數據集也能夠相互連接。因爲數據集是創建在做業之上的，因此連接數據集就很簡單。例如，假設咱們剛剛將 file0.txt 導入 imp-0，而且 file1.txt 中存儲了更多數據。咱們能夠導入後一個文件並提供一個指向前一個數據集的連接。因爲數據集已連接，如今可使用 imp-1（或 imp-1/default）數據集引用來訪問從這兩個數據集導入的全部數據文件。

在處理隨時間增加的數據（如日誌數據）時，使用連接十分方便。咱們能夠經過連接擴展具備更多行的數據集，這是一個很是輕量級的操做。

將新列添加到數據集

添加列是很經常使用操做，Accelerator 經過連接來處理新列。

原理很簡單，假設咱們有一個「源」數據集，咱們要添加一個新列，只須要建立一個只包含新列的新數據集，並在建立它時讓 Accelerator 將全部源數據集的列連接到新數據集。

並行執行

Accelerator 專爲並行處理而設計，主要經過分片數據集和並行 analysis() 調用組合來實現並行處理。

迭代器在 analysis() 函數內部運行，該函數爲每一個數據集片斷 fork 一次。analysis() 函數的返回值將做爲 synthesis() 函數的輸入。咱們能夠顯式地合併結果，不過 analysis_res 帶有一個至關神奇的方法 merge_auto()，它根據數據類型將全部片斷的結果合併爲一個。

urd

咱們已經看到 Accelerator 如何跟蹤已經建立好的做業，並在必要時重用做業。這樣節省了時間並將相關的計算連接在一塊兒，不過，在這之上還有另外一個層能夠進一步提升可視性和做業重用性，它就是 urd 服務器。

urd 將做業清單及其依賴關係存儲在基於日誌文件的數據庫中。在構建腳本中發生的全部事情均可以記錄到 urd 中。爲了作到這一點，咱們須要一個名單來存儲信息，還須要一個密鑰，而在大多數狀況下還須要一個日期，方便往後查找。

性能測試

新做業的啓動時間只有幾分之一秒。如下是一些不一樣做業類型的處理時間。

準備數據：導入、類型轉換和散列

示例數據文件大小爲 1.1TB（壓縮後 280GB），包含 63 億行和 14 列。Accelerator 運行在具備 72 核心和快速磁盤的大型機上。

上述數值是基於所有數據得出的。導入做業（A）：導入 gz 壓縮文件。有趣的是，導入比普通的 zcat file.gz> /dev/null 要快 30％。在 FreeBSD 上，zcat 速度更快。類型轉換做業（B）：5 個 json-list、5 個數字、2 個日期和 2 個 unicode 列，每行平均有 172 個字節。該做業的讀取速度超過每秒半千兆字節，同時保存幾乎相同數量的數據到磁盤上，所以磁盤帶寬高於每秒 1 千兆字節。因爲散列速度取決於被散列的列，所以顯示的值（C）是四個散列做業的平均值。

處理數據

爲了計算Σ(a×b×c)，咱們經過一個 method 讀取三個列、將它們的值相乘並將結果寫入新列。第二個做業爲新列添加值。

能夠看到，將三個 float64 相乘並寫回到磁盤其實是很快的——每秒 7 千 7 百萬行。將這些值彙總在一塊兒甚至更快——每秒超過十億個值。而在 Python 中，執行一樣操做須要 6 秒。

結論

Accelerator 是一款用於快速數據處理的工具。在單機上，每秒能夠處理數百萬行數據，若是任務簡單，能夠每秒處理 10 億行。除了速度快以外，Accelerator 還能夠減小手動管理源文件、數據文件、計算以及相關結果的工做。它已被成功用在多個項目中，eBay 如今正式將其開源。

相關連接：

ExpertMaker Accelerator 代碼倉庫 (https://github.com/eBay/accelerator)

Installer 倉庫 (https://github.com/eBay/accelerator-project_skeleton)

Accelerator 用戶參考手冊 (https://berkeman.github.io/pdf/acc_manual.pdf)

更多幹貨內容請關注微信公衆號「AI 前線」，（ID：ai-front）