• 業(yè)務(wù)簡(jiǎn)介
• 數(shù)倉演進(jìn)
• 場(chǎng)景實(shí)踐
• 未來展望

01業(yè)務(wù)簡(jiǎn)介

構(gòu)建實(shí)時(shí)數(shù)據(jù)分析平臺(tái)是為了更好的解決業(yè)務(wù)對(duì)更高數(shù)據(jù)時(shí)效性的需求，先簡(jiǎn)單介紹一下業(yè)務(wù)流程。

從日常的場(chǎng)景說起，當(dāng)我們打開手機(jī) APP 時(shí)，常會(huì)看到廣告。在這樣一個(gè)場(chǎng)景中，涉及到了兩個(gè)比較重要的角色。一是手機(jī) APP，即流量方；另一個(gè)是投廣告的廣告主，如支付寶、京東會(huì)投放電商廣告。廣告主購買流量方的流量投廣告就產(chǎn)生了交易。

訊飛構(gòu)建了一個(gè)流量交易平臺(tái)，流量交易平臺(tái)主要的職能是聚合下游流量，上游再對(duì)接廣告主，從而幫助廣告主和流量方在平臺(tái)上進(jìn)行交易。訊飛還構(gòu)建了投放平臺(tái)，這個(gè)平臺(tái)更側(cè)重于服務(wù)廣告主，幫助廣告主投放廣告，優(yōu)化廣告效果。

在上述的業(yè)務(wù)流程圖中，APP 與平臺(tái)交互時(shí)會(huì)向平臺(tái)發(fā)起請(qǐng)求，然后平臺(tái)會(huì)下發(fā)廣告，用戶隨后才能看到廣告。用戶看到廣告的這個(gè)動(dòng)作稱之為一次曝光，APP 會(huì)把這次曝光行為上報(bào)給平臺(tái)。如果用戶點(diǎn)擊了廣告，那么 APP 也會(huì)上報(bào)點(diǎn)擊行為。

廣告在產(chǎn)生之后發(fā)生了很多行為，可以將廣告的整個(gè)過程稱為廣告的一次生命周期，不僅限于圖中的請(qǐng)求、曝光、點(diǎn)擊這三次行為，后面可能還有下單、購買等。

在這樣一個(gè)業(yè)務(wù)流程中，業(yè)務(wù)的核心訴求是什么呢？在廣告的生命周期中有請(qǐng)求、曝光和點(diǎn)擊等各種行為，這些行為會(huì)產(chǎn)生對(duì)應(yīng)的業(yè)務(wù)日志。那么就需要從日志生成數(shù)據(jù)供業(yè)務(wù)側(cè)分析，從日志到分析的過程中就引入了數(shù)倉構(gòu)建、數(shù)倉分層，數(shù)據(jù)呈現(xiàn)的時(shí)效性就帶來了實(shí)時(shí)數(shù)據(jù)倉庫的發(fā)展。

02數(shù)倉演進(jìn)

上圖是一個(gè)典型的數(shù)倉分層框架，最底層是 ODS 數(shù)據(jù)，包括業(yè)務(wù)日志流、OLTP 數(shù)據(jù)庫、第三方文檔數(shù)據(jù)。經(jīng)過 ETL 將 ODS 層的數(shù)據(jù)清洗成業(yè)務(wù)模型，也就是 DWD 層。

最初是建立了 Spark 數(shù)倉，將業(yè)務(wù)日志收集到 Kafka 中再投遞到 HDFS 上，通過 Spark 對(duì)日志進(jìn)行清洗建模，然后將業(yè)務(wù)模型再回寫到 HDFS 上，再使用 Spark 對(duì)模型進(jìn)行統(tǒng)計(jì)、分析、輸出報(bào)表數(shù)據(jù)。后續(xù)，訊飛沿用了 Spark 技術(shù)棧引入了 spark-streaming。

隨后逐漸將 spark-streaming 遷移到了 Flink 上，主要是因?yàn)?Flink 更高的時(shí)效性和對(duì)事件時(shí)間的支持。

當(dāng)初 spark-streaming 的實(shí)踐是微批的，一般設(shè)置 10 秒或是 30 秒一批，數(shù)據(jù)的時(shí)效性頂多是秒級(jí)的。而 Flink 可以支持事件驅(qū)動(dòng)的開發(fā)模式，理論上時(shí)效性可以達(dá)到毫秒級(jí)。

