為了應對大規模數據處理中的性能和互操作性挑戰。Arrow 致力于提供一種高效的跨平臺數據交換機制，使數據能夠在不同操作系統和編程語言之間迅速、一致地流動。其設計注重性能優化，并得到開源社區廣泛的支持，成為許多數據處理項目的核心組件，推動了數據科學和分析領域的創新。本文將分享如何使用 Apache Arrow 來構建一個完整的數據系統。

一、為什么要構建新的數據系統

首先需要回答的問題就是為什么要構建一個新的數據系統。

1. One Size Fits All or Not？

據圖靈獎得主 Stonebraker 在 2005 年的一篇論文中指出，"One Size Fits All" 的概念已經過時。這意味著使用一個通用的數據系統（如 Oracle 或 MySQL）無法完全滿足所有問題的需求。因此，許多領域都需要特定的專用系統。一個明顯的例子是交易場景中的 OLTP（聯機事務處理）和分析場景中的 OLAP（聯機分析處理）的分離。此外，還有眾多由于數據處理模型的不同而產生的新系統，如適合實時計算的流處理和更適應云計算的 NoSQL 等等。然而有些人持不同意見，他們嘗試在一個系統中解決所有問題。例如，試圖同時處理 OLTP 和 OLAP 的 HTAP，還有流批一體、NewSQL 等新概念。無論是否支持"One Size Fits All"的觀點，新的想法和概念總是不斷涌現， 所以也需要不斷構建新的數據系統。

2. 數據庫的黃金時代

當前，我們正處于一個數據系統的黃金時代。有許多新的數據庫正在不斷出現。根據 DBDB（Database of Databases）的統計，自 2020 年以來已經出現了 124 個新的數據庫。這意味著幾乎每周都會有一個新的數據系統問世。

有了這么多數據系統，我們為什么還需要構造一個新的數據系統，這是否是重復造輪子呢？

3. 讀時建模

我們面對的場景是通用的日志處理，這是一項頗具挑戰的任務，因為不同廠商，甚至同一廠商不同應用的日志格式各不相同，而且一個尚處于開發中的系統的日志格式也會經常變動。常見的方法之一是寫時建模，即通過在數據庫中預先定義表結構或者模式來存儲日志數據。然后，通過一些 ETL 工具，將數據轉化為所需的表格類型。然而，當面對不同類型的數據時，例如日志來自 Nginx、Apache 和 Windows IIS 等不同系統，就需要維護多個 ETL 流程來處理它們。另一種解決方案是讀時建模，即先將原始數據存儲起來，不進行 ETL 處理，在實際使用時對字段進行抽取。例如圖中查詢 1，就是從這三類日志數據中抽取出各自的 method、time 和 client 字段，并實時構建一個新的結果表格。這種方法更適合處理多樣化的、不固定的數據模式。有興趣可以掃描上圖中的二維碼，它提供了一個有關讀時建模更加詳細的分享。

讀時建模的關鍵是有一個支持動態數據表模式的數據查詢引擎，才能夠在沒有預先定義數據模式的情況下查詢各種類型的數據。

下面介紹一下動態數據表模式可能是什么樣子的。在我們收集的數據中，可能有不同格式的日志。例如，有以鍵值對表示的日志，如綠色框所示；也可能有以 JSON 格式表示的日志，如紅色框所示。在支持動態數據表模式的系統中，可以在同一數據集中同時查詢到這兩種類型的日志。

接下來，我們可以通過字段提取的方式對這兩類數據進行格式化處理，從而得到一個合并在一起的二維表結構。

最后，可以使用一些函數對數據進行歸一化和合并。這樣做可以得到一個直觀明確的數據表，便于后續機器學習系統和報表系統進行處理。在這個過程中，我們可能會遇到同一個字段有不同類型的情況。例如，某個同名字段在某類日志中可能以整數形式存儲，而在另一個日志中可能以浮點數形式存儲；或者在導入數據時沒有進行適當處理，導致某些字段被導入為字符串類型。因此，同一個字段中可能同時包含多種不同類型的數據。

構建數據庫是一項復雜的工程，其中包含許多子任務?？梢赃x擇改造一個開源項目，也可以從頭開始構建?，F有項目要么無法滿足讀時建模的需求，要么架構相對陳舊，不適合現代云計算服務的架構。而且改造現有系統，難度也比較大，因為具有固定的模式是很多系統的前置條件。因此，我們決定從頭開始構建這個系統，第一步就是定義內存中的數據格式。

