得物基於 StarRocks 的 OLAP 需求實踐

語言: CN / TW / HK

時間 2022-09-30 17:01:57 得物技術

1. 什麼是 StarRocks

新一代極速全場景MPP資料庫，可以用 StarRocks 來支援多種資料分析場景的極速分析；
架構簡潔，採用了全面向量化引擎，並配備全新設計的 CBO 優化器，查詢速度（尤其是多表關聯查詢）；
很好地支援實時資料分析，並能實現對實時更新資料的高效查詢，還支援現代化物化檢視，以進一步加速查詢；
使用者可以靈活構建包括大寬表、星型模型、雪花模型在內的各類模型；
相容 MySQL 協議，支援標準 SQL 語法，易於對接使用，全系統無外部依賴，高可用，易於運維管理。

2. 系統架構

核心程序：FE（Frontend）、BE（Backend）。

注：所有節點都是有狀態的。

FE（Frontend）負責管理元資料，管理客戶端連線，進行查詢規劃、查詢排程等工作。
Follower
Leader：Follower會通過類Paxos的BDBJE協議選主出一個Leader，所有事務的提交都是由Leader發起，並完成；
Follower：提高查詢併發，同時參與投票，參與選主操作。
Observer：不參與選主操作，只會非同步同步並且回放日誌，主要用於擴充套件叢集的查詢併發能力。
BE（Backend）負責資料儲存以及SQL執行等工作。

3. 儲存架構

在StarRocks裡，一張表的資料會被拆分成多個Tablet，而每個Tablet都會以多副本的形式儲存在BE節點中，如下圖：

Table資料劃分 + Tablet三副本的資料分佈：

StarRocks支援Hash分佈、Range-Hash的組合資料分佈（推薦）。

為了等到更高的效能，強烈建議使用Range-Hash的組合資料分佈，即先分割槽後分桶的方式。

Range分割槽可動態新增和刪減；
Hash分桶一旦確定，不能再進行調整，只有未建立的分割槽才能設定新的分桶數。

分割槽和分桶的選擇是非常關鍵的。在建表時選擇好的分割槽分桶列，可以有效提高叢集整體效能。

以下是針對特殊應用場景下，對分割槽和分桶選擇的一些建議：

資料傾斜：業務方如果確定資料有很大程度的傾斜，那麼建議採用多列組合的方式進行資料分桶，而不是隻單獨採用傾斜度大的列做分桶。
高併發：分割槽和分桶應該儘量覆蓋查詢語句所帶的條件，這樣可以有效減少掃描資料，提高併發。
高吞吐：儘量把資料打散，讓叢集以更高的併發掃描資料，完成相應計算。

‍

3.1 表的儲存

對錶進行儲存時，會對錶進行分割槽和分桶兩層處理，將表的資料分散到多臺機器進行儲存和管理。

分割槽機制：高效過濾，提升查詢效能。

分割槽類似分表，是對一個表按照分割槽鍵進行分割，可以按照時間分割槽，根據資料量按照天/月/年劃分等等。可以利用分割槽裁剪對少數訪問量，也可以根據資料的冷熱程度把資料分到不同介質上。

分桶機制：充分發揮叢集效能，避免熱點問題。

使用分桶鍵Hash以後，把資料均勻的分佈到所有的BE上，不要出現bucket資料傾斜的情況，分桶鍵的選擇原則就是高基數的列或者多個列組合成為一個高基數的列，儘量將資料充分打散。

注：Bucket數量的需要適中，如果希望充分發揮效能可以設定為：BE數量 * CPU core/2，最好tablet控制在1GB左右，tablet太少並行度可能不夠，太多可能遠資料過多，底層scan併發太多效能下降。

Tablet：最小的資料邏輯單元，可以靈活設定平行計算資源。

一張表被切分成了多個Tablet，StarRocks在執行SQL語句時，可以對所有Tablet實現併發處理，從而充分的利用多機、多核提供的計算能力。

