導讀 本文将介紹 Apache Doris 的前世今生，資料湖分析的技術細節以及後期的規劃。
分享包括以下幾大部分：
1. Doris 簡介
2. Doris 資料湖分析技術細節
3. Doris 社群發展以及後期開發規劃
分享嘉賓｜陳明雨 Apache Doris PMC
編輯整理｜鐘理 遠信集團
出品社群｜DataFun
01
Doris 簡介
什麼是 Apache Doris？簡單來說，Doris 是一款基于 MPP 架構的高性能實時的分析型資料庫。
下圖是 Doris 的發展曆程。最早可以追溯到 2013 年。
它是百度内部自研的一個多元分析平台。經過了幾年在百度内部大規模的應用和實踐，2017 年的時候，正式開源到 Github 上。在 2018 年 Doris 進入到 Apache 孵化器，在孵化的過程中，不斷發展社群，培養使用者和開發者。到目前為止，在孵化期内釋出了七個重要的版本，每月的活躍開發者接近一百位。在 2022 年，Doris 從 Apache 孵化器畢業，成為一個頂級項目。成為頂級項目之後，我們也快速的推動社群的發展。在 2022 年 12 月份釋出了 Doris 1.2.0 版本。這一版本中有很多重要功能更新。其中很重要的一部分就是對資料服務能力的優化和支援。這也是本次分享的重點。
下圖中可以看到 Doris 在整個資料流中的定位：
它的上遊有一些 OLTP 系統，日志系統，埋點系統。經過一些流處理或者說批處理，比如 Sparquet，Hive，Flink 等等。經過加工和處理之後，會把處理後的結構化資料存儲在 Doris 中。Doris 本身是一個擁有完備 MPP 架構的資料倉庫。它可以直接對外提供報表分析、資訊查詢，以及資料庫分析等功能。同時它也可以作為一個 SQL 引擎，對外部的資料源進行查詢加速，包括 Hive ，Iceberg，Hudi 等等。也支援 MySQL，ElectricSearch 等外部資料源。同時，也提供了官方的 Flink Connector 和 Spark Connector。使用者可以通過這兩類 Connector，友善的去把 Doris 存儲中的資料和其它資料源的資料進行聯邦分析查詢，保證 Doris 最終的資料不會成為資料孤島。
這就是 Doris 在整個資料流中的定位，以及它是如何在企業資料流中發揮價值的。
--
02
Doris 資料湖分析内幕
接下來進入本文的重點，也就是 Doris 的資料湖分析内幕。
先來講一下什麼叫資料湖分析。其實 Doris 本身是一套完備的資料庫管理系統，包括查詢層和存儲層。在我們正常使用 Doris 的時候，隻需要把資料灌入到 Doris 中來，就可以在 Doris 内部對資料進行管理、存儲和查詢，我們叫做内置資料存儲（Internal Storage）或者說自管理的資料存儲（Self-Managed Storage）。在實際業務應用中，還會有大量資料存儲在外部資料源中的，比如可能有很多曆史資料，本身已經存在 Hive 系統中，或者是最近比較火的 Iceberg、Hudi 等資料湖格式中。如果使用者要把這些系統中的資料通過導入操作導入到 Doris 中來，代價是非常大的，因為資料量可能是 TB 甚至 PB 級别的。把這些資料進行一次清理加工的計算量和存儲開銷都是非常大的。是以很多使用者希望借助 Doris 的高速查詢能力，直接對外部資料源的資料進行分析。這也是 Doris 資料湖分析、或者外部資料源加速分析的一個初衷。
在早期的 1.0.0 版本中，Doris 就已經支援了對外部資料源的一些通路，比如對 Hive 外表的建立，對 Iceberg 外表的建立，或者對 MySQL 外表的建立。但是在老版本中建立這些外部資料源的映射時，隻能通過表級别的映射。這會帶來一個問題，比如這些表可能多達幾十上百張，甚至是上千張，如果采用這種方式，使用者需要對每一張表通過 create external table 這種方式去單獨地建立一個映射關系，這是一個很費時費力的工作。
是以在 Doris 的新版本中，通過引入 Catalog 的概念，簡化這一操作，讓使用者可以通過一行指令就能快速開始對外部資料進行分析。
Catalog 是标準 SQL 定義中的三個層級之一，就是 Catalog-Database-Table。我們将 Catalog 分為兩大類，一類是 Internal Catalog，另一類是 External Catalog。其中 Internal Catalog 是管理 Doris 的内部表。External Catalog 可以直接映射到一個資料源。比如一個 Hive 叢集、 ES 叢集、 MySQL 資料庫等等。通過資料源映射，Doris 内部會自動的把外部的 database 和 table 進行同步。是以在上圖中可以看到，中間這一層， database 和 table 用虛線來表示。也就是當建立完 Catalog 後，Doris 内部會自動把資料源中的 database，table 的資訊全都同步過來，進而省去了單獨手動映射每一張表的繁瑣工作，快速接入到這些外部資料源進行查詢。通過這種方式，可以極大的降低對外部資料源接入的門檻兒。
下圖是新版本中架構變動的 Metadata 全景圖：
External Catalog 現已支援幾種主要資料源和中繼資料中心。第一類就是 Hive Metastore 或者是相容 Hive Metastore 的中繼資料中心。比如雲上的 AWS Glue、阿裡雲的 Data Lake Formation 等。通過接入中繼資料中心，可以友善地把中繼資料中心中記錄的庫表資訊自動同步到 Doris 裡來。支援的表格式包括 Hive 的表、Iceberg 的表、Hudi 的表等等。這些表都可以進行統一的管理。第二類是 JDBC Catalog，理論上可以接入任何支援 jdbc 連接配接協定的資料庫。目前隻支援了 MySQL，接下來很快會支援 PostgreSQL、SQL Server、Oracle、ClickHouse 等等這些支援 jdbc 連接配接協定的資料庫。第三類是 ElasticSearch。Doris 在之前版本中就對 ES 有非常好的支援，可以為 ES 提供一個完備的分布式的 查SQL 詢能力。在新版本中，通過引入 ES catalog，更友善的去對接 ES 資料源。這樣使用者不用再一張一張表的去做中繼資料的映射，可以很快地幫助使用者去分析這些外部資料源的資料。同時，在代碼設計層面，Doris 也對整個 Catalog 做了一層抽象，包括 Catalog 層級、database 層級和 table 層級都做了接口的抽象。開發者可以通過這些接口，擴充自己的資料源。
第二個架構變動就是資料通路上的功能統一。
Doris 是一個極速 OLAP 資料庫，擁有性能優異的分布式查詢引擎。我們希望對外部資料源的查詢加速，能夠充分利用現有查詢引擎的優勢。在查詢引擎中，上層計算節點的算子的優化，比如 join 的優化，聚合節點的優化、scan 排程的優化等等，與資料源本身是無關的，它本身是查詢層的一些優化。是以我們對查詢層進行了代碼結構的重構，把一些公共部分提取出來，這些公共部分可以幫助我們去利用 Doris 完備的極速向量化引擎、基于代價的查詢優化器、以及各類算子的優化。同時，也重構了底層的 scan 任務的排程，如 scan 的并發度、CPU 時間分片的排程等，以確定這些功能能夠被内部資料和外部資料源共同使用。
在做完這些架構調整之後，對于外表查詢或者資料湖上的資料查詢，開發者隻需要關注資料源自身的一些通路特性。比如對于 Hive 表查詢，我們可以實作一個 FileScanNode 的，專注于對遠端存儲上的檔案的掃描優化。對于特定的資料格式，我們隻需要實作對于特定資料格式的 format reader。如此一來，開發者在接入一個資料源時，隻需專注于處理資料，掃描相關的一些優化和資料源通路的一些特性的優化，而不需要去關心整個查詢層的優化措施。通過這個架構調整，對一個新的資料源接入，隻需要大概一周的時間就可以完成，同時可以利用到所有已經存在的優化能力去加速資料源的查詢。
接下來看一下資料源通路的整體流程：
熟悉 Doris 的同學都知道，Doris 分為兩個部分：FE 節點和 BE 節點。FE 節點是 java 寫的，主要負責使用者請求的接入，中繼資料管理，查詢查詢計劃生成；BE 節點是 C++ 寫的，負責資料的存儲和查詢計劃的執行，它是一個高性能的分布式查詢執行程序。
以 Hive 為例，當我們通過 Doris 去查詢一張 Hive 表的時候，首先使用者請求進入到 FE，第一步是通過 Hive Metastore 去擷取 table 的 schema，接着擷取 partition。擷取到 partition 以後，FE 會根據 SQL 中的分區的值的謂詞條件對分區進行裁剪，得到最終的分區清單。拿到分區清單以後，再去通路 Hive Metastore 去擷取分區所對應的檔案清單。擷取到檔案清單以後，第五步就是在 FE 中對檔案掃描任務進行拆分，均勻分布到所有的 BE 節點上，保證一個查詢任務，可以充分的利用整個叢集的計算資源進行資料查詢。任務配置設定完以後會下發給 BE，BE 的邏輯就比較簡單，隻需要對指定檔案進行掃描、過濾和讀取。第七步，它會直接去通路 HDFS 或者 S3 上的資料，進行資料的讀取。接下來上層會有一些中文節點，agg 節點，join 節點等等的一些查詢執行。最終把它的結果傳回給使用者。這就是 Doris 通過 Hive Metastore 去查詢 Hive 外表的整體流程。
接下來介紹一下在整個流程中 Doris 有哪些優化。
第一點優化就是對中繼資料和資料通路的優化。一些表的中繼資料資訊是非常大的，比如一張 Hive 表可能有十萬個分區，如果把十萬個分區資訊在一開始的時候就全都同步到的 FE 節點，對 FE 節點記憶體壓力會非常大。因為 Doris 中所有的中繼資料都是在記憶體存放的，如果把這些外部資料源的資訊全都一次性同步過來的話，FE 的中繼資料壓力會非常大。是以我們在 FE 上做了多種類型的 cache。
第一種是 schema cache，對于外表的所有列資訊的 cache。這些 cache 全都是按需加載的。比如我們有一千張表，隻需要通路到其中的一張表的時候，我們才會把這張表的 schema 緩存到 FE 的緩存集中。這樣可以保證記憶體中隻有需要用到的 schema。
第二個是 partition value cache。當查詢一個外表時需要對分區進行裁剪。分區裁剪隻需使用分區值。是以我們單獨實作了一個 partition value cache 專門去緩存分區值的資訊，用于分區裁剪。分區值的記憶體空間占用是非常少的。通過分區值裁剪，可以得到最終需要通路的分區清單。
當得到分區清單以後，就進入到第三步，即需要通路 partition cache 去拿到完整的分區資訊。拿到這些資訊以後，我們就來到第四步，就是 file cache。一個分區下面會有多個檔案。我們拿到了分區的位置資訊以後，就可以通過 file cache，去擷取這個分區下的所有的檔案的資訊（檔案路徑）。拿到檔案路徑後，我們在 FE 中會做任務的拆分。最終會把這些檔案清單拆分以後，發給 BE。
BE 節點負責檔案的讀取和通路。這裡我們也實作了兩個大類的緩存的優化。
第一個是資料預讀（prefetch buffer），在通路 HDFS 和 S3 時，本質是一個 RPC 請求。第一個優化點就是如何能夠盡量減少 RPC 的開銷。一次 RPC 開銷本身的 overhead 很重。是以我們增加了一個預取緩存，把多個小的 Remote IO 請求合并成一個大的 IO 請求，把原先可能幾個位元組的請求，合并成 4KB 到 1MB 的資料請求一次性讀取過來，在本地記憶體中形成一個緩存。後續的 IO 可以直接在記憶體緩存中去擷取資料，極大的減少 remote IO 的次數。
第二個是檔案塊級别的緩存（file block cache）。在通路 HDFS 或者 S3 上的資料檔案時，可能隻需通路其中的一小部分。比如一個列存格式的 parquet 檔案，如果隻需要通路其中的三列資料，那麼隻會讀取整個檔案的一部分，Doris 會在第一次檔案讀取時，将讀取的檔案塊緩存到本地磁盤。緩存檔案的檔案名是檔案的路徑，加上讀取偏移量的組合辨別。之後如果有相同的檔案通路，會先檢視本地是否已經有緩存的資料檔案。如果有，則直接去讀本地檔案，減少通路遠端資料的開銷。通過 file block cache，可以極大地提升一些熱資料的通路效率。
第二個優化點就是 native 的 file format reader。以 parquet 為例，在老版本的 Doris，是通過 Apache arrow 庫内置的 parquet reader 完成讀取的。這個 Reader 的實作會有一些額外的開銷。比如它會多一層記憶體格式的轉換。因為它在讀取的時候，首先需要把遠端的檔案轉換成内部的 arrow 的格式。然後再把 arrow 的格式轉換成 Doris 的内部記憶體格式。第二是 Apache arrow 内置的 parquet reader 對一些新的 Parquet 特性是不支援的，比如不支援 page index、不支援 parquet 的 bloom filter 的讀取、不支援這種更精細的字典編碼的優化等等。
基于以上考慮，我們在 Doris 内部實作了一個 native 的 C++ 的 parquet reader。首先是能直接轉換内部存儲格式，對于讀取到的資料，直接轉為内部記憶體格式，減少一次記憶體格式的拷貝和轉換開銷；第二點，我們能夠利用 bloom filter 進行更精确的資料過濾。使用者寫資料的時候，對某一列使用的 bloom filter，可以利用 bloom filter 去對資料進行過濾。其次我們也支援了基于字典編碼的謂詞過濾，可以把謂詞下推到 parquet reader 中。把謂詞中的，比如 “a=‘北京’” 這樣的一個條件改成一個對應字典編碼的值。比如 “a=‘100’”，後續用 ‘100’ 在檔案内部進行資料過濾。因為整數型的過濾效果是比字元串的過濾效果要好很多的。過濾完了以後，在最終傳回結果的時候，我們再把字典編碼值轉換成真正的資料的值。這樣來達到加速的效果。
最後一點也是非常重要的一點，就是在 Parquet Reader 上支援了延遲物化。延遲物化是通路遠端存儲的時候，減少 IO 的一個非常重要的特性。尤其是在帶謂詞條件的寬表查詢上。簡單來說，Doris 會優先讀取帶謂詞條件的列。讀取完這些列以後，我們先對這些列進行過濾得到最終過濾後的行号集合，再去讀取剩餘的其他的列。這樣就能保證剩餘其他列都是隻會去讀取過濾後的資料。進而極大地減少從遠端去讀取資料的 IO 開銷。
第三點就是彈性計算節點（compute node）。當我們去通路外部資料源的時候，Doris 本身是不會去存儲這些資料的，是以不需要 BE 節點本身的存儲能力。一旦我們不再需要 BE 的存儲能力，它就變成了一個無狀态的節點。當一個節點是有狀态的，删除節點或者添加節點時都要考慮到資料如何安全下線，資料如何遷移，重新 rebalance。而一個無狀态節點可以非常友善的進行彈性擴縮容。是以我們在新的版本中給 BE 節點增加了兩種類型：
第一種類型是 mix node，mix 就是标準的 BE 類型。既支援資料計算，也支援本地的檔案存儲；第二種類型叫 compute node，即計算節點，計算節點可以很友善的進行彈性伸縮。比如可以很快速地在新機器或者雲上建立一些新的 compute node。這些 compute node 可以分擔通路遠端存儲的一些計算的開銷。通過這種無狀态的 BE 節點，可以快速去承接外部資料源的計算負載。來達到彈性伸縮的目的。
下圖是我們在版本釋出之初做的一個測試。
可以看到在同一資源規格下，我們去查詢 Iceberg TPCH 100G 的資料集。相比 Trino，Doris 有三到五倍的性能提升。
最後看一下目前 Doris 對資料湖的一些功能的支援程度：
在 1.2.0 版本中，Doris 支援三個主流的外部資料服務或者資料倉庫。第一個就是 Hive，支援 Managed table 和 External table。支援 parquet、orc 和 text 格式的讀取。Iceberg 完整的支援 V1 Format，V2 支援了 position delete。Hudi 暫時隻支援 MOR 的表，包括 COW Snapshot Query 以及 MOR Read Optimized Query。
--
03
Doris 社群發展以及後期開發規劃
最後介紹一下我們在資料湖分析這塊的一些規劃。
第一個規劃就是增量資料通路。增量資料也是 Iceberg，Hudi 這類新興的資料庫系統所提供的核心價值之一。它可以應用于 A/B Test，或者是用其 Time Travel 的能力、CDC 的能力來進行增量資料的處理和通路。Doris 在後續也要對這一類的功能進行支援。其次就是基于增量資料的視圖查詢。比如我們會通過基于增量資料的多表的物化視圖的更新，或者邏輯視圖的權限控制等等，來幫助使用者很好的去管理資料湖上的資料，并且能夠對資料進行很精細的通路。
第二點就是資料湖寫入能力。剛才我們介紹這些功能時候，其實都是在介紹如何去通路和讀取這些外部資料源的能力。如果使用者想完整的通路管理這些資料源的話，必須在外部對接例如 Spark 這些系統進行資料寫入。是以我們後續希望在 Doris 内部提供統一的操作入口，來消除使用者操作資料的割裂感，來保證對資料庫的寫入操作和查詢操作，都可以統一在 Doris 中完成。
最後一點是深入內建 Iceberg 的能力。希望以 Doris 本身作為 Iceberg 的中繼資料中心來提供托管 Iceberg 的能力，提升自身對于資料湖，或者說是對結構化、半結構化大規模資料的管理能力。
以上就是對 Doris 資料湖的一些介紹。最後簡單介紹一下 Doris 社群現狀和未來規劃。
Apache Doris 是國内最活躍的開源社群之一。
累計貢獻者人數已經超過了四百位，平均每月的活躍使用者貢獻者人數也超過了一百人。可以看到我們每個月所送出的 commit 量和 push 量都是相當可觀的，發展也是非常快速的。也歡迎對分布式資料庫或者對 MPP、OLAP 資料庫感興趣的同學加入到社群中來，我們可以一起去完善這樣的一個資料庫系統。
下圖是 Doris 在 2022 年底到 2023 年初的一個大緻規劃：
首先我們在 2022 年的 Q4 季度，釋出了 1.2.0 版本。在該版本中，實作了多中繼資料目錄，其中就包括資料分析這部分的一些能力；其次我們還加入了半結構化資料的一些支援，包括 Array 和 Binary Json 格式的支援；我們也支援了新的 unique 模型，可以幫助使用者在可變更的或者可更新的資料中依然能進行快速的資料通路；其次我們還支援了包括 JDBC 的 External table，以及 Java UDF 等一些新的特性。歡迎大家到官網去體驗。
在 2023 年 Q1，我們會釋出新的優化器的 preview 版本。新的優化器完全重構了現有的優化器的架構。實作了一個可插拔可擴充 CPU 的查詢優化器。可以幫助使用者解決複雜 SQL 的擷取 best plan 的問題；其次我們也會釋出 Pipeline 執行引擎的 preview 版本，使 Doris 能夠更細力度的去規劃 BE 節點的執行資源。保證 BE 節點可以充分利用我們的單機多核的特性，并且使用者不再需要手動去設定查詢并發度等等。比如在閑時，利用更多的 CPU 資源；在忙時，可以進行大小查詢，這種動态的資源隔離。前文提到的 compute node，在 Q1 季度會釋出完整的 release 版本。還有 Vertical conduction，解決大寬表場景下的背景資料merge的記憶體開銷問題。
在 2023 年 Q2，會釋出 2.0.0 版本。除了剛才提到的優化器和 Pipeline 達到 GA 狀态以外，還會有一些新的特性，比如半結構化資料的一些查詢，存算分離的一些架構演進等等。希望在未來的一年能夠繼續給我們的使用者提供更便捷、統一的分析型資料庫。
如果大家想體驗這種極速、易用、統一的分析型資料庫，歡迎大家來關注 Apache Doris。下面是我們的官網和 Github 位址，還可以掃碼加入我們的使用者群，群中的任何問題都會有專人去解答和處理。
--
04
問答環節
Q1：Doris 通過連接配接外部的 Hive，能不能自動監控 Hive 的表結構或資料的變化？
A1：現在提供了手動的 refresh，可以手動 refresh Catalog 級别，DB 級别，table 級别和 partition 級别。最新的 1.2.2 版本，也支援通過 Hive Metastore 的 Hook 機制來自動監聽 Hive 的中繼資料變動，達到一個增量的中繼資料同步的效果。
Q2：Doris 和 Flink 的對接方式推薦哪種？
A2：建議用 Doris 官方提供的 Flink connector。在我們的官方庫上可以找到對應的代碼庫下載下傳連結和釋出版本。
Q3：Doris 讀對象存儲資料湖對性能和時延的影響會怎麼樣？
A3：在之前也講了 Doris 的一些優化點，包括它的 read，消除小的 IO，本地的 file block cache 等等。做這些功能的出發點都是為了盡量避免通路遠端存儲，以及避免大量的小 IO 遠端通路。我們做這些優化的初衷，都是為了能夠盡量的去把大塊資料一次性的讀過來，然後在本地進行處理。據我們的測試情況來看，經過我們的這些優化，Doris 對熱資料的通路，幾乎是可以達到和本地表一樣的通路性能。
Q4：Doris 怎麼處理高并發的請求？
A4：關于高并發請求，可以分為兩部分，第一部分要解決單一查詢所占用的資源開銷的問題。比如一個查詢需要發送到一百台機器去查詢，它的扇出特别大，并發是肯定很低的。是以我們會通過分區裁剪，分桶裁剪等，盡量把一個查詢限制在某一台機器上，甚至是某一個資料分片上。這樣單個查詢的資源開銷足夠的小，那整個叢集的整體的并發支援就會很高。第二，如果是在資料湖上的這種高頻發查詢的話，其實本質上還是會回歸到關于遠端存儲 IO 的一些問題上來。也就是盡量去減少小 IO 的遠端查詢或者通過緩存來解決熱資料查詢，因為 remote IO 的 overhead 是沒法徹底根除的。它跟遠端存儲的網絡，通路的特性都有關系。是以說本質上還是要通過一些 cache 和 buffer 特性來去消除這些遠端存儲的 IO 的次數以達到一個高并發的效果。
今天的分享就到這裡，謝謝大家。