二、什么是 Apache Arrow，為什么是 Apache Arrow

下面介紹一下什么是 Apache Arrow，我們為什么選擇了 Apache Arrow，以及我們要用它來做一些什么事情。

1. 內存數據格式

內存中的數據可以根據分布的不同，分為行式存儲和列式存儲。在行式存儲中，數據按照行的方式排列，類似于 C 或 C++ 中的二維數組，每一行連續存儲，將相關的信息放在一起。而在列式存儲中，同一列的數據放在一起，構成一個長數組。圖中是包含三個屬性（session_id，timestamp 和 ip）的二維表，左邊是它在行式存儲中的內存表示，右邊是它在列式存儲中的內存表示。

在事務處理的場景中，行式存儲是非常自然的選擇。因為大部分操作都是以一行為單位進行的，可以使用 session_id 構建 B 樹或 B+ 樹等索引，通過這個索引很快定位到 session_id 對應的具體的某一行，然后進行刪除或修改操作。

但在分析型場景中，用戶通常不會使用所有屬性，而是只會對某些屬性進行運算。例如，如果想知道表中有多少個不同的 IP 地址時，只需要關心 ip 這一列，而不關心 session_id 或 timestamp。在這種情況下，列式存儲就更為適合，因為只需讀取 ip 這一列數據，從而避免讀取其他不必要數據的開銷，減少了 IO 和內存的消耗。而且，列式存儲的數據在磁盤或內存中都是連續存儲的，可以更好保證數據的局部性，從而充分利用現代 CPU 的緩存和 SIMD（單指令多數據）向量化等運算機制。另外，相同屬性的數據放在一起可以更好地進行壓縮。例如，可以使用字典或其他高效的壓縮算法將這些具有相同屬性的數據放在一起進行壓縮。對于日志處理這種分析型的應用來說，列存儲是更合適的選擇。

2. Apache Arrow 的優勢

那么我們還要不要繼續造輪子？還是定義一套自己的列式存儲內存格式？

現在許多系統都定義了自己的內存數據格式，這帶來了數據轉換的問題。試想，如果我們想在 pandas 中調用 Spark 進行數據處理，那么要從 pandas 基于的 Python 環境轉換到 Spark 基于的 Java 環境，其中需要經過 Py4J、JVM 和 Spark 三層數據轉換。同樣地，如果我們定義了自己的內存格式，也會面臨類似的數據轉換問題，特別是在需要與其他系統進行互操作時。幸運的是，借助 Apache Arrow，可以在 PySpark 啟用 Arrow 格式，就可以與 pandas 直接共享內存，實現內存交換。

Arrow 到底是什么呢？Arrow 本身并不是一個數據存儲或執行引擎，而是一種高性能、內存中的列式存儲標準。它與具體的語言或應用程序無關，無論是用 C++、Python 還是 Rust 等語言，都可以進行跨語言跨系統的互操作。因為在任何環境中，Arrow 數據的內存表示是完全一致的，所以在進行跨系統傳輸時，不需要進行內存拷貝、序列化或反序列化等工作，實現了零拷貝。Arrow 沒有發明新的數據存儲方式，比如浮點數仍然按 IEEE 754 標準進行表示，但 Arrow 在標準化方面做了很多工作，例如如何表示空值 NULL、如何處理時間戳以及時區的表示等等。這些細節看起來很微小，但它們的重要性是在任何平臺和任何語言下的標準化。因此，一個全新的數據引擎也無需重新發明這些內存格式。

使用 Arrow 后，可以實現在不同系統之間共享內存，從而實現零拷貝。這意味著我們不是避免了繁重的數據復制和轉換，而是直接共享內存中的數據，使得數據處理過程更加高效。

Arrow 還有以下幾個好處。

首先，Arrow 原生實現了七種程序語言，并在此基礎上實現了更多語言的綁定，包括 Rust、C++、C、Python 等，基本覆蓋了主流的程序語言。并且得到大量數據系統的支持，如 PyTorch、Spark、ClickHouse 和 DuckDB 等，在這些系統中，數據可以采用 Arrow 格式進行輸出。

其次，Arrow 的性能表現不錯。有一個 Benchmark 對比了基于 Arrow 的數據引擎 DataFusion、Polars 與 DuckDB 的性能，雖然前兩者稍慢于 DuckDB，但仍然是可接受的成績。盡管看起來 Arrow 在每個小功能點上沒有什么創新，但綜合起來，它提供了一個相對完整的解決方案，并且模塊化做得非常好，API 對于系統的侵入性也較小。