當(dāng)初基于 spark-streaming 的實(shí)時(shí)數(shù)據(jù)流邏輯較為簡(jiǎn)陋，沒有形成一個(gè)數(shù)倉分層的結(jié)構(gòu)。而 Flink 可以基于 watermark 支持事件時(shí)間，并且支持對(duì)延遲數(shù)據(jù)的處理，對(duì)于構(gòu)建一個(gè)業(yè)務(wù)邏輯完備的數(shù)倉有很大的幫助。

由上圖可見，ODS 的業(yè)務(wù)日志收集到 Kafka 中，F(xiàn)link 從 Kafka 中消費(fèi)業(yè)務(wù)日志，清洗處理后將業(yè)務(wù)模型再回寫到 Kafka 中。然后再基于 Flink 去消費(fèi) Kafka 中的模型，提取維度和指標(biāo)，統(tǒng)計(jì)后輸出報(bào)表。有些報(bào)表會(huì)直接寫到 sql 或 HBase 中，還有一些報(bào)表會(huì)回寫到 Kafka 中，再由 Druid 從 Kafka 中主動(dòng)攝取這部分報(bào)表數(shù)據(jù)。

在整個(gè)數(shù)據(jù)流圖中 Flink 是核心的計(jì)算引擎，負(fù)責(zé)清洗日志、統(tǒng)計(jì)報(bào)表。

03場(chǎng)景實(shí)踐

3.1 ODS - 日志消費(fèi)負(fù)載均衡

ODS 業(yè)務(wù)中，請(qǐng)求日志量級(jí)大，其他日志量級(jí)小。這樣請(qǐng)求日志（request_topic）在 Kafka 上分區(qū)多，曝光和點(diǎn)擊日志（impress/click_topic）分區(qū)少。

最初是采用單 source 的方法，創(chuàng)建一個(gè) FlinkKafkaConsumer011 消費(fèi)所有分區(qū)，這可能導(dǎo)致 task 消費(fèi)負(fù)載不均。同一 topic 的不同分區(qū)在 task 上可均勻分配，但不同 topic 的分區(qū)可能會(huì)被同一 task 消費(fèi)。期望能達(dá)到的消費(fèi)狀態(tài)是：量級(jí)大的 topic，其 task 和 partition 一一對(duì)應(yīng)，量級(jí)小的 topic 占用剩下的 task。

解決方法是把單 source 的消費(fèi)方式改成了多 source union 的方式，也就是創(chuàng)建了兩個(gè) consumer，一個(gè) consumer 用來消費(fèi)大的 topic，一個(gè) consumer 用來消費(fèi)小的 topic，并單獨(dú)為它們?cè)O(shè)置并行。

3.2 DWD - 日志關(guān)聯(lián)及狀態(tài)緩存

DWD 是業(yè)務(wù)模型層，需要實(shí)現(xiàn)的一個(gè)關(guān)鍵邏輯是日志關(guān)聯(lián)?；?sid 關(guān)聯(lián)廣告一次生命周期中的不同行為日志。業(yè)務(wù)模型記錄了 sid 級(jí)別的維度和指標(biāo)。

最初是基于 30s 的 window 來做關(guān)聯(lián)，但這種方式會(huì)導(dǎo)致模型輸出較第一次事件發(fā)生延遲有 30s，并且 30s 僅能覆蓋不到 12% 的曝光日志。如果擴(kuò)大窗口時(shí)間則會(huì)導(dǎo)致輸出延遲更多，并且同一時(shí)刻存在的窗口隨時(shí)間增長(zhǎng)，資源消耗也比較大。

后續(xù)改成了基于狀態(tài)緩存的方式來實(shí)現(xiàn)日志關(guān)聯(lián)，即 ValueState。同一 sid 下的日志能夠訪問到相應(yīng)的 ValueState。不過為保證及時(shí)輸出，將請(qǐng)求、曝光、點(diǎn)擊等不同指標(biāo)，拆分到了多條數(shù)據(jù)中，輸出的數(shù)據(jù)存在冗余。

隨著業(yè)務(wù)的增長(zhǎng)和變化，需要緩存的狀態(tài)日益變大，內(nèi)存已無法滿足。于是我們將狀態(tài)從內(nèi)存遷移至 HBase 中，這樣做的好處是支持了更大的緩存，并且 Flink checkpoint 負(fù)載降低。但同時(shí)也帶來了兩個(gè)問題：引入第三方服務(wù)，需要額外維護(hù) HBase；HBase 的穩(wěn)定性也成為計(jì)算鏈路穩(wěn)定性的重要依賴。

