原文出處：大數據最佳實踐
鏈接：https://mp.weixin.qq.com/s/VlYyzLvTECM5XSRLklGrOg
目前的數倉大概分為離線數倉和實時數倉。離線數倉一般是T+1的數據ETL方案；實時數倉一般是分鍾級別甚至更短的時間內的ETL方案。實時數倉一般是將上游業務庫的數據通過binlog等形式，實時抽取到Kafka，進行實時ETL。但目前主流的實時數倉也會細分為兩類，一類是標准的實時數倉，所有的ETL過程都通過Spark或Flink等實時計算、落地，也就是說數據從binlog抽取到kafka，后續所有的ETL都是讀取kafka、計算、寫入kafka的形式串聯起來的，這種符合完整的數倉定義；還有一類是簡化的實時數倉，ETL簡化為有限的兩層，binlog落地到kafka之后，Spark或Flink讀取kafka計算完指標后落地HBase等存儲供外部查詢分析，當然也有通過Kylin或Druid來完成指標計算的。
那么“准實時數倉”又是一種什么方案呢？
其實“准實時數倉”是離線數倉的一種簡單的升級，它將離線的、天級別的ETL過程，縮短為小時或半小時級，但同時又對外提供實時的ODS層數據查詢。縮短離線數倉的計算頻率比較簡單，就是每小時或半小時增量抽取數據，MERGE到ODS層，后續的ETL過程與離線數倉完全一致。
對外提供實時的ODS層數據查詢有什么使用場景呢？
互聯網公司的業務庫一般都是MySQL，數據量比較大的情況下，會進行分庫分表；每個業務庫又會有不同的MySQL實例。如果想跨產品查詢數據就會非常麻煩，那誰會跨產品查詢數據呢？客服系統。客服系統一般可以查詢到某用戶所有的信息，如果用戶的信息分布在不同的MySQL實例、不同的庫、不同的表，查詢起來一定會涉及到sharding-jdbc，如果各個產品的sharding字段不同、算法不同，查詢一定會比較慢且非常復雜。此時就會需要有一個數據庫把這些數據匯總到一塊，而數倉的ODS層比較適合做這個工作。
准實時數倉”的兩個功能會涉及兩個技術難點：1）數據的增量抽取和增量MERGE；2）提供實時查詢接口。下面針對這兩個技術難點分別介紹對應的解決方案。
數據的增量抽取和MERGE。
其實增量抽取還算簡單，就是根據數據的某個字段進行增量抽取，這個字段可能是自增的ID或者更新時間。兩者有什么不同呢？
可以按照ID抽取的表，其中的數據一般不做更新，只是簡單的追加。那是不是只需要記錄上次抽取的最大ID，下一次從整個ID開始抽取就可以了呢？當然不是。
現在有以下場景：USER_LOGIN_HISTORY表有10個分表，分別為USER_LOGIN_HISTORY_0~ USER_LOGIN_HISTORY_9，其中ID是自增的，比如是auto_increment類型。但假如現在有3個並發事務，分配了3個ID值，比如是1/2/3。但這3個並發事務還沒提交的情況下，又來了3個並發事務，他們的ID應該是4/5/6。假如后面3個事務，提前提交，那么在進行增量抽數據的時候，當前ID的最大值是6,，很不巧，此時正值進行增量抽取數據，ID為1/2/3的數據並沒有抽取進來。那么這3個事務的數據就再也抽取不過來了！因為下次抽取時，ID的最大值是6，應該會從7開始抽取！
此時應該從上次抽取數據ID的最小值開始抽取，雖然有重復數據，但卻可以保證抽取的數據不會丟失。也就是說，如果當前批次是3，則應該從批次1抽取數據的最小ID開始抽取。這是為什么呢？請看下圖。
Batch1/batch2/batch3抽取時，MySQL表當前的最大ID分別是1001/2001/4002。Batch2抽取時應該是抽取10012001的數據，但很不巧的是，此時ID范圍在19901999的10條數據還沒有提交，抽取時就會漏掉。Batch3抽取時應該從哪里開始抽取呢？1001，1990，還是2001？
理想的情況應該從1990開始抽取，因為只是漏了1990~1999的數據。但怎么才能確定漏了哪些數據呢？答案是不知道。因為你根本不知道抽數時，哪些事務還沒有提交。很顯然應該從1001開始抽取。其實簡單來說就是每次都要抽取兩個batch的數據，來避免事務的影響。那一定就是兩個batch嗎？其實此處設置為兩個batch，其實是假設事務的最大持續時間小於每個batch的間隔時間的。對於准實時倉庫來說，每個batch一般都是小時或半小時，都會比事務最大持續時間大，所以兩個batch就夠了。如果batch時間間隔很小，那么久多向前推幾個batch就行了。
數據增量抽取后，MERGE就比較簡單了，其實就是用增量數據表與ODS全量表進行FULL JOIN，以增量數據為准就行了。但還需要考慮的是業務庫的表是否允許物理刪除，比如我們是不允許物理刪除的，所以FULL JOIN就行了。允許物理刪除就比較麻煩了，增量抽取是無法查詢到已經刪除的數據的！怎么辦呢？可以使用binlog把業務庫的Delete數據抽取到另外一張表，再用它來清洗ODS全量表就好了。
按照數據更新時間抽取，與上面的方案差不多，但有一點需要注意。抽取的時候，只能限制時間的最小值，而不能限制最大值。比如某個batch抽取時，當前時間是“2019年2月22日18:00:00.153”，因為從計算當前時間到實際抽取可能還是會相差幾毫秒或者幾秒，那么這期間更新的數據就可能會丟失，因為這些數據可能每次都會在這個間隔內被更新掉！
實時查詢接口。
實時且跨產品、跨庫、跨表查詢的系統一般都是中后台業務系統，這類系統的特點就是查詢數據源多、查詢結果數據量比較小。一般都是查詢某個或某些用戶的數據。這可以通過sharding-jdbc或ElasticSearch全文檢索來實現。sharding-jdbc雖然可能會有問題，但實施起來比較簡單，配置好sharding規則、寫好sql就可以了。
ElasticSearch全文檢索就比較麻煩了，由於ES沒有完善的sql接口，所以只能先將所需的數據匯總好，這又涉及到多表實時關聯匯總的問題。假設某個查詢結果涉及上游3張表，他們之間的關聯條件又不同，在實時匯總時，如果其中一張表的數據沒有到，另外兩張表就無法入庫，只能是先緩存數據等所有數據到達時再次匯總，實施難度還是比較大的。
當然也可以將所需的所有業務庫，抽取到某一個MySQL庫，進行查詢。數據量比較大時，這種方案就會很糟糕。
那比較好的方案是什么呢？我們可以把數據按照邏輯表（分庫分表整合后的表）通過binlog實時抽取到Phoenix，前台業務系統通過Phoenix的JDBC接口實時查詢。由於Phoenix支持索引，我們可以像使用MySQL一樣查詢Phoenix，當然了SQL可能需要優化。
由於Phoenix底層是基於HBase做的，可以承載海量數據的讀寫；而且HIVE也可以映射Phoenix進行離線查詢。這樣我們就把實時查詢和離線分析的需求進行了統一！那么這個方案有沒有什么問題呢？還是有一點需要考慮的：實時抽取的准確性如何保證呢？也就是說，binlog到Phoenix過程中，如果某一條更新日志丟失了該怎么辦呢？
很顯然可以用增量抽取的數據，補充到Phoenix中。那按照上面的增量抽取、補數邏輯是不是就沒事了呢？
其實還是有問題的，仍然是事務的問題，只不過這次是補數時的事務問題。增量數據一般比較大，那么耗時就比較久，假設為3分鍾，那么這個時間段內，實時更新的數據會不會被覆蓋掉呢？很顯然，一定會。既然會覆蓋，補數時就判斷一下主鍵相同的數據的更新時間嘍，以時間最大的為准。這還是有問題的，因為Phoenix默認是不開啟事務的，也就是說，判斷的時候，增量數據是最新的，但更新到Phoenix時，增量數據就不一定是最新的了，因為這個時間差內，實時數據進來了。
那就開啟Phoenix的事務唄，開啟應該能解決這個問題，但目前Phoenix的事務機制還是Beta版本，而且這還可能帶來性能問題和死鎖問題。
那怎么解決實時數據和離線增量數據相互覆蓋的問題了？有沒有兩全其美的方案呢？
熟悉HBase的同學一定知道，HBase有時間戳的概念，通過時間戳又支持多版本的查詢。通過Phoenix插入HBase時，所有列的時間戳都是RegionServer的當前時間，也就是說同一ID插入時，時間戳是遞增的，查詢時只能查詢到最新的數據。那這個跟上面的問題有啥關系呢？
如果我們把數據的更新時間映射到Phoenix底層HBase表的時間戳，是不是就完美解決事務的問題了呢？很簡單，數據的更新時間映射到HBase的時間戳，實時數據和增量數據，只需要簡單的插入Phoenix就好了，Phoenix查詢時只會查詢最新的數據！
然而，理想是完美的，現實是殘酷的，目前Phoenix不支持不同字段映射到HBase的時間戳！
沒辦法，只能改源碼。通過修改Phoenix源碼，我們使Phoenix支持了ROW_TS這一特殊的字段類型，這個類型的值會寫入HBase的時間戳，也就是說Phoenix插入數據時可以自由指定時間戳！下面是改造后的結果，很顯然，符合預期。
至此，“准實時數倉”的方案就介紹完了，下面通過架構圖簡單總結一下。
1） 上游MySQL的binlog通過Debezium (或Canal)和flume實時寫入Phoenix。新增字段時，實時修改Phoenix表結構
2） 按照數據更新時間，每小時抽取MySQL增量數據，將該部分數據批量MERGE到Phoenix
3） 每天自動創建Hive到Phoenix表的外部表（也可以創建Hive到Phoenix底層HBase表的外部表），進行后續的ETL過程。
4） 實時查詢平台通過JDBC連接Phoenix，按照主鍵或索引實時查詢數據