此外，Arrow 的擴展性較強，比如可以擴展 Arrow 的類型，將機器學習中的一些類型如 tensor 在 Arrow 中實現；也可以使用 API 擴展自定義的計算函數。

總而言之，Arrow 的主要貢獻在于為列式存儲提供了一個標準和生態系統，因此對開發者和用戶來說，它可以作為一個現代數據技術棧的標準和基礎。

Arrow 擁有非?；钴S的開源社區。除了 Arrow Rust 等相關項目外，截至 2023 年 10 月，Arrow 本身已經得到了超過一萬兩千個 GitHub 的 star。上圖展示了 Arrow 在最近一個月的活躍程度，包括 PR、issue 以及貢獻者等方面的數據，足以看到 Arrow 是一個非常活躍的項目，并不需要擔心它的持續性和穩定性，可以預期 Arrow 能夠長期存在并會持續完善。

Arrow 通過幫助標準化內存格式，為構建數據系統提供了一個起點。然而，這只是開始，還有許多其他任務有待完成。例如，需要聚合、排序等更多的算子；需要開發客戶端 API 和數據交換功能；需要支持新的硬件，尤其是在信創領域，需要考慮對 ARM 指令集和國產硬件的兼容。

如此一來，構建一個數據系統變得非常復雜。上述工作也只能初步構建一個勉強滿足小規模使用的數據系統，而完成這個階段可能需要 10 年甚至更長時間。如果我們想要構建一個更大規模、分布式和高可用的數據系統，所需要的時間可能是前一階段的幾倍甚至更長。

按照人月神話的理論，投入更多的人力并不能線性地減少完成時間。因此，構建新的數據庫是一個非常昂貴的事情，需要巨大的時間、人力和經濟成本，這也是為什么新的數據庫創業公司需要籌集大量資金和足夠的時間。

雖然現在是數據庫的黃金時代，但也是最具挑戰性的時期。如果新的想法不能迅速實現，很難在市場上生存。好在有 Arrow。Arrow 不僅提供了基本的內存數據格式和模型，還提供了一些算子和計算功能，以及持久化、數據交換和跨平臺執行等模塊。通過使用 Arrow，能夠大大節省構建數據系統的時間和開發成本。

三、使用 Apache Arrow 構建數據系統

下面介紹 Arrow 如何助力數據系統的開發以及如何使用 Arrow 構建一個數據系統。

1. 數據系統執行流程

一個數據系統的執行流程通常包括以下幾個步驟。

首先，當系統接收到用戶的查詢請求時，會利用存儲和索引來獲取相關資源。

接下來，系統會根據用戶查詢生成一個邏輯計劃，該計劃表示了執行查詢所需的關系代數和操作的抽象。

然后，邏輯計劃會在經過優化之后轉為物理計劃，即如何真正執行查詢的計劃。

之后，在執行引擎中，系統會執行具體的操作，如表達式執行、聚合、排序和物化視圖等算子。

最后，系統將結果保存到用戶指定的路徑或傳輸到用戶的客戶端。

2. 數據存儲

我們的數據存儲模型是基于事件的，即基本的存儲單元抽象成了事件，類似于日志中的每一條日志。每個事件都有時間戳、原始信息和其他基本屬性，比如主機名、數據類型等。這些都是事件的元信息定義，我們將其抽象出來，并進行索引。

對于日志中的其他內容，我們將其作為原始數據存儲。底層存儲使用了 Parquet 這一列式持久化存儲標準，其對 Arrow 有很好的支持。Parquet 還會存儲一些元數據，比如每列的存儲位置和一些統計信息，如最大值和最小值等。這樣就可以支持一些查詢的下推操作。如果數據中存儲了多列，但只想訪問某一列，可以直接定位到該列的存儲位置，而不需要將整個文件都讀入內存中。

然而，Parquet 需要預先給定數據的模式，即存儲數據時需要先定義一個模式，無法直接支持動態模式或者無模式數據。為了支持動態模式的數據，在 Parquet 的基礎上我們進行了一些擴展，這樣就可以在 Arrow 和 Parquet 的基礎上進行簡單的修改，從而完成數據存儲。

