1、需求背景

根據(jù)目前大數(shù)據(jù)這一塊的發(fā)展，已經(jīng)不局限于離線的分析，挖掘數(shù)據(jù)潛在的價(jià)值，數(shù)據(jù)的時(shí)效性最近幾年變得剛需，實(shí)時(shí)處理的框架有storm，spark-streaming，flink等。想要做到實(shí)時(shí)數(shù)據(jù)這個(gè)方案可行，需要考慮以下幾點(diǎn)：1、狀態(tài)機(jī)制 2、精確一次語(yǔ)義 3、高吞吐量 4、可彈性伸縮的應(yīng)用 5、容錯(cuò)機(jī)制，剛好這幾點(diǎn)，flink都完美的實(shí)現(xiàn)了，并且支持flink sql高級(jí)API，減少了開(kāi)發(fā)成本，可用實(shí)現(xiàn)快速迭代，易維護(hù)等優(yōu)點(diǎn)。

2、離線數(shù)倉(cāng)和實(shí)時(shí)數(shù)倉(cāng)對(duì)比

離線數(shù)倉(cāng)的架構(gòu)圖：

離線數(shù)倉(cāng)的架構(gòu)圖.png

實(shí)時(shí)數(shù)倉(cāng)架構(gòu)圖:

實(shí)時(shí)數(shù)倉(cāng)架構(gòu)圖.png

3、實(shí)時(shí)數(shù)倉(cāng)的架構(gòu)詳細(xì)介紹

3.1、橫向劃分介紹（層級(jí)劃分）

3.1.1、數(shù)據(jù)接入（source）

1. 流量日志
   流量數(shù)據(jù)天然就是流式的，具有實(shí)時(shí)性，主要是如何采集的問(wèn)題。流量數(shù)據(jù)對(duì)應(yīng)著流量數(shù)據(jù)域，binlog往往是其它數(shù)據(jù)域，比如交易域，營(yíng)銷域等，以流量日志為例，打點(diǎn)日志上報(bào)到nginx服務(wù)器，使用flume進(jìn)行數(shù)據(jù)采集，sink進(jìn)kafka，目前kafka只保留最近三天的數(shù)據(jù)，考慮到流量日志的數(shù)據(jù)量大，并且也沒(méi)有保留多天的意義，保留三天，可以預(yù)留重刷歷史數(shù)據(jù)的機(jī)會(huì)，如果是要查看多天以前的數(shù)據(jù)情況，完全可以用離線的。所以整套實(shí)時(shí)數(shù)倉(cāng)體系建設(shè)都是為了保障近一天的數(shù)據(jù)分析。
2. 關(guān)系型數(shù)據(jù)庫(kù)
   目前大多數(shù)互聯(lián)網(wǎng)公司使用的生產(chǎn)數(shù)據(jù)庫(kù)都是mysql，以mysql為例，mysql發(fā)生任意變更都會(huì)產(chǎn)生binlog，使用開(kāi)源的canal解釋原生二進(jìn)制的數(shù)據(jù)，然后將解釋完的數(shù)據(jù)sink進(jìn)kafka，在將kafka作為flink source進(jìn)行消費(fèi)。這里值得一提的是，關(guān)系型數(shù)據(jù)庫(kù)的數(shù)據(jù)往往是用于構(gòu)建除流量域以外的數(shù)據(jù)域使用的。

3.1.2、數(shù)據(jù)計(jì)算（transform）

1. ods層
   該成屬于明細(xì)層，主要擔(dān)任的職責(zé)是對(duì)數(shù)據(jù)進(jìn)行清洗，解析，規(guī)范化處理，拿流量日志來(lái)說(shuō)，使用flink sql對(duì)接kafka，使用自定義的udtf函數(shù)解析kafka當(dāng)中的原始log，產(chǎn)生結(jié)構(gòu)化數(shù)據(jù)，并且在次寫(xiě)入kafka的另一個(gè)topic當(dāng)中，這就是我們的實(shí)時(shí)ods層數(shù)據(jù)了。當(dāng)然關(guān)系型數(shù)據(jù)庫(kù)相對(duì)來(lái)得簡(jiǎn)單一點(diǎn)，因?yàn)槲覀円呀?jīng)使用canal解析好了，直接使用flink sql讀取kafka源就好了。

