業務背景
貨拉拉成立于 2013 年，成長于粵港澳大灣區，是一家從事同城、跨城貨運、企業版物流服務、搬家、汽車銷售及車后市場服務的互聯網物流公司。截至 2022 年 4 月，貨拉拉的業務范圍已經覆蓋了國內 352 座城市，月活司機達到 58 萬，月活用戶達到 760 萬，包含 8 條以上的業務線。
貨拉拉大數據體系為支撐公司業務，現在已經成立三個 IDC 集群、擁有上千臺規模的機器數量，存儲量達到了 20PB、日均任務數達到了 20k 以上，并且還處在快速增長的過程中。
大數據體系
貨拉拉大數據體系從下往上分為 5 層，最下面的是基礎層和接入層，這兩層主要會提供基礎數據的存儲、計算以及集群的管理功能。在基礎層和接入層之上是平臺層和數倉。在平臺層之中包含了數據研發平臺和數據治理平臺，基于平臺層的能力和數據倉庫的數據體系，在這之上面包含了含有業務屬性的服務層和應用層。整個體系自下而上相互支持，實現支持業務和賦能業務的能力。
圖1.1 貨拉拉大數據體系
數據處理流
貨拉拉典型的數據處理流，可以分成數據集成、采集、數據的存儲計算、數據服務四部分，同時也包含了實時、離線以及在線三大業務場景。
圖1.2 貨拉拉大數據數據流
在數據采集階段會存在實時采集和離線采集兩條路線。
- 實時采集比較典型的場景為用戶端上埋點會直接同步到大數據平臺做存儲，供后續的在線和離線計算使用。
- 離線的數據主要是來自于業務方的數據庫，會通過天或者是小時定期采集到大數據存儲中，以供后續使用。
中間是數據的存儲和計算階段。在離線場景中會通過對數據 ETL 之后轉換為構造數倉的分層體系。實時比較典型的場景為數據在經過 Flink 的處理后會直接落在線存儲系統，類似于 HBase 和 OLAP 等等，為后續的業務系統提供數據服務。
OLAP 演進概覽
貨拉拉從 2021 年開始進行 OLAP 的技術研究，截至目前已經經歷 3 個階段：
- 2021 年上半年為貨拉拉的 OLAP1.0 階段，這個階段我們主要是支持公司的羅盤業務，我們引入的是能夠提供較好的單表依據和查詢能力的 Apache Druid 引擎。
- 2021 年下半年為貨拉拉的 OLAP2.0 階段，我們支持了智能定位工具，這個階段引入了夠提供單表明細查詢，并且還有較高的壓縮率 ClickHouse。
- 今年為貨拉拉的 OLAP3.0 階段，伴隨著公司業務需求的不斷增多，我們也會需要用到多數據源的關聯分析。基于此，由于 Apache Doris 具備大表關聯分析的能力，我們引入了 Apache Doris 引擎。
圖2.1 貨拉拉OLAP體系演進過程
OLAP1.0 孕育期
業務需求分析
先看下沒有引入 OLAP 之前的業務數據流：
圖3.1 OLAP1.0業務場景
根據該圖可以看到業務的數據通過實時和離線處理之后會落在 MySQL，MySQL 之中儲存了維度聚合之后的結果數據，這也意味著會在 Flink 之中做大量的聚合分析，根據業務需要的相應維度所做的一系列組合都是在Flink之中做實時聚合，最后將結果儲存到 MySQL。
存在的問題：
- 存在存儲瓶頸，類似于 Kylin 之中的維度爆炸的問題。
- 開發成本、高效率低。當業務側需要新增維度的時候會需要對 Flink 中的所有作業都做一定的修改，然后再重新上線。
- 無法支持部分聚合需求。
對于存在的這些問題，我們經過分析之后，總結出了 3 個背后存在的需求點：
- 業務側希望能夠橫向擴容，解決存儲瓶頸。
- 希望能夠自由組合維度做分析，提升業務側開發效率。
- 希望能夠支持任意維度實現跨度的分析。
解決方案
根據業務需求，并通過調研，我們決定使用 OLAP 引擎來支持業務需求。那我們如何選擇一款 OLAP 引擎，并把它穩定的應用到生產之中呢？
我們總結了如下的 4 個步驟作為解決思路：
圖3.2 OLAP 1.0 解決思路
技術調研
技術調研階段，我們對比了 Durid、ClickHouse、Kylin、Presto 和 Doris 等等引擎。結合我們上述的 3 個業務需求，最終我們選擇了 Druid 引擎。
原因是 Druid 除了能夠滿足我們的業務需求之外，還有一個比較重要的影響因素是 Druid 引擎是純 Java 開發，與我們的技術棧比較吻合，可控性更高。
圖3.3 OLAP1.0技術調研
POC 階段
POC 過程中，從以下 3 個步驟著手：
- 功能驗證。在功能驗證中，我們會收集業務側的 SQL，之后提取 SQL Pattern，然后再根據 Druid 引擎的 Rollup 語義做 SQL 的改寫，涉及到大量 UDF 的改寫、Rollup 語義兼容以及 Count Distinct 語義兼容等等。
- 性能驗證。我們會直接采用業務真實的數據和業務真實的 SQL 來執行。驗證過程中我們會將 Cache 關閉，分別統計 P75、P90、P99 的查詢耗時。在這過程中，我們會發現有部分查詢的性能沒有達到要求，之后我們會做性能分析。Druid 引擎本身沒有比較完善的性能分析工具，不能夠很好的打印出它的執行計劃以及各個算子的耗時，所以我們采用了第三方的 Arthas 火焰圖進行分析。定位了相應的算子后，最終我們通過優化我們建表導數的邏輯以及索引構建的邏輯，并主要通過調整 Segment 大小的同時加入物化視圖的方法，進行一些參數的調整以此來優化性能。
- 準確性驗證。我們將業務真實數據同時寫 Hive 表和 Druid，之后跑 Hive SQL和 Druid SQL，來進行數據質量的校對。在這個過程中我們會發現例如 StringLast 函數等一些函數會在特定的場景下出現計算值不穩定的問題。
圖3.4 OLAP1.0 POC 驗證
穩定性保障
當 POC 驗證完成之后，接下來我們會進行穩定性的保障。我們將穩定性保障分為事前、事中、事后 3 個階段：
圖3.5 OLAP1.0 穩定性保障
上線階段
當穩定性保障建立完成之后就會進入到上線階段。上線過程我們同樣分成了 3 個階段：
- OLAP測試階段。在這個階段中，業務的數據會接入到 Druid 之中，但是業務的真實查詢還是通過原來的 MySQL 庫。這個階段主要會驗證 Druid 引擎的數據質量和 Druid 集群的穩定性。
- 上線觀察階段。在這個階段，業務的查詢會切回到 Druid。同時舊的 MySQL 鏈路還沒有下線，業務側能夠隨時切回 MySQL 鏈路。
- OLAP運行穩定階段。我們會把 MySQL 舊的鏈路下線，做資源的回收。
圖3.6 OLAP1.0 上生產
問題總結
下面總結了 1.0 階段時遇到的問題：
- 數據導入部分中，實時數據亂序為典型問題。
- 在數據準確性驗證階段發現 StringLast 的函數值不穩定。
- Durid 沒有一個高效的精準去重的函數。
圖3.7 OLAP1.0 問題總結
OLAP2.0 完善期
業務需求分析
在 OLAP2.0 階段主要有以下 4 個業務需求：
圖4.1 OLAP2.0 業務需求分析
下圖是簡單的業務工具的截圖，從圖中可以看到，OLAP2.0 需要能夠支持匯總與明細，同時基于這些能力能夠做一個快速的問題定位。
圖4.2 OLAP2.0 業務需求分析驟去實現。
解決方案
圖4.3 OLAP2.0 技術調研
OLAP2.0 我們引入了 CliclkHouse。ClickHouse 能夠比較好地支持復雜的數據類型，同時因為業務側是埋點數據，所以對于實時導入語義要求并沒有那么高。
沒有采用 Druid 主要是有 2 個原因：
- Druid 對于復雜的數據結構支持度并不是很好。
- Druid 雖然能夠支持明細查詢，但是 Druid 的明細查詢和聚合查詢得分成不同的表，這樣就會額外的引入一系列的存儲成本。
剩下的部分就是 POC 、上生產的步驟，這兩個步驟和 OLAP1.0 階段比較類似，所以在這里就不過多展開介紹。
OLAP3.0 成熟期
業務需求分析
2022 年隨著公司業務的發展，更多的產品線對于多數據源關聯場景下的在線分析需求也會變得越來越迫切。比如說 AB 實驗場景與實時數倉場景，這兩個場景對于多表關聯需求，尤其是大表的多表關聯需求也變得越來越迫切。
圖5.1 OLAP3.0 需求分析
舉一個 AB 實驗的例子。從下圖可以看到，例子中是需要把 AB 實驗的一個數據和后面相應的司機與用戶的埋點數據關聯到一起并做分析。在這種情況下，我們就會發現之前的兩種工具都會存在一系列的弊端。
圖5.2 OLAP3.0 需求分析
解決方案
技術調研
在技術調研階段我們觀察了 Druid 和 ClickHouse。Druid 引擎可以支持一些維表的簡單 Join，ClickHouse 則能夠支持 Broadcast 這種基于內存的 Join，但是對于大數據量千萬級甚至億級的一些表的 Join 而言，ClickHouse 的性能表現不是很好。
圖5.3 OLAP3.0 技術調研
接下來我們對 Doris 進行了調研，我們發現 Doris 是能夠支持小表的 Join，對大表的話也同樣能夠支持基于 Shuffle 的 Join，對于復雜數據類型（Array、JSon）的支持，經過跟 Apache Doris 社區溝通，預計將在 2022 年 7 月份的新版本中發布。通過在多個維度和需求滿足度上進行對比，我們最終選擇了 Apache Doris，也是因為 Apache Doris 的 SQL 支持度非常的完善。
圖5.4 OLAP3.0 技術調研
POC 階段
我們除了引用業務真實的數據和場景做驗證以外，還引入了 TPC-DS 的數據集做了驗證。
在多表關聯的場景下對單天數據進行查詢，對 5 億左右的數據量進行 Join，TP75 大概是 9 秒左右。在數據質量階段我們也是把 TPC- DS 的數據集以及業務真實數據集，分別在 Hive 和 Doris 里面做了雙跑驗證，發現兩者都是能夠完全對得上的。
圖5.5 OLAP3.0 POC
穩定性保障
與之前一樣依然是從事前的容量評估和壓測、事中的監控和定位來進行。
圖5.6 OLAP3.0 穩定性測試
下面是我們的監控圖，主要是關于 Compaction 相關的一些監控，感興趣的同學可以看看。（文末 QA 環節有部分講解）
圖5.7 OLAP3.0 穩定性監控
問題總結
第一個問題是查詢性能的優化。
業務側的需求為 7 天的查詢 RT 需要在 5 秒內完成，在優化前，我們發現 7 天的查詢 RT 是在 30 秒左右。對于這個問題，我們的主要優化策略是把小表 Join 大表改成了大表 Join 小表，主要原理是因為 Doris 默認會使用右表的數據去構建一個 Hashtable。
還有類似下圖中的情況：union all 是在子查詢中，然后再和外層的另外一張大表做 Join 的查詢方式。這種查詢方式沒有用到 Runtime Filter 的特性，因此我們將 union all 提到子查詢外，這樣就能夠用到 Runtime Filter，這應該是由于這里的條件下沒有推下去所導致的。同時運行時采用的 Bloom Filter 是可以將 HashKey 條件下推到大表 Scan 階段做過濾。在經過對這兩者優化之后便能夠滿足我們的查詢性能需求了。
圖5.8 OLAP3.0 問題1
第二個問題是 UnhealthyTablet 不下降，并且在查詢階段會出現 -230 的報錯。
這個問題的場景是我們在沒有停 FIink 寫任務的時候，對 BE 機器交替重啟，重啟完會出現很多 UnhealthyTablet。經過我們后續的分析發現，其中一個原因主要是在 Coordinator BE 在做二階段提交的時候比較巧合，Coordinator BE 的二階段提交 Commit 后，也就是大部分的副本是已經 Commit 后且在 Publish 前，在這短短的時間范圍內 BE 機器被重啟，這也導致會出現 Tablet 狀態不一致的情況。同時由于我們當時把參數調整的過大，導致了 Compaction 壓力過大。
最后的解決辦法：與 Aapache Doris 社區的同學經過互助排查，引入了社區 1.1.0的 Patch，同時對相應的數據做了恢復。
圖5.9 OLAP3.0 問題2
參數優化
- 打開 Profile。Doris 對于查詢的性能分析具有非常好的 Profile 文件，這一點是非常贊的！我們可以看到各個算子在每一個階段查詢耗時以及數據處理量，這方面相比于 Druid 來說是非常便捷的！
- 調大單個查詢的內存限制，同時把 BE 上的執行個數由 1 個調整成為 8 個，并且增加了 Compaction 在單個磁盤下的數據量。對于 Stream Load，我們把 Json 格式的最大的內存由 100 兆調整成為 150 兆，增大了 Rowset 內 Segment 的數量，并且開啟了 SQL 級和 Partition 級的緩存。
圖5.10 OLAP3.0 參數優化
數據流
下圖是使用 Doris 之后的數據流圖：
圖5.11 OLAP3.0 數據流
數據流中，我們在 Flink 中做的事情已經很少了，經過數據簡單的 ETL 后就可以把數據直接灌入到 Doris。經過 Doris 一系列的聚合計算、union 計算以及多表關聯計算之后，業務側就可以直接查詢 Doris 來獲取相關數據。
總結與思考
總結：我們 OLAP 的引進主要還是從業務需求的角度出發來匹配合適的引擎，為業務精細化運維提供技術支持。在這之后，我們也思考了一套較為完善的上線流程及穩定性保證方案，為業務的平穩運行提供能力保障。
思考：我們認為很難有單個引擎能夠富含各種場景。因此在技術選型時，需要針對于需求特點和引擎特點進行合理選擇。
后續規劃
我們希望可以向 OLAP 平臺化發展，通過實現自助化建模的同時在這方面做一些多引擎的路由，使其能夠支持各類聚合、明細以及關聯等場景。
圖6.1 后續規劃 OLAP 平臺化
除 OLAP 平臺化之外，后續我們的引擎演進計劃從高效、穩定和內核演進三部分來進行。
圖6.2 后續規劃 引擎演進
穩定性方面：對 Doris 還要繼續深入內核理解，提供一定的二次開發。另外 Doris 社區的相關原理以及代碼級別的教程數量十分豐富，這也間接性降低了我們深入 Doris 原理的難度。
內核演進方面：我們發現 Doris 基本能夠覆蓋 Druid 所有場景，因此后續計劃以 Doris 引擎為主，Clickhous 引擎為輔，逐漸將 Druid 的相關業務向 Doris 遷移。
Q&A 環節
Q：剛才講到了后續要從 Druid 引擎遷移到 Doris，要實現遷移的成本有多大呢？
A：遷移成本方面和我們之前的成本是一樣的。我們上線的時候也會采用以下方式：先把業務的數據同時往 Druid 和 Doris 之中寫，寫完之后的業務遷移會涉及一些 SQL 改造。因為 Doris 更加接近 MySQL 的協議，比起 Druid SQL 會更加便捷，所以這部分的遷移成本不是很大。
Q：剛才介紹的第二個場景之中的監控圖都看了哪些指標呢？
A:關于監控圖，我們會比較關注 Doris 的數據導入。而在數據導入部分，我們最關注的就是 Compaction 的效率，是否有 Compaction 的堆積。我們現在還是采用的默認參數，也就是 Compaction 的分數就代表它的版本號，所以我們監控的更多的是它的版本。對于這方面的監控，社區也已經有了比較完善的相應技術方案，我們也是參考了社區的技術方案來進行了監控的指標搭建。
Q：從指標上看，Doris 的實時服務在線查詢性能怎么樣？在數據導入情況下性能損耗可以從這些指標上看出來嗎？
A：實時導入方面主要是從 Compaction 的效率來看。結合到我們這邊的業務場景，最多的一張表，單表一天也有 6 億到 10 億的數據量的導入，也是一張埋點。另外關于峰值，它的 QPS 也是能達到千到萬的，所以導入這一塊壓力不是很大。
Q：SQL 緩存和分區緩存實際效果怎么樣？
A：SQL 緩存方面效果還好，對于很多離線場景，尤其是首頁這種查詢的數據量而言。比如以昨天或者是過去一個小時之前的這種情況來說，SQL 緩存命中率會非常高。分區級緩存方面，我們分區的時間還是設的是小時級，這意味著如果這個查詢里面涉及到的一些分區在一個小時內沒有數據更新的話，那么就會走 SQL 緩存；如果有更新的話就會走分區級緩存。總體來看效果還好，但是我們這邊命中比較多的還是 SQL 級的緩存。
Q：Doris 的查詢導入合并和緩存的 BE 節點的內存一般怎么分配？
A：緩存方面我們分配的不大，還是采用的偏默認的 1G 以內。導入方面我們設計的是 parallel_fragment_exec_instance_num 這個參數，大概在 8G 左右。
Q：可以解釋一下 OLAP3.0 的解決思路嗎？
A：對于 OLAP3.0 方面來說，業務的主要訴求就是大表 Join。除此之外，還有一些類似于導入的進度一致等等。
在大表 Join 方面，我們也對比了很多的引擎。Druid 這方面就是偏維表；Clickhouse這方面還是偏基于內存方面的 Broadcast。正因如此，主要是基于大表 Join 的出發點，我們選擇引入了在 Join 這方面能力更強的 Doris。
Q：Druid、ClickHouse 和 Doris 應該都是近實時的，就是 Near Real-time，他們的寫入不是立刻可見的，是這樣嗎？
A：是這樣的。像 Doris 和 ClickHouse 之前的寫入都是 Flink 直接去寫，我們也沒有完全做到來一條數據就寫一條，都是一個微批次。一個批次最大可以達到 150 兆的數據堆積，寫入一次的時間間隔也是到 10 秒左右，沒有做到完全的實時寫入。
Q：方便透露一下貨拉拉目前 Doris 的集群的使用情況，比如機器的數量和數據量嗎？
A：我們的集群數量還不算很多，10 多臺。
Q：對于 Doris 的運維方面，它的便捷性和 Druid、ClickHouse、Kylin、Presto 這些相比，有很好的擴展性嗎？
A：我們覺得是有的。第一個是在我們 Druid 方面碰到了一個比較大的痛點，就是它的角色特別多，有 6 種角色，所以需要部署的機器會非常多。另外一點是 Druid 的外部依賴也非常多，Druid 依賴于 HDFS、離線導入還需要有 Hadoop 集群。
第二個是 ClickhHouse 方面，我們當時使用的版本對于 Zookeeper 也是有比較大的依賴。另外，ClickHouse 也是偏偽分布式的，有點類似于數據庫的一種分表。Doris 自身就只有 FE、BE，外部依賴會非常少，所以我們從部署的角度同時考慮到 Doris 的橫向擴展方面，Doris 的擴縮容也能夠做到自平衡，所以相比而言 Doris 會更好一些。
Q：在實時特征場景下，分鐘級的數據更新對服務性能要求比較高，可以用 Doris 嗎？能達到 TP99 200 毫秒以下嗎？
A：TP99 能夠否達到200毫秒以下主要和你查詢 SQL 相關。
例如我們這邊的很多涉及到大表 Join 的查詢，涉及的分區數據量大概在 10 億量別，業務側對于查詢性能要求是 5 秒以內，通過 Doris 是可以滿足我們需求的。如果是實時特征這種業務，是否能達到 200 毫秒可能需要經過一輪實際測試才能得到結果。