數據存儲之后，需要讀入到內存中。每個數據在內存中會以 Arrow 定義的 Record Batch 形式存在。這種表示方法用于描述一組數據，并由其 Schema 指定數據的模式。

例如，有一列包含 session_id 字段的 Int64 類型數據，一列包含 datetime 字段的String 類型數據，還有一列包含 source_ip 字段的 String 類型數據，Schema 中定義并存儲了這些字段的類型，而具體的數據存儲在 Arrow Array 中，不同 Record Batch 的 Schema 是可以變動的。例如，在下一個 Record Batch 中，session_id 字段可能變成 String 類型，而 time 字段可能變成 Timestamp 類型。

通過不同模式 Record Batch 的組合，就可以獲得不同模式的數據。這樣，就實現了從數據存儲到內存表示的映射關系。

3. 索引/代碼/硬件資源

Arrow 并不是一個完整的查詢引擎。它缺少索引和用戶自定義函數等功能的支持，在我們的系統中，我們使用了時間戳索引和倒排索引，這樣用戶可以通過關鍵字和時間來定位到日志的位置。至于用戶自定義函數方面，我們向 Arrow 提交了一系列 PR，使其能夠支持用戶自定義函數。Arrow 在硬件資源方面有一些簡單的實現，比如內存管理和線程池。但是，如果想要進行更細粒度的管理，例如限制每個查詢的內存使用或設置不同查詢任務的優先級，仍然需要自己開發。所以，從這個角度來看，Arrow 在這方面還有繼續完善的空間。

4. SQL 解析/計劃生成/執行與傳輸

Arrow 也沒有提供將用戶的 SQL 語句解析成抽象語法樹的功能，但是我們可以使用一些開源工具，比如 ANTLR 和 Calcite，將 SQL 語句轉換成抽象語法樹。我們選擇使用 ANTLR 而不是 Calcite，是因為 Calcite 過于復雜且基于固定數據模式的假設，在處理動態模式時不太適用。

之后可以將抽象語法樹進一步轉換成邏輯計劃，邏輯計劃描述了數據執行的具體操作。在進行查詢優化時，我們可以調整邏輯計劃來提高性能。

例如想要找到特定 ip 的最新訪問時間，首先需要從數據集中讀取相關數據。然后，根據指定的條件（這里是 ip 等于某個特定字符串）進行數據過濾，并將需要的數據篩選出來。接下來，對過濾后的數據進行聚合運算計算其時間的最大值。

在此過程中，可以進行一些優化，其中一個常見的優化是下推操作。通過下推，可以將讀取 ip 和 _time 兩個字段的操作下推到表掃描階段，從而每次讀取數據時都跳過其他不必要的字段。此外，我們還可以將條件表達式（例如，ip 等于特定字符串）嵌入到操作中，這樣每次讀取時只會讀取與我們需要的 ip 相匹配的數據。

通過在表掃描階段進行這些優化，可以節省大量的 IO 開銷和內存資源，提高查詢性能。

邏輯計劃是一個抽象層，不包含在 Arrow 中，因此需要自己編寫邏輯計劃的代碼。

相對而言，邏輯計劃相對簡單，因為大多數 SQL 查詢語言及關系代數和邏輯計劃可以相互對應。物理計劃則相對復雜，因為它與底層機器有關，需要處理線程、并發和各種硬件。

最近，Arrow 提供了一個查詢執行引擎——Acero，可以提供很大幫助。Acero 是一個基于推送（Push）的引擎，其最小執行單元是 execution node，它的代碼非常清晰，并且具有清晰的 API 接口，包括如何處理其上游輸入和下游輸出，如何處理接收到的數據和停止接收數據，以及暫停和繼續運行等功能。

如果需要擴展，只需按照 API 定義自己的節點，并在 Acero 中注冊即可，就可以借助 Acero 進行計算的調度和執行，而不需要修改 Arrow 代碼。

我們注意到 Arrow 在處理動態數據模式方面存在一些限制，因此對 Arrow 進行了一些擴展。例如，添加了支持動態模式的匯聚節點 Schemaless SinkNode，它可以消除數據模式方面的一些限制。通過使用這個節點，可以處理沒有嚴格定義模式的數據。這允許我們更靈活地處理各種數據類型，而不僅僅限于特定的固定模式。

