背景
HiTSDB時序資料庫引擎在服務于阿裡巴巴集團内的客戶時，根據集團業務特性做了很多針對性的優化。 然而在HiTSDB雲産品的打磨過程中逐漸發現，很多針對性的優化很難在公有雲上針對特定使用者去實施。
于此同時， 在公有雲客戶使用HiTSDB的過程中，發現了越來越多由于聚合查詢導緻的問題，比如： 傳回資料點過多會出現棧溢出等錯誤，聚合點過多導緻OOM， 或者無法完成聚合，執行個體完全卡死等等問題。這些問題主要由于原始的聚合引擎架構上的缺陷導緻。
是以HiTSDB開發團隊評估後決定圍繞新的聚合引擎架構對HiTSDB引擎進行更新，包含： 存儲模型的改造，索引方式的更新，實作全新的流式聚合，資料遷移，性能評測。 本文主要圍繞這5個方面進行梳理，重點在“全新的流式聚合部分”。
1. 時序資料存儲模型：
1.1 時序的資料存儲格式。
一個典型的時序資料由兩個次元來表示，一個次元表示時間軸，随着時間的不斷流入，資料會不斷地追加。 另外一個次元是時間線，由名額和資料源組成，資料源就是由一系列的标簽标示的唯一資料采集點。例如名額cpu.usage的資料來自于機房，應用，執行個體等次元組合成的采集點。 這樣大家邏輯上就可以抽象出來一個id+{timestamp, value}的時序資料模型。這種資料模型的存儲是如何呢。一般有兩種典型的資料存儲思路：
一種按照時間視窗次元劃分資料塊，同一段自然時間視窗内的連續資料放到相鄰的位置，比如{1:00, 2:00}->(id1, id2, id3, ... ... ,idN)。 采用這種方式的典型時序資料庫包含InfluxDB, Promethues等等TSMT結構的資料庫。OpenTSDB有些特殊，因為OpenTSDB是單值模型，名額這個次元在查詢的時候是必帶的。 是以可以先按照名額做了一級劃分，再根據時間視窗做二級的劃分，本質上還是同一時間視窗内的連續資料。 按照時間視窗切分的方式，優勢是寫入的時候可以很天然的按照視窗去落盤，對于高緯度的标簽查詢基本上是一些連續Scan. 這種方式有個比較難解的問題就是"out of order"亂序問題，對于時間視窗過期後再來的時間點，Promethues直接采用丢棄的方式，InfluxDB在這種情況下性能會有損耗。
另外一種按照時間線次元劃分資料塊，同一時間線的資料放到相鄰的位置，比如(id1)->(1:00, 2:00, 3:00, ... ... , 23:00)。 HiTSDB采用時間線次元劃分的方式：目前落盤資料存儲于HBASE，底層Rowkey由名額+标簽+自然視窗的方式組合而成. Rowkey按照大小順序合并某個時間線的資料點是連續相鄰的。 是以對于一些低維的查詢效率是非常高效的。根據目前接觸的一些物聯網服務，更多的是一些低維的通路。對于中等次元的查詢采用流式scan。對于極高緯度标簽的查詢HiTSDB采用預聚合的服務(不在本文讨論範圍内)。
1.2 時序模型的熱點問題處理
生産環境中業務方采集的名額類型多種多樣，對名額的采集周期各不相同。比如cpu.usage這個名額的變化頻率比較快，業務方關注度高，采集周期通常很短，1秒，5秒，10秒等等。 然而名額disk.usage這個名額變化趨勢相對平滑，采集周期通常為1分鐘，5分鐘， 10分鐘等。這種情況下，資料的存儲如果針對同一個名額不做特殊處理，容易形成熱點問題。 假設按照名額類型進行存儲資源的分片，想象一下如果有20個業務，每個業務10個叢集，每個叢集500台主機，采集周期是1秒的話，每秒就會有10萬個cpu.usage的名額資料點落到同一個存儲資源執行個體中， 而disk.usage采集周期為1分鐘，是以大約隻有1666個名額資料點落到另外一個存儲資源上，這樣資料傾斜的現象非常嚴重。
1.2.1 分桶
這類問題的經典解法就是分桶。比如除了名額類型外，同時将業務名和主機名作為次元辨別tags，把名額cpu.usage劃分到不同的桶裡面。 寫入時根據時間線哈希值分散寫入到不同的桶裡面。 OpenTSDB在處理熱點問題也是采用了分桶模式，但是需要廣播讀取，根本原因在于查詢方式需要在某個時間視窗内的全局掃描。 是以設定OpenTSDB的分桶數量需要一個平衡政策，如果數量太少，熱點還是有局部性的問題，如果太多，查詢時廣播讀帶來的開銷會非常大。
與其相比較，HiTSDB避免了廣播讀，提高了查詢效率。由于HiTSDB在查詢時，下發到底層存儲掃描資料之前，首先會根據查詢語句得到精确命中的時間線。 有了具體的時間線就可以确定桶的位置，然後到相應的塊區域取資料，不存在廣播讀。 關于HiTSDB如何在查詢資料的時候擷取命中的時間線，相信讀者這個疑問會在讀取完倒排這一節的時候消釋。
1.2.2 Region Pre-Split
當一個表剛被建立的時候，HBase預設配置設定一個Region給新表。所有的讀寫請求都會通路到同一個regionServer的同一個region中。 此時叢集中的其他regionServer會處于比較空閑的狀态，這個時候就達不到負載均衡的效果了。 解決這個問題使用pre-split,在建立新表的時候根據分桶個數采用自定義的pre-split的算法，生成多個region。 byte[][] splitKeys =new byte[bucketNumber-1][]; splitKeys[bucketIndex-1] = (bucketIndex&0xFF);
2. 反向索引：
2.1 時序資料中的多元時間線
多元支援對于任何新一代時序資料庫都是極其重要的。 時序資料的類型多種多樣,來源更是非常複雜,不止有單一次元上基于時間的有序數值，還有多元時間線相關的大量組合。 舉個簡單例子，cpu的load可以有三個次元描述cpu core， host， app應用，每個次元可以有百級别甚至萬級别的标簽值。 sys.cpu.load cpu=1 host=ipA app=hitsdb，各個次元組合後時間線可以輕松達到百萬級别。 如何管理這些時間線，建立索引并且提供高效的查詢是時序資料庫裡面需要解決的重要問題。 目前時序領域比較主流的做法是采用反向索引的方式。
2.2 反向索引基本組合
基本的時間線在倒排中的組合思路如下：
時間線的原始輸入值：
倒排建構後：
查詢時間線 cpu=3 and host=ipB：
取交集後查詢結果為7：
2.3 倒排面臨的問題以及優化思路
倒排主要面臨的是記憶體膨脹的問題：
posting list過長， 對于高緯度的tag，比如“機房=杭州”，杭州可能會有千級别甚至萬級别的機器，這就意味着posting list需要存儲成千上萬個64-bit的id。 解決這個問題的思路是采用壓縮posting list的方式， 在建構posting list的時候對數組裡面的id進行排序，然後采用delta編碼的方式壓縮。
如果Tag鍵值對直接作為term使用，記憶體占用取決于字元串的大小，采用字元串字典化，也可大大減少記憶體開銷。
3. 流式聚合引擎
3.1 HiTSDB聚合引擎的技術痛點
HiTSDB現有聚合引擎公有雲公測以及集體内部業務運作中，暴露發現了以下問題：
3.1.1 Materialization執行模式造成Heap記憶體易打爆
下圖顯示了原查詢引擎的架構圖。HiTSDB以HBase作為存儲，原引擎通過Async HBase client 從HBase擷取時序資料。由于HBase的資料讀取是一個耗時的過程，通常的解法是采用異步HBase client的API，進而有效提高系統的并行性。但原聚合引擎采用了一種典型的materialization的執行方式：1）啟動多個異步HBase API啟HBase讀，2）隻有當查詢所涉及的全部時序資料讀入到記憶體中後，聚合運算才開始啟動。這種把HBase Scan結果先在記憶體中materialized再聚合的方式使得HiTSDB容易發生Heap記憶體打爆的現象。尤其當使用者進行大時間範圍查詢，或者查詢的時間線的資料非常多的時候，因為涉及的時序資料多，HiTSDB會發生Heap OOM而導緻查詢失敗。
3.1.2 大查詢打爆HBase的問題
兩個原因造成HiTSDB處理聚合查詢的時候，容易發生将底層HBase打爆。
HBase 可能讀取多餘時間線資料。HiTSDB的時間線采用名額+時間視窗+标簽的編碼方式存儲在HBase。典型的查詢是使用者指定一個名額，時間範圍，以及空間次元上标簽要尋找的比對值。空間次元的标簽查詢條件并不都是在标簽編碼字首。當這種情況發生時，HiTSDB反向索引不能根據空間次元的查詢條件，精确定位到具體的HBase的查詢條件，而是采用先讀取再過濾的方式。這意味着HBase有可能讀取很多備援資料，進而加重HBase的負載。
HiTSDB有可能在短時間内下發太多HBase讀請求。一方面，HiTSDB在HBase采用分片存儲方式，對每一個分片，都至少啟動一個讀請求，另一方面，因為上面提到的materialization的執行方式，一個查詢涉及到的HBase讀請求同時異步送出，有可能在很短時間内向HBase下發大量的讀請求。這樣，一個大查詢就有可能把底層的HBase打爆。
當這種情況發生時，更糟糕的場景是HiTSDB無法處理時序資料的寫入請求，造成後續新資料的丢失。
3.1.3 執行架構高度耦合，修改或增加功能困難
聚合引擎主要針對應用場景是性能監控，查詢模式固定，是以引擎架構采用單一模式，把查詢，過濾，填值/插值，和聚合運算的邏輯高度耦合在一起。這種引擎架構對于監控應用的固定查詢沒有太多問題，但HiTSDB目标不僅僅是監控場景下的簡單查詢，而是着眼于更多應用場景下的複雜查詢。
我們發現采用原有引擎的架構，很難在原有基礎上進行增加功能，或修改原來的實作。本質上的原因在于原有聚合引擎沒有采用傳統資料庫所通常采用的執行架構，執行層由可定制的多個執行算子組成，查詢語義可以由不同的執行算子組合而完成。這個問題在産品開發開始階段并不感受很深，但确是嚴重影響HiTSDB拓寬應用場景，增加新功能的一個重要因素。
3.1.4 聚合運算效率有待提高
原有引擎在執行聚合運算的時候，也和傳統資料庫所通常采用的iterative執行模式一樣，疊代執行聚合運算。問題在于每次iteration執行，傳回的是一個時間點。Iterative 執行每次傳回一條時間點，或者一條記錄，常見于OLTP這樣的場景，因為OLTP的查詢所需要通路的記錄數很小。但對HiTSDB查詢有可能需要通路大量時間線資料，這樣的執行方式效率上并不可取。
原因1）每次處理一個時間點，都需要一系列的函數調用，性能上有影響，2）iterative循環疊代所涉及到的函數調用，無法利用新硬體所支援的SIMD并行執行優化，也無法将函數代碼通過inline等JVM常用的hotspot的優化方式。在大資料量的場景下，目前流行的通用做法是引入Vectorization processing, 也就是每次iteration傳回的不再是一條記錄，而是一個記錄集(batch of rows)，比如Google Spanner 用batch-at-a-time 代替了row-at-a-time, Spark SQL同樣也在其執行層采用了Vectorization的執行模式。
3.2 流式聚合引擎設計思路
針對HiTSDB原有聚合運算引擎上的問題，為了優化HiTSDB,支援HiTSDB商業化營運，我們決定改造HiTSDB聚合運算引擎。下圖給出了新聚合查詢引擎的基本架構。
3.2.1 pipeline執行模式
借鑒傳統資料庫執行模式，引入pipeline的執行模式（aka Volcano / Iterator 執行模式）。Pipeline包含不同的執行計算算子(operator), 一個查詢被實體計劃生成器解析分解成一個DAG或者operator tree， 由不同的執行算子組成，DAG上的root operator負責驅動查詢的執行，并将查詢結果傳回調用者。在執行層面，采用的是top-down需求驅動 (demand-driven)的方式，從root operator驅動下面operator的執行。這樣的執行引擎架構具有優點：
這種架構方式被很多資料庫系統采用并證明是有效;
接口定義清晰，不同的執行計算算子可以獨立優化，而不影響其他算子;
易于擴充：通過增加新的計算算子，很容易實作擴充功能。比如目前查詢協定裡隻定義了tag上的查詢條件。如果要支援名額值上的查詢條件(cpu.usage >= 70% and cpu.usage <=90%),可以通過增加一個新的FieldFilterOp來實作。
每個operator,實作如下接口：
Open : 初始化并設定資源
Next : 調用輸入operator的next()獲得一個batch of time series, 處理輸入，輸出batch of time series
Close : 關閉并釋放資源
我們在HiTSDB中實作了以下算子：
ScanOp: 用于從HBase異步讀取時間線資料
DsAggOp: 用于進行降采樣計算，并處理填值
AggOp：用于進行分組聚合運算，分成PipeAggOp, MTAggOp
RateOp: 用于計算時間線值的變化率
3.2.2 執行計算算子一個batch的時間線資料為運算機關
在計算算子之間以一個batch的時間線資料為機關，提高計算引擎的執行性能。其思想借鑒于OLAP系統所采用的Vectorization的處理模式。這樣Operator在處理一個batch的多條時間線，以及每條時間線的多個時間點，能夠減少函數調用的代價，提高loop的執行效率。
每個Operator以流式線的方式，從輸入獲得時間線batch, 經過處理再輸出時間線batch, 不用存儲輸入的時間線batch，進而降低對記憶體的要求。隻有當Operator的語義要求必須将輸入materialize，才進行這樣的操作（參見下面提到的聚合算子的不同實作）。
3.2.3. 區分不同查詢場景，采用不同聚合算子分别優化
HiTSDB原來的聚合引擎采用materialization的執行模式，很重要的一個原因在于處理時序資料的插值運算，這主要是因為時序資料的一個典型特點是時間線上不對齊：不同的時間線在不同的時間戳上有資料。HiTSDB相容OpenTSDB的協定，引入了插值(interpolation)的概念，目的在于聚合運算時通過指定的插值方式，在不對齊的時間戳上插入計算出來的值，進而将不對齊的時間線資料轉換成對齊的時間線。插值是在同一個group的所有時間線之間比較，來決定在哪個時間戳上需要進行插值 （參見OpenTSDB 文檔）。
為了優化聚合查詢的性能，我們引入了不同的聚合運算算子。目的在于針對不同的查詢的語義，進行不同的優化。有些聚合查詢需要插值，而有些查詢并不要求插值；即使需要插值，隻需要把同一聚合組的時間線資料讀入記憶體，就可以進行插值運算。
PipeAggOp: 當聚合查詢滿足以下條件時，
1）不需要插值： 查詢使用了降采樣(downsample),并且降采樣的填值采用了非null/NaN的政策。這樣的查詢，經過降采樣後，時間線的資料都是對齊補齊的，也就是聚合函數所用到的插值不再需要。
2）聚合函數可以支援漸進式疊代計算模式 (Incremental iterative aggregation), 比如sum, count ,avg, min, max, zerosum, mimmim, mimmax，我們可以采用incremental聚合的方式，而不需要把全部輸入資料讀入記憶體。這個執行算子采用了流水線的方式，每次從輸入的operator獲得一系列時間線，計算分組并更新聚合函數的部分值，完成後可以清理輸入的時間線，其自身隻用保留每個分組的聚合函數的值。
MTAgOp: 需要插值，并且輸入算子無法幫助将時間線ID預先分組，這種方式回退到原來聚合引擎所采用的執行模式。
對于MTAggOp, 我們可以引入分組聚合的方法進行優化：
GroupedAggOp: 需要插值，但是輸入算子能夠保證已經将時間線的ID根據辨別(tags)進行排序分組，這樣在流水線進行中，隻要materialize最多一個組的資料，這樣的算子比起記憶體保留所有分組時間線，記憶體要求要低，同時支援不同組之間的并行聚合運算。
3.2.4 查詢優化器和執行器
引入執行算子和pipeline執行模式後，我們可以在HiTSDB分成兩大子產品，查詢優化器和執行器。優化器根據查詢語義和執行算子的不同特點，産生不同的執行計劃，優化查詢處理。例如HiTSDB可以利用上面讨論的三個聚合運算算子，在不同的場景下，使用不同的執行算子，以降低查詢執行時的記憶體開銷和提高執行效率為目的。這樣的處理方式相比于原來聚合引擎單一的執行模式，更加優化。
4. 資料遷移
HiTSDB新的聚合引擎采用的底層存儲格式與以前的版本并不相容。 公有雲公測期間運作在舊版本執行個體的資料，需要遷移至新的聚合引擎。 同時熱更新出現了問題，資料遷移還應復原功能，将新版本的資料點轉換成舊的資料結構，實作版本復原。 整體方案對于使用者的影響做到：寫入無感覺，更新過程中，曆史資料不可讀。
4.1 資料遷移架構
并發轉換和遷移資料: 原有的HiTSDB資料點已經在寫入的時候進行了分片。預設有20個Salts。資料遷移工具會對每個Salt的資料點進行并發處理。 每個“Salt”都有一個Producer和一個Consumer。Producer負責開啟HBase Scanner擷取資料點。 每個Scanner異步對HBase進行掃描，每次擷取HBASE_MAX_SCAN_SIZE行數的資料點。然後将HBase的Row Key轉換成新的結構。
最後将該Row放到所有的一個Queue上等待Consumer消費。 Consumer每次會處理HBASE_PUT_BATCHSIZE或者HBASE_PUT_MIN_DATAPOINTS的資料量。 每次Consumer順利寫入該Batch的時候，我們會在UID表中記錄對應“Salt”的資料處理位置。 這樣便于故障重新開機時Producer從最後一次成功的地方重新開始擷取資料點進行轉換。 資料遷移工具對HBase的操作都采用異步的讀寫。當掃描資料或者寫入資料失敗的時候，我們會進行有限制的嘗試。 如果超出嘗試次數，我們就終止該“Salt”的資料遷移工作，其他”Salt“的工作不受到任何影響。 當下次工具自動重新開機時，我們會出現問題的”Salt“資料繼續進行遷移，直到所有資料全部順利轉換完成。
流控限制: 大部分情況下，Producer對HBase的掃描資料要快于Consumer對HBase的寫入。 為了防止Queue的資料積壓對記憶體造成壓力同時為了減少Producer掃描資料時對HBase的壓力，我們設定了流控。 當Queue的大小達到HBASE_MAX_REQUEST_QUEUE_SIZE時候，Producer會暫時停止對HBase的資料掃描等待Consumer消費。 當Queue的大小減少到HBASE_RESUME_SCANNING_REQUEST_QUEUE_SIZE時候，Producer會重新恢複。
Producer和Consumer程序的退出
順利完成時候如何退出: 當一切進展順利時候，當Producer完成資料掃描之後，會在Queue上放一個EOS（End of Scan），然後退出。 Consumer遇到EOS就會知道該Batch為最後一批，成功處理完該Batch之後就會自動退出。
失敗後如何關閉: Consumer遇到問題時：當Consumer寫入HBase失敗之後，consumer會設定一個Flag，然後退出線程。 每當Producer準備進行下一個HBASE_MAX_SCAN_SIZE的掃描時候，他會先檢查該Flag。 如果被設定，他會知道對應的Consumer線程已經失敗并且退出。Producer也會停止掃描并且退出。 Producer遇到問題時：當Producer掃描資料失敗時，處理方式和順利完成時候類似。都是通過往Queue上EOS來完成通知。 下次重新開機時，Producer會從上次記錄的資料處理位置開始重新掃描。
4.2 資料遷移的一緻性
由于目前雲上版本HiTSDB為雙節點，在結點更新結束後會自動重新開機HiTSDB。自動啟動腳本會自動運作資料遷移工具。 如果沒有任何預防措施，此時兩個HiTSDB節點會同時進行資料遷移。雖然資料上不會造成任何丢失或者損壞， 但是會對HBase造成大量的寫入和讀取壓力進而嚴重影響使用者的正常的寫入和查詢性能。
為了防止這樣的事情發生，我們通過HBase的Zoo Keeper實作了類似FileLock鎖，我們稱為DataLock，的機制保證隻有一個結點啟動資料遷移程序。 在資料遷移程序啟動時，他會通過類似非阻塞的tryLock()的形式在Zoo Keeper的特定路徑建立一個暫時的節點。 如果成功建立節點則代表成果獲得DataLock。如果該節點已經存在，即被另一個HiTSDB建立，我們會收到KeeperException。這樣代表未獲得鎖，馬上傳回失敗。 如果未成功獲得DataLock，該節點上的資料遷移程序就會自動退出。成果獲得DataLock的節點則開始進行資料遷移。
4.3 資料遷移中的"執行一次"
當所有“Salt”的資料點全部順利完成遷移之後，我們會在HBase的舊表中插入一行新資料，data_conversion_completed。 此行代表了資料遷移工程全部順利完成。同時自動腳本會每隔12個小時啟動資料遷移工具，這樣是為了防止上次資料遷移沒有全部完成。 每次啟動時，我們都會先檢查“data_conversion_completed”标志。如果标志存在，工具就會馬上退出。 此項操作隻會進行一次HBase的查詢，比正常的健康檢查成本還要低。是以周期性的啟動資料遷移工具并不會對HiTSDB或者HBase産生影響。
4.4. 資料遷移的評測
測試機型: 4core,8G,SSD
效果：上線後無故障完成100+執行個體資料的遷移，熱更新。
5. 查詢性能評測
測試環境配置
192.168.12.3 2.1.5版本
192.168.12.4 2.2.0版本（Pipelined Engine）
測試資料 - 1萬條時間，不同的采集頻率和時間視窗，還有查詢命中的時間線數量。
Case 1: 資料采集頻率5s, 查詢命中1000條，時間視窗3600s
Case 2: 資料采集頻率1s，查詢命中1條，時間視窗36000s
總結： 新的查詢聚合引擎将查詢速度提高了10倍以上。
其他
本文介紹了高性能時間序列資料庫HiTSDB引擎在商業化營運之前進行的優化更新，目的是提高HiTSDB引擎的穩定性，資料寫入和查詢性能以及新功能的擴充性。HiTSDB已經在阿裡雲正式商業化營運，我們将根據使用者回報，進一步提高HiTSDB引擎，更好服務于HiTSDB的客戶。
原文釋出時間為：2018-04-18
本文作者：阿裡HiTSDB團隊
本文來自雲栖社群合作夥伴“
阿裡技術”，了解相關資訊可以關注“
”。