2. 隱藏的校驗(yàn)層
   為了校驗(yàn)實(shí)時(shí)數(shù)據(jù)的準(zhǔn)確性（不管是流量數(shù)據(jù)還是binlog數(shù)據(jù)），還需要將存于kafka的ods層數(shù)據(jù)，寫(xiě)入hdfs上，使用hive和hdfs的文件進(jìn)行映射，產(chǎn)生實(shí)時(shí)的hive表（目前是小時(shí)級(jí)別），該hive表可用于和離線hive表進(jìn)行數(shù)據(jù)校正。

3. dwd層
   dwd層的數(shù)據(jù)是從ods層讀取，然后根據(jù)需求進(jìn)行邏輯處理，包括關(guān)聯(lián)相應(yīng)的維度表（維度表的建設(shè)后續(xù)會(huì)提及），即進(jìn)行降維操作。

4. DM層
   DM層存儲(chǔ)的是一些按照一定粒度進(jìn)行聚合的值，我們之所以選擇將DM層的數(shù)據(jù)存于hbase，原因在于hbase集群可擴(kuò)展，支持巨量數(shù)據(jù)，并且根據(jù)rowkey查找，性能也很感人（前提是使用得當(dāng)），最關(guān)鍵的是hbase的rowkey是天然唯一的，剛好符合聚合的模式，我們只需要將聚合的字段作為rowkey的因子，在用MD5加密一下，就是rowkey了。

5. APP/RPT層
   該層對(duì)應(yīng)著離線的應(yīng)用層/報(bào)表層，這一層跟業(yè)務(wù)是緊密耦合的，但是這一層產(chǎn)出的數(shù)據(jù)來(lái)源離不開(kāi)我們底層的建設(shè)。

3.1.3、數(shù)據(jù)存儲(chǔ)（sink）

目前是將實(shí)時(shí)維度表和DM層數(shù)據(jù)存于hbase當(dāng)中，實(shí)時(shí)公共層都存于kafka當(dāng)中，并且以寫(xiě)滾動(dòng)日志的方式寫(xiě)入HDFS（主要是用于校驗(yàn)數(shù)據(jù)）。其實(shí)在這里可以做的工作還有很多，kafka集群，flink集群，hbase集群相互獨(dú)立，這對(duì)整個(gè)實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)的穩(wěn)定性帶來(lái)一定的挑戰(zhàn)。

3.2、縱向劃分介紹（數(shù)據(jù)域劃分）

一個(gè)數(shù)據(jù)倉(cāng)庫(kù)想要成體系，成資產(chǎn)，離不開(kāi)數(shù)據(jù)域的劃分。所以參考著離線的數(shù)據(jù)倉(cāng)庫(kù)，想著在實(shí)時(shí)數(shù)倉(cāng)做出這方面的探索，理論上來(lái)講，離線可以實(shí)現(xiàn)的，實(shí)時(shí)也是可以實(shí)現(xiàn)的。 并且目前已經(jīng)取得了成效，目前劃分的數(shù)據(jù)域跟離線大致相同，有流量域，交易域，營(yíng)銷域等等。當(dāng)然這里面涉及到維表，多事務(wù)事實(shí)表，累計(jì)快照表，周期性快照表的設(shè)計(jì)，開(kāi)發(fā)，到落地這里就不詳述了。

3.3、實(shí)時(shí)維度表介紹

維度表也是整個(gè)實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)不可或缺的部分。從目前整個(gè)實(shí)時(shí)數(shù)倉(cāng)的建設(shè)來(lái)看，維度表有著數(shù)據(jù)量大，但是變更少的特點(diǎn)，我們?cè)囅脒^(guò)構(gòu)建全平臺(tái)的實(shí)時(shí)商品維度表或者是實(shí)時(shí)會(huì)員維度表，但是這類維度表太過(guò)于復(fù)雜，所以針對(duì)這類維度表下面介紹。還有另外一種就是較為簡(jiǎn)單的維度表，這類維度可能對(duì)應(yīng)著業(yè)務(wù)系統(tǒng)單個(gè)mysql表，或者只需要幾個(gè)表進(jìn)行簡(jiǎn)單ETL就可以產(chǎn)出的表，這類維表是可以做成實(shí)時(shí)的。以下有幾個(gè)實(shí)施的關(guān)鍵點(diǎn)：