在這個過程中，我們得到了一個支持動態模式物理計劃的執行節點。此外，Arrow 的另一個限制在于執行節點創建時就需要預先定義數據的輸出模式。為了克服這個限制，我們進行了一些改造，將數據輸出模式延遲到實際輸出時動態生成。這樣，就能更好地支持動態模式的數據引擎。另外，我們也對 Arrow 提供的一些聚合函數和標量函數進行了動態模式的擴展。

這樣就可以使用 Arrow 來處理動態模式數據，并使用它執行并調度查詢的。目前，Acero 還不支持物化視圖，但對于大規模數據來說，物化視圖非常重要。物化視圖可以預先計算并且儲存一些耗時或復雜場景的結果，在查詢時可以快速訪問和利用這些預先計算的結果。同樣，我們對 Acero 進行了一些擴展，添加了中間狀態的處理方式，以便在 Arrow 中實現物化視圖，我們也計劃將這些一系列擴展提交給 Arrow。

最后，當查詢結束時，需要進行數據傳輸，可以是傳輸給用戶的客戶端，也可以是傳輸到用戶前端進行顯示。如果直接使用 ODBC 或者 JDBC，因為 ODBC 和 JDBC 本質上只能處理行式數據，行列的轉換無法避免，我們可以使用 Arrow Flight 和 Arrow Flight SQL 來規避這個問題。

Arrow Flight 是 Arrow 提供的基于 gRPC 或者 REST  的列式數據交換框架，無需復雜的開發，直接使用其 API 即可實現列式數據傳輸，而避免了數據轉換。在 Arrow Flight 之上得到了與 SQL 數據庫交互的協議 Arrow Flight SQL。這樣我們就可以利用與 SQL 兼容的現有客戶端直接進行查詢。

將來，Arrow 還將推出一個類似于 Arrow 自己的 JDBC 或者 ODBC 的工具，稱為 Arrow ADBC。這樣，原本與 ODBC 和 JDBC 兼容的數據庫客戶端將無需或只需極少修改代碼，就可以直接與 Arrow 進行通信。

Arrow 幫助我們實現了數據存儲、物理計劃和傳輸這三個方面的重要功能。如果在自己的數據系統實現中不是動態模式的，而僅僅是針對特定領域開發固定模式的新系統，那么只需構建索引、解析用以查詢的 SQL 或 Dataframe API，并轉換成邏輯計劃，然后使用 Calcite 的優化器將其轉換為 Arrow 的物理計劃，最后直接使用 Arrow 執行即可，需要構建的東西非常少。

四、一些 Tips

我們在 Arrow 的使用中積累了一些經驗和教訓。作為一個新的數據產品或數據產品的底座，Arrow 還存在不少問題。

1. 踩過的一些坑

首先更新頻繁是 Arrow 社區活躍的體現，意味著會有新的功能和改進，但同時它的接口還是不夠完善，我們建議盡量少修改原始代碼，而是向 Arrow 社區貢獻改進并多做擴展。Arrow 代碼庫可以分為三個層次：

• Core 層：提供數據類型表示，這一層非常穩定，新版本可以完全保證和之前版本的兼容。
• Compute 層：提供計算算子，相對穩定但可能有一些 bug，當使用一些比較高級的指令集如 AVX512 指令集可能會有一些內存對齊的問題。
• Acero 層：是最新的執行引擎，不夠穩定而更適合開發測試。

Arrow 對于復雜類型的處理還不夠完備，比如 Union、List、JSON 等，需要額外的代碼實現。另外，Arrow 始于 2016 年，仍需要時間和大規模數據的驗證。各個相關項目（包括 DuckDB 等）主要使用的是 Core 部分，對于 Arrow 的 Compute 和 Acero 等部分，仍然需要在更大規模的數據上進行進一步的驗證。我們在開發過程中遇到了一些問題已經修復并向 Arrow 社區提交了改進。目前看來，Arrow 處于相對穩定的狀態。

2. DATA FUSION

最后，對于追求安全和現代化的考慮，我們建議使用 Arrow Rust 的實現。而且 Arrow 在 Rust 實現的基礎上推出了一個完整的數據引擎 DataFusion，它提供了比 Arrow 更強大的功能。DataFusion 在 Arrow 內存格式的基礎上提供了 SQL 解析和查詢計劃等功能，也支持子查詢和其他高級函數。此外，DataFusion 也繼承了 Arrow 出色的模塊化和可擴展的代碼風格，基于 DataFusion 構建新的數據引擎可以減少開發所需的時間，同時也能在開源社區獲得更多支持。