一、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 在整個數據流中的定位，以及它是如何在企業數據流中發揮價值的。

二、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。

三、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 狀態以外，還會有一些新的特性，比如半結構化數據的一些查詢，存算分離的一些架構演進等等。希望在未來的一年能夠繼續給我們的用戶提供更便捷、統一的分析型數據庫。

四、問答環節

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 的次數以達到一個高并發的效果。