表在建立的時候可以指定副本數，多副本夠保證資料儲存的高可靠，以及服務的高可用。

Rowset：每一次的資料變更就會產生一個Rowset。

就是以組列存方式組織的的一些檔案，每次的commit都會產生一個新的版本，每個版本包含哪些Rowset。

每次寫入都會增加一個版本（無論是單條、還是stream load幾個G的檔案）。

Segment：如果一個Rowset資料量比較大，則拆分成多個Segment資料斷落盤。

4. 需求背景

案例一：

業務背景

指標工廠服務主要面向業務人員，通過對業務指標的採集和處理，實時反映產品狀態，為運營提供資料支撐、檢測產品漏洞或服務異常、提供指標異常告警功能等。

業務場景分析

業務指標埋點方式多樣，並不侷限於某種方式，只要符合埋點標識明確、業務引數豐富、資料滿足可解析的基本要求皆可作為資料來源，大致可以分為：SDK、MySQL BinLog、業務日誌、阿里雲ODPS資料分析。

存在的挑戰，各種業務場景眾口難調，歸納資料特徵如下：

需要全量日誌明細；
需要資料可以始終是最新的，即滿足實時更新場景；
需要對資料做層級聚合的，即可能是月、周、日、小時等；
需要可以承載更大的寫入量；
每個業務資料都要靈活地配置資料的儲存時間；
資料來源來源多，報表定製化比較高，有多個數據源合併成一個大寬表的場景、也有多表連線的的需求；
各種監控圖、報表展示、業務實時查詢等，即較高的並非查詢。

引入StarRocks

幸運的是StarRocks有比較豐富的資料模型，覆蓋了上面的所有業務場景的需求，即：明細模型、更新模型、聚合模型、主鍵模型，同時選擇更為靈活的星型模型代替大寬表的方式，即直接使用多表關聯來查詢。

明細模型：

埋點資料經過結構化處理後按明細全量儲存；
該場景對DB在億級資料量下查詢效能有較高的要求；
資料可以通過配置動態分割槽來配置過期策略；
場景使用時從結構化資料選擇個別欄位維度線上聚合查詢。

聚合模型：

埋點資料資料量巨大，且對明細資料不要求溯源，直接做聚合計算，比如計算PV、UV場景；
資料可以通過配置動態分割槽來配置過期策略。

更新模型：

埋點資料狀態會發生變動，且需要實時更新資料，更新資料範圍不會跨度多個分割槽的，比如：訂單、優惠券狀態等；
資料可以通過配置動態分割槽來配置過期策略。

基於以上業務場景的分析，這三種模型可以完美解決資料的問題。

需要實時的資料寫入場景，我也沿用了業內流行的解決方案，即資料採集到 Kafka 之後，使用Flink做實時寫入到StarRocks。StarRocks提供了非常好用的Flink-connector外掛。

小tips：

1. 雖然StarRocks已經很好的優化了寫入效能，當寫入壓力大，仍會出現寫入拒絕，建議可適當增大單次匯入資料量，降低頻率，但同時也會導致資料落庫延遲增加。所以需要做好一定的取捨，做到收益最大化。

2. Flink的sink端不建議配置過大，會引起併發事務過多的報錯，建議每個flink任務source可以配置多些，sink的連線數不能過大。

小結

叢集規模：5FE(8c32GB)、5BE(32c128GB)

目前該方案已支援數百個業務指標的接入，涉及幾十個大盤的指標展示和告警，資料儲存TB級，每日淨增長上百G，總體執行穩定。

案例二：

業務背景

內部系統業務看板，主要服務於全公司員工，提供專案及任務跟蹤等功能。

業務場景分析

分析業務特點：

資料變更頻繁(更新)，變更時間跨度長
查詢時間跨度多
報表需準實時更新
關聯維表查詢多，部門/業務線/資源域等
冷熱資料，最近資料查詢頻繁

歷史架構與痛點