1. mysql是天然的實(shí)時(shí)維度表，可惜不能用？
   在流式程序里面，維度表跟流進(jìn)行join，面臨著很高的QPS，或者可以理解為長(zhǎng)連接，如果直接用flink去讀取mysql作為維表，大概率是會(huì)掛掉的，對(duì)生產(chǎn)系統(tǒng)的穩(wěn)定性有很大影響。
2. 實(shí)時(shí)維表存于hbase當(dāng)中
   既然mysql行不通，只能另尋他方，如果hbase里面能維護(hù)一份跟mysql實(shí)時(shí)同步的維表，那么問(wèn)題應(yīng)該是可以解決的，因?yàn)閔base跟生產(chǎn)系統(tǒng)無(wú)關(guān)。那么如何實(shí)現(xiàn)實(shí)時(shí)同步呢？我們做法是先全量同步一份數(shù)據(jù)到hbase當(dāng)中，（注意rowkey的設(shè)計(jì)，這個(gè)因子一定是用于維表關(guān)聯(lián)用的），然后將該mysql表的binlog日志使用canal解析完落到kafka，使用flink去消費(fèi)kafka的數(shù)據(jù)，然后將新增和變更的數(shù)據(jù)實(shí)時(shí)同步到hbase，這樣就實(shí)現(xiàn)了mysql和hbase的數(shù)據(jù)實(shí)時(shí)同步。

4.實(shí)時(shí)數(shù)倉(cāng)難點(diǎn)討論

4.1 如何保證接入數(shù)據(jù)的準(zhǔn)確性

如下是離線數(shù)據(jù)同步架構(gòu)圖：

離線數(shù)據(jù)同步架構(gòu)圖.png

4.1.1實(shí)時(shí)和離線數(shù)據(jù)接入的差異性

實(shí)時(shí)數(shù)據(jù)的接入其實(shí)在底層架構(gòu)是一樣的，就是從kafka那邊開(kāi)始不一樣，實(shí)時(shí)用flink的UDTF進(jìn)行解析，而離線是定時(shí)（目前是小時(shí)級(jí)）用camus拉到HDFS，然后定時(shí)load HDFS的數(shù)據(jù)到hive表里面去，這樣來(lái)實(shí)現(xiàn)離線數(shù)據(jù)的接入。實(shí)時(shí)數(shù)據(jù)的接入是用flink解析kafka的數(shù)據(jù)，然后在次寫(xiě)入kafka當(dāng)中去。

4.1.2如何建立實(shí)時(shí)數(shù)據(jù)和離線數(shù)據(jù)的可比較性

由于目前離線數(shù)據(jù)已經(jīng)穩(wěn)定運(yùn)行了很久，所以實(shí)時(shí)接入數(shù)據(jù)的校驗(yàn)可以對(duì)比離線數(shù)據(jù)，但是離線數(shù)據(jù)是小時(shí)級(jí)的hive數(shù)據(jù)，實(shí)時(shí)數(shù)據(jù)存于kafka當(dāng)中，直接比較不了，所以做了相關(guān)處理，將kafka的數(shù)據(jù)使用flink寫(xiě)HDFS滾動(dòng)日志的形式寫(xiě)入HDFS，然后建立hive表小時(shí)級(jí)定時(shí)去load HDFS中的文件，以此來(lái)獲取實(shí)時(shí)數(shù)據(jù)。

4.1.3如何確定比較的時(shí)間區(qū)間