在 HBase 狀態(tài)緩存中，遇到一個(gè)數(shù)據(jù)傾斜的問題，某條測(cè)試 sid 的曝光重復(fù)上報(bào)，每小時(shí)千次量級(jí)。如上圖，該條 sid 對(duì)應(yīng)的狀態(tài)達(dá)到 MB 級(jí)別，被頻繁的從 HBase 中取出并寫回，引起頻繁的 gc，影響所在 task 的性能。解決辦法是根據(jù)業(yè)務(wù)邏輯對(duì) impress 進(jìn)行去重。

3.3 DWS - 實(shí)時(shí) OLAP

在 DWD 層基于 Flink 的事件驅(qū)動(dòng)已經(jīng)實(shí)現(xiàn)了實(shí)時(shí)模型，再由 Flink 來消費(fèi)處理實(shí)時(shí)模型，從中提取出維度和指標(biāo)，然后逐條的向后輸出。在這個(gè)過程中已是能輸出一個(gè)實(shí)時(shí) OLAP 的結(jié)果了，但也需要有個(gè)后端的存儲(chǔ)來承接，我們因此引入了 Druid。Druid 可以支持?jǐn)?shù)據(jù)的實(shí)時(shí)攝入，并且攝入的結(jié)果實(shí)時(shí)可查，也可以在攝入的同時(shí)做自動(dòng)的聚合。

上圖左側(cè)：每張表需要啟動(dòng)常駐任務(wù)等待 push 過來的數(shù)據(jù)。常駐任務(wù)被動(dòng)接收數(shù)據(jù)，易被壓崩；常駐任務(wù)異常重啟麻煩，需要清理 zk 狀態(tài)；常駐任務(wù)的高可用依賴備份任務(wù)，浪費(fèi)資源。

上圖右側(cè)：一張報(bào)表對(duì)應(yīng)一個(gè) Kafka 消費(fèi)任務(wù)。消費(fèi)任務(wù)自己控制攝入速率更加穩(wěn)定；任務(wù)可依賴 offset 平滑的失敗自啟。

3.4 ADS - 跨源查詢

Presto 是分布式的 SQL 查詢引擎，可從不同的數(shù)據(jù)源抽取數(shù)據(jù)并關(guān)聯(lián)查詢。但會(huì)帶來 Druid 的下推優(yōu)化支持不完善的問題。

3.5 流批混合現(xiàn)狀

如上圖所示是 Lambda 大數(shù)據(jù)框架，流式計(jì)算部分是 Kafka+Flink，批處理則是 HDFS+Spark。

流式計(jì)算的特點(diǎn)：

• 響應(yīng)快，秒級(jí)輸出；
• 可重入性差，難以重復(fù)計(jì)算歷史日志；
• 流的持續(xù)性重要，異常需迅速介入。

批處理的特點(diǎn)：

• 響應(yīng)慢，小時(shí)級(jí)輸出；
• 可重入性好，可重復(fù)計(jì)算歷史數(shù)據(jù)；
• 數(shù)據(jù)按小時(shí)粒度管理，個(gè)別異常可從容處理。

流批混合痛點(diǎn)：

• 兩遍日志清洗的計(jì)算量；
• 兩套技術(shù)框架；
• 數(shù)據(jù)一致性問題。

04未來展望

流批混合優(yōu)化，直接將實(shí)時(shí)模型輸出到 HDFS。

好處是：

• 避免了對(duì)日志的重復(fù)清洗；
• 統(tǒng)一了建模的技術(shù)框架；
• 支持延遲數(shù)據(jù)對(duì)模型的更新。

但也有以下兩個(gè)問題：

• 實(shí)時(shí)模型重復(fù)，量級(jí)更大，計(jì)算消耗大；
• 支持?jǐn)?shù)據(jù)更新的技術(shù)如 Hudi，會(huì)改變模型的使用方式，對(duì)后續(xù)使用者不友好。

最后聊一下對(duì) Flink-SQL 的想法：檢索近 10 分鐘的某條異常日志、快速評(píng)估近 10 分鐘新策略的效果都屬于即時(shí)、微批、即席查詢。批處理鏈路小時(shí)級(jí)響應(yīng)太慢；實(shí)時(shí)檢索系統(tǒng)如 ES，資源消耗大?？梢岳?Kafka + Flink-SQL 解決上述問題，Kafka + Flink-SQL 也是今后計(jì)劃嘗試的方向。?