當初資料庫選型時，結合業務特點，使用者需要動態、靈活的增刪記錄自己的任務，因而選擇了JOSN 模型減少了應用程式程式碼和儲存層之間的阻抗，選擇MongoDB作為資料儲存。

伴隨著公司快速快發，當需要報表展示，特別是時間跨度比較大，涉及到多部門、多維度、細粒度等報表展示時，查詢時間在MongoDB需要執行10s甚至更久。

引入StarRocks

調研了StarRocks、ClickHouse兩款都是非常優秀的分析型資料庫，在選型時，分析了業務應用場景，主要集中在單表聚合查詢、多表關聯查詢、實時更新讀寫查詢。維度表更新頻繁，即儲存在MySQL中，StarRocks比較好的支援外表關聯查詢，很大程度上降低了開發難度，最終決定選用StarRocks作為儲存引擎。

改造階段，將原先MongoDB中的一個集合拆分成3張表。使用明細模型，記錄每天的對應人員的任務資訊，按天分割槽，由之前的每人每天一條記錄改為，以事件為單位，每人每天可以多條記錄。

實現頻繁更新的維表，則選擇使用外部表，減少維度資料同步到StarRocks的複雜度。\

小結

改造前，MongoDB查詢，寫法複雜，多次查詢。

db.time_note_new.aggregate(
    [
       {'$unwind': '$depart'},
       {'$match': {
           'depart': {'$in': ['部門id']},
           'workday': {'$gte': 1609430400, '$lt': 1646064000},
           'content.id': {'$in': ['事項id']}, 
           'vacate_state': {'$in': [0, 1]}}
       }, 
       {'$group': { 
           '_id': '$depart', 
           'write_hour': {'$sum': '$write_hour'}, 
           'code_count': {'$sum': '$code_count'}, 
           'all_hour': {'$sum': '$all_hour'}, 
           'count_day_user': {'$sum': {'$cond': [{'$eq': ['$vacate_state', 0]}, 1, 0]}}, 
           'vacate_hour': {'$sum': {'$cond': [{'$eq': ['$vacate_state', 0]}, '$all_hour', 0]}}, 
           'vacate_write_hour': {'$sum': {'$cond': [{'$eq': ['$vacate_state', 0]}, '$write_hour', 0]}}}
           -- ... more field
       }, 
       {'$project': {
           '_id': 1, 
           'write_hour': {'$cond': [{'$eq': ['$count_day_user', 0]}, 0, {'$divide': ['$vacate_write_hour', '$count_day_user']}]}, 
           'count_day_user': 1, 
           'vacate_hour': 1, 
           'vacate_write_hour': 1, 
           'code_count': {'$cond': [{'$eq': ['$count_day_user', 0]}, 0, {'$divide': ['$code_count', '$count_day_user']}]}, 
           'all_hour': {'$cond': [{'$eq': ['$count_day_user', 0]}, 0, {'$divide': ['$vacate_hour', '$count_day_user']}]}}
           -- ... more field
       }
    ]
)

改造後，直接相容SQL，單次聚合。

WITH cont_time as (
    SELECT b.depart_id, a.user_id, a.workday, a.content_id, a.vacate_state
        min(a.content_second)/3600 AS content_hour,
        min(a.write_second)/3600 AS write_hour,
        min(a.all_second)/3600 AS all_hour
    FROM time_note_report AS a
    JOIN user_department AS b ON a.user_id = b.user_id
    -- 更多維表關聯
    WHERE b.depart_id IN (?)  AND a.content_id IN (?) 
      AND a.workday >= '2021-01-01' AND a.workday < '2022-03-31' 
      AND a.vacate_state IN (0, 1)
    GROUP BY b.depart_id, a.user_id, a.workday, a.content_id,a.vacate_state
)
SELECT M.*, N.*
FROM ( 
    SELECT t.depart_id,
         SUM(IF(t.content_id = 14, t.content_hour, 0))   AS content_hour_14,
         SUM(IF(t.content_id = 46, t.content_hour, 0))   AS content_hour_46,
         -- ...more
    FROM cont_time t
    GROUP BY t.depart_id
) M
JOIN ( 
    SELECT depart_id                                  AS join_depart_id,
      SUM(write_hour)                                 AS write_hour,
      SUM(all_hour)                                   AS all_hour
      -- 更多指標
    FROM cont_time
    GROUP BY depart_id
) N ON M.depart_id = N.join_depart_id
ORDER BY depart_id ASC