完成以上兩點(diǎn)，剩余還需要考慮一點(diǎn)，都是小時(shí)級(jí)的任務(wù)，這個(gè)時(shí)間卡點(diǎn)使用什么字段呢?首先要確定一點(diǎn)就是離線和實(shí)時(shí)任務(wù)卡點(diǎn)的時(shí)間字段必須是一致的，不然肯定會(huì)出問(wèn)題。目前離線使用camus從kafka將數(shù)據(jù)拉到HDFS上，小時(shí)級(jí)任務(wù)，使用nginx_ts這個(gè)時(shí)間字段來(lái)卡點(diǎn)，這個(gè)字段是上報(bào)到nginx服務(wù)器上記錄的時(shí)間點(diǎn)。而實(shí)時(shí)的數(shù)據(jù)接入是使用flink消費(fèi)kafka的數(shù)據(jù)，在以滾動(dòng)日志的形式寫(xiě)入HDFS的，然后在建立hive表load HDFS文件獲取數(shù)據(jù)，雖然這個(gè)hive也是天/小時(shí)二級(jí)分區(qū)，但是離線的表是根據(jù)nginx_ts來(lái)卡點(diǎn)分區(qū)，但是實(shí)時(shí)的hive表是根據(jù)任務(wù)啟動(dòng)去load文件的時(shí)間點(diǎn)去區(qū)分的分區(qū)，這是有區(qū)別的，直接篩選分區(qū)和離線的數(shù)據(jù)進(jìn)行對(duì)比，會(huì)存在部分差異，應(yīng)當(dāng)?shù)淖龇ㄊ呛Y選范圍分區(qū)，然后在篩選nginx_ts的區(qū)間，這樣在跟離線做對(duì)比才是合理的。

4.2如何保證接入數(shù)據(jù)的時(shí)延

目前實(shí)時(shí)數(shù)據(jù)接入層的主要時(shí)延是在UDTF函數(shù)解析上，實(shí)時(shí)的UDTF函數(shù)是根據(jù)上報(bào)的日志格式進(jìn)行開(kāi)發(fā)的，可以完成日志的解析功能。

解析流程圖如下：

日志解析流程.png

解析速率圖如下：

解析速率圖.png

該圖還不是在峰值數(shù)據(jù)量的時(shí)候截的，目前以800記錄/second為準(zhǔn)，大概一個(gè)記錄的解析速率為1.25ms。
目前該任務(wù)的flink資源配置核心數(shù)為1，假設(shè)解析速率為1.25ms一條記錄，那么峰值只能處理800條/second，如果數(shù)據(jù)接入速率超過(guò)該值就需要增加核心數(shù)，保證解析速率。

4.3 維度表設(shè)計(jì)成實(shí)時(shí)的復(fù)雜度過(guò)高

4.3.1實(shí)時(shí)維表背景介紹

介紹一下目前離線維度表的情況，就拿商品維度表來(lái)說(shuō)，全線記錄數(shù)將近一個(gè)億，計(jì)算邏輯來(lái)自40-50個(gè)ods層的數(shù)據(jù)表，計(jì)算邏輯相當(dāng)復(fù)雜，如果實(shí)時(shí)維度表也參考離線維度表來(lái)完成的話，那么開(kāi)發(fā)成本和維護(hù)成本非常大，對(duì)于技術(shù)來(lái)講也是很大的一個(gè)挑戰(zhàn)，并且目前也沒(méi)有需求要求維度屬性百分百準(zhǔn)確。所以目前（偽實(shí)時(shí)維度表）準(zhǔn)備在當(dāng)天24點(diǎn)產(chǎn)出，當(dāng)天的維度表給第二天實(shí)時(shí)公共層使用，即T-1的模式。偽實(shí)時(shí)維度表的計(jì)算邏輯參考離線維度表，但是為了保障在24點(diǎn)之前產(chǎn)出，需要簡(jiǎn)化一下離線計(jì)算邏輯，并且去除一些不常用的字段，保障偽實(shí)時(shí)維度表可以較快產(chǎn)出。

實(shí)時(shí)維度表的計(jì)算流程圖：

實(shí)時(shí)維度表的計(jì)算流程圖.png

4.3.2在實(shí)施的過(guò)程當(dāng)中的細(xì)節(jié)點(diǎn)

1. 根據(jù)實(shí)時(shí)維度表需要的屬性字段對(duì)離線維度表進(jìn)行簡(jiǎn)化操作，并且裁剪ods層的計(jì)算邏輯，理順實(shí)時(shí)維度表的計(jì)算邏輯。

2. 實(shí)時(shí)維度表使用到的stage和ods層數(shù)據(jù)表保存周期都不需要太長(zhǎng)，一般保存數(shù)天就好。

3. 由于實(shí)時(shí)維度表需要在24點(diǎn)之前產(chǎn)出并寫(xiě)入到hbase當(dāng)中，所以要考慮將任務(wù)定于幾點(diǎn)開(kāi)始跑，比如所有抽取任務(wù)和ods計(jì)算任務(wù)都從23點(diǎn)開(kāi)始跑，當(dāng)然要看具體任務(wù)耗時(shí)來(lái)定，如果耗時(shí)過(guò)長(zhǎng)需要在提前一點(diǎn)。

4. 根據(jù)以上步驟去完成，感覺(jué)剩下來(lái)只要將數(shù)據(jù)寫(xiě)入hbase就好了，但是這里也有一個(gè)巨坑。如果將rowkey設(shè)計(jì)成md5（pt+維度表主鍵），然后hbase保存近兩天的數(shù)據(jù)，這樣當(dāng)實(shí)時(shí)數(shù)據(jù)出現(xiàn)問(wèn)題，我們還可以進(jìn)行重刷數(shù)據(jù)。但是我們不管是商品維度表還是用戶維度表都達(dá)到了數(shù)千萬(wàn)的級(jí)別，如果每天全量寫(xiě)入hbase的話，我們做了壓測(cè)計(jì)算hbase的寫(xiě)入速率，大概400百萬(wàn)條/10min，如果同步以一億條記錄的話，大概就需要250分鐘，對(duì)于時(shí)效要求這么高的實(shí)時(shí)維度表，這個(gè)時(shí)間肯定是接收不了的，所以row的設(shè)計(jì)不能將pt放入，但是這樣的話就無(wú)法保存歷史數(shù)據(jù)，如果實(shí)時(shí)數(shù)據(jù)發(fā)生異常，重刷數(shù)據(jù)時(shí)部分實(shí)時(shí)公共層關(guān)聯(lián)的維度信息是不準(zhǔn)確的，所以我們?cè)谶@點(diǎn)上做了取舍，放棄重刷數(shù)據(jù)，畢竟出現(xiàn)數(shù)據(jù)異常的概率很小，就算出現(xiàn)了，關(guān)聯(lián)的維度信息不準(zhǔn)確的部分也很少（維度信息每天只會(huì)有部分發(fā)生變化，可能不到百分之一）。既然這種全量走不通，就要考慮增量同步，如果區(qū)分該條記錄是否發(fā)生了屬性變化，我們采用的是將全字段做md5處理，只要任一一個(gè)字段發(fā)生變化，md5就會(huì)發(fā)生變化，在使用一個(gè)flag字段來(lái)做標(biāo)識(shí)，flag的計(jì)算邏輯就是拿當(dāng)天的md5和昨天的md5進(jìn)行比較，相同為0（表示未變化），不同為1（表示發(fā)生變化），到時(shí)候我們只將flag=1的數(shù)據(jù)同步到hbase就好了，rowkey設(shè)計(jì)為md5（維度表主鍵），這樣每天只會(huì)把變化差異維度記錄同步到hbase，大概每天有幾百萬(wàn)，這樣的同步時(shí)間是可以接受的。其實(shí)這里還有一個(gè)小點(diǎn)沒(méi)有考慮到，實(shí)時(shí)維度表假設(shè)是在23:50產(chǎn)出，那么23:50到24:00使用的就是最新的實(shí)時(shí)維度表了，而不是昨天的實(shí)時(shí)維度表，這也是存在部分差異的點(diǎn)，但是從目前這個(gè)情況考慮，暫時(shí)需要做一些取舍。

4.4 用flink進(jìn)行窗口計(jì)算，窗口過(guò)大，內(nèi)存問(wèn)題

4.4.1、背景介紹

目前使用flink作為公司主流的實(shí)時(shí)計(jì)算引擎，使用內(nèi)存作為狀態(tài)后端，并且固定30s的間隔做checkpoint，使用HDFS作為checkpoint的存儲(chǔ)組件。并且checkpoint也是作為任務(wù)restart以后恢復(fù)狀態(tài)的重要依據(jù)。熟悉flink的人應(yīng)該曉得，使用內(nèi)存作為狀態(tài)后端，這個(gè)內(nèi)存是JVM的堆內(nèi)存，畢竟是有限的東西，使用不得當(dāng)，OOM是常有的事情，下面就介紹一下針對(duì)有限的內(nèi)存，如果完成常規(guī)的計(jì)算。