以查詢報表2021/01/01~2022/03/01之間資料對比：

StarRocks: 1次查詢聚合，可完全通過複雜SQL聚合函式計算，耗時 295ms
Mongodb: 需分2次查詢+計算，共耗時3s+9s=12s

5. 經驗分享

在使用StarRocks時遇到的一些報錯和解決方案（網上資料較少的報錯資訊）：

a.資料匯入Stream Load報錯：“current running txns on db 13003 is 100, larger than limit 100”

原因：超過了每個資料庫中正在執行的匯入作業的最大個數，預設值為100。可以通過調整max_running_txn_num_per_db引數來增加每次匯入作業的個數，最好是通過調整作業提交批次。即攢批，減少併發。

b. FE報錯：“java.io.FileNotFoundException: /proc/net/snmp (Too many open files)”

原因：檔案控制代碼不足，這裡需要注意，如果是supervisor管理程序，則需要將檔案控制代碼的配置加到fe的啟動指令碼中。

if [[ $(ulimit -n) -lt 60000 ]]; then
  ulimit -n 65535
fi

c. StarRocks 支援使用 Java 語言編寫使用者定義函式 UDF，在執行函式報錯：“rpc failed, host: x.x.x.x”，be.out日誌中報錯：

start time: Tue Aug 9 19:05:14 CST 2022
Error occurred during initialization of VM
java/lang/NoClassDefFoundError: java/lang/Object

原因：在使用supervisor管理程序，需要注意增加JAVA_HOME環境變數，即使是BE節點也是需要呼叫Java的一些函式，也可以直接將BE啟動指令碼增加JAVA_HOME環境變數配置。

d. 執行Delete操作報錯如下：

SQL > delete from tableName partition (p20220809,p20220810) where `c_time` > '2022-08-09 15:20:00' and `c_time` < '2022-08-10 15:20:00';
ERROR 1064 (HY000): Where clause only supports compound predicate, binary predicate, is_null predicate and in predicate

原因：目前delete後的where條件不支援between and操作，目前只支援 =、>、>=、<、<=、!=、IN、NOT IN

e. 使用Routine Load消費kakfa資料的時候產生了大量隨機group_id

建議：建routine load的時候指定一下group name。

f. StarRocks連線超時，查詢語句報錯：“ERROR 1064(HY000):there is no scanNode Backend”，當重新啟動BE節點後，短暫的恢復。日誌報錯如下：

kafka log-4-FAIL, event: [thrd:x.x.x.x:9092/bootstrap]: x.x.x.x:9092/1: ApiVersionRequest failed: Local: Timed out: probably due to broker version < 0.10 (see api.version.request configuration) (after 10009ms in state APIVERSION_QUERY)

原因：當Routine Load連線kafka有問題時，會導致BrpcWorker執行緒耗盡，影響正常訪問連線StarRocks。臨時解決方案是找到問題任務，暫停任務，即可恢復。

6. 未來規劃

接下來我們會有更多業務接入 StarRocks，替換原有 OLAP 查詢引擎；運用更多的業務場景，積累經驗，提高叢集穩定性。未來希望 StarRocks 優化提升主鍵模型記憶體佔用，支援更靈活的部分列更新方式，持續優化提升 Bitmap 查詢效能，同時優化多租戶資源隔離。今後我們也會繼續積極參與 StarRocks 的社群討論，反饋業務場景。

*文/沈睿

關注得物技術，每週一三五晚18:30更新技術乾貨

要是覺得文章對你有幫助的話，歡迎評論轉發點贊～

「其他文章」