1. 滑動(dòng)窗口 or 滾動(dòng)窗口？
   其實(shí)只是從內(nèi)存使用的角度來(lái)講，滑動(dòng)窗口和滾動(dòng)窗口都是可取的，關(guān)鍵看如何使用。首先要了解你的每個(gè)task產(chǎn)出的數(shù)據(jù)存于內(nèi)存是個(gè)什么形式，比如你只是計(jì)算pv這種指標(biāo)，哪怕你是使用窗口大小為24hours的滑動(dòng)窗口也是可取的，因?yàn)檫@個(gè)狀態(tài)在內(nèi)存當(dāng)中只是一個(gè)聚合值，不怎么占內(nèi)存，當(dāng)然如果多維聚合，條數(shù)特別多也是另當(dāng)別論。舉另外一個(gè)極端的例子，如果是流量數(shù)據(jù)，直接做map操作，這時(shí)候哪怕你是使用10min的滾動(dòng)窗口，內(nèi)存可能就吃不消了（具體要看每個(gè)flink任務(wù)分配多少內(nèi)存和核數(shù)），因?yàn)榱髁繑?shù)據(jù)非常大，而且在內(nèi)存當(dāng)中是以明細(xì)的形式存在，這時(shí)候就會(huì)非常占用內(nèi)存。
2. 數(shù)據(jù)量大，時(shí)間跨度長(zhǎng)，需要聚合數(shù)據(jù)
   這種情況好說(shuō)，直接上滑動(dòng)窗口，窗口大小可以大一點(diǎn)，因?yàn)闋顟B(tài)不怎么占內(nèi)存（多維度聚合值條數(shù)可能也很大，需要具體判斷）。
3. 數(shù)據(jù)量大，時(shí)間跨度長(zhǎng)，并且需要明細(xì)數(shù)據(jù)
   在實(shí)際應(yīng)用當(dāng)中，這情況很常見(jiàn)，如果只是使用上述的兩個(gè)窗口，你會(huì)發(fā)現(xiàn)，好像都不能很好應(yīng)對(duì)這種情況，為了應(yīng)對(duì)這種情況，一般使用較小窗口大小的滾動(dòng)窗口，比如10s的滾動(dòng)窗口，然后將這個(gè)窗口計(jì)算完的值存于持久化存儲(chǔ)hbase當(dāng)中，然后在進(jìn)行下一個(gè)10s窗口的計(jì)算，那么這樣狀態(tài)也不會(huì)丟失，flink內(nèi)存也只需要存儲(chǔ)10s的狀態(tài)就好。
4. 數(shù)據(jù)量大，時(shí)間跨度長(zhǎng)，需要去重累計(jì)
   這種情況應(yīng)該是流式計(jì)算里面最麻煩的一種情況了，但是確實(shí)又存在，比如計(jì)算天級(jí)的uv（按userid進(jìn)行去重得到的指標(biāo)），這種情況，不能將狀態(tài)直接聚合累計(jì)，像上述的2描述的一樣，因?yàn)樗枰ブ兀驗(yàn)橐ブ鼐鸵S護(hù)著整個(gè)時(shí)間跨度內(nèi)的明細(xì)數(shù)據(jù)，但是這樣又非常占用內(nèi)存，看似非常矛盾的一件事情，該如何去解決。
   其實(shí)有多種方法，使用Hbase或者Redis都可以實(shí)現(xiàn)，這里以hbase為例，比如現(xiàn)在需要按照stat_date（日期）維度計(jì)算uv指標(biāo)，那么可以將MD5(start_date+userid)作為rowkey插入到hbase當(dāng)中，那么如果是同樣的start_date+userid記錄插入到hbase當(dāng)中，記錄數(shù)是不會(huì)增加的，因?yàn)閞owkey一定是全局唯一，這樣就實(shí)現(xiàn)了去重，那么如何實(shí)現(xiàn)累計(jì)呢，累計(jì)其實(shí)就是將hbase里面符合的條數(shù)取出，自己寫(xiě)一個(gè)小方法，思路大概就是這樣。Redis其實(shí)是一樣的原理，這里就不多做解釋了。