隨著互聯(lián)網(wǎng)的發(fā)展進入下半場，數(shù)據(jù)的時效性對企業(yè)的精細化運營越來越重要， 商場如戰(zhàn)場，在每天產(chǎn)生的海量數(shù)據(jù)中，如何能實時有效的挖掘出有價值的信息， 對企業(yè)的決策運營策略調(diào)整有很大幫助。

此外，隨著 5G 技術(shù)的成熟、廣泛應(yīng)用， 對于工業(yè)互聯(lián)網(wǎng)、物聯(lián)網(wǎng)等數(shù)據(jù)時效性要求非常高的行業(yè)，企業(yè)就更需要一套完整成熟的實時數(shù)據(jù)體系來提高自身的行業(yè)競爭力。

本文從上述現(xiàn)狀及實時數(shù)據(jù)需求出發(fā)，結(jié)合工業(yè)界案例、筆者的實時數(shù)據(jù)開發(fā)經(jīng)驗， 梳理總結(jié)了實時數(shù)據(jù)體系建設(shè)的總體方案，本文主要分為三個部分：

• 第一部分主要介紹了當下在工業(yè)界比較火熱的實時計算引擎 Flink 在實時數(shù)據(jù)體系建設(shè)過程中主要的應(yīng)用場景及對應(yīng)解決方案;
• 第二部分從實時數(shù)據(jù)體系架構(gòu)、實時數(shù)據(jù)模型分層、實時數(shù)據(jù)體系建設(shè)方式、流批一體實時數(shù)據(jù)架構(gòu)發(fā)展等四個方面思考了實時數(shù)據(jù)體系的建設(shè)方案;
• 第三部分則以一個具體案例介紹如何使用 Flink SQL 完成實時數(shù)據(jù)統(tǒng)計類需求。

一、Flink 實時應(yīng)用場景

目前看來，F(xiàn)link 在實時計算領(lǐng)域內(nèi)的主要應(yīng)用場景主要可分為四類場景， 分別是實時數(shù)據(jù)同步、流式 ETL、實時數(shù)據(jù)分析和復(fù)雜事件處理，具體的業(yè)務(wù)場景和對應(yīng)的解決方案可詳細研究下圖， 文字層面不再詳述。 

二、實時數(shù)據(jù)體系架構(gòu)

實時數(shù)據(jù)體系大致分為三類場景：流量類、業(yè)務(wù)類和特征類，這三種場景各有不同。

• 在數(shù)據(jù)模型上，流量類是扁平化的寬表，業(yè)務(wù)數(shù)倉更多是基于范式的建模，特征數(shù)據(jù)是 KV 存儲;
• 從數(shù)據(jù)來源區(qū)分，流量數(shù)倉的數(shù)據(jù)來源一般是日志數(shù)據(jù)，業(yè)務(wù)數(shù)倉的數(shù)據(jù)來源是業(yè)務(wù) binlog 數(shù)據(jù)，特征數(shù)倉的數(shù)據(jù)來源則多種多樣;
• 從數(shù)據(jù)量而言，流量和特征數(shù)倉都是海量數(shù)據(jù)，每天十億級以上，而業(yè)務(wù)數(shù)倉的數(shù)據(jù)量一般每天百萬到千萬級;
• 從數(shù)據(jù)更新頻率而言，流量數(shù)據(jù)極少更新，則業(yè)務(wù)和特征數(shù)據(jù)更新較多，流量數(shù)據(jù)一般關(guān)注時序和趨勢，業(yè)務(wù)數(shù)據(jù)和特征數(shù)據(jù)關(guān)注狀態(tài)變更;

在數(shù)據(jù)準確性上，流量數(shù)據(jù)要求較低，而業(yè)務(wù)數(shù)據(jù)和特征數(shù)據(jù)要求較高。

1、實時數(shù)據(jù)體系整體架構(gòu) 

整個實時數(shù)據(jù)體系架構(gòu)分為五層，分別是接入層，存儲層，計算層、平臺層和應(yīng)用層，上圖只是整體架構(gòu)的概要圖，每一層具體要做的事情，接下來通過文字來詳述。

1)接入層：該層利用各種數(shù)據(jù)接入工具收集各個系統(tǒng)的數(shù)據(jù)，包括 binlog 日志、埋點日志、以及后端服務(wù)日志，數(shù)據(jù)會被收集到 Kafka 中;這些數(shù)據(jù)不只是參與實時計算，也會參與離線計算，保證實時和離線的原始數(shù)據(jù)是統(tǒng)一的;

2)存儲層：該層對原始數(shù)據(jù)、清洗關(guān)聯(lián)后的明細數(shù)據(jù)進行存儲，基于統(tǒng)一的實時數(shù)據(jù)模型分層理念，將不同應(yīng)用場景的數(shù)據(jù)分別存儲在 Kafka、HDFS、Kudu、 Clickhouse、Hbase、Redis、Mysql 等存儲引擎中，各種存儲引擎存放的具體的數(shù)據(jù)類型在實時數(shù)據(jù)模型分層部分會詳細介紹;

3)計算層：計算層主要使用 Flink、Spark、Presto 以及 ClickHouse 自帶的計算能力等四種計算引擎，F(xiàn)link 計算引擎主要用于實時數(shù)據(jù)同步、 流式 ETL、關(guān)鍵系統(tǒng)秒級實時指標計算場景，Spark SQL 主要用于復(fù)雜多維分析的準實時指標計算需求場景，Presto 和 ClickHouse 主要滿足多維自助分析、對查詢響應(yīng)時間要求不太高的場景;

4)平臺層：在平臺層主要做三個方面的工作，分別是對外提供統(tǒng)一查詢服務(wù)、元數(shù)據(jù)及指標管理、數(shù)據(jù)質(zhì)量及血緣;

5)應(yīng)用層：以統(tǒng)一查詢服務(wù)對各個業(yè)務(wù)線數(shù)據(jù)場景進行支持，業(yè)務(wù)主要包括實時大屏、實時數(shù)據(jù)產(chǎn)品、實時 OLAP、實時特征等。

其中，平臺層詳細工作如下：

• 統(tǒng)一查詢服務(wù)支持從底層明細數(shù)據(jù)到聚合層數(shù)據(jù)的查詢，支持以SQL化方式查詢Redis、Hbase等KV存儲中的數(shù)據(jù);
• 元數(shù)據(jù)及指標管理：主要對實時的Kafka表、Kudu表、Clickhouse表、Hive表等進行統(tǒng)一管理，以數(shù)倉模型中表的命名方式規(guī)范表的命名，明確每張表的字段含義、使用方，指標管理則是盡量通過指標管理系統(tǒng)將所有的實時指標統(tǒng)一管理起來，明確計算口徑，提供給不同的業(yè)務(wù)方使用;
• 數(shù)據(jù)質(zhì)量及血緣分析：數(shù)據(jù)質(zhì)量分為平臺監(jiān)控和數(shù)據(jù)監(jiān)控兩個部分，血緣分析則主要是對實時數(shù)據(jù)依賴關(guān)系、實時任務(wù)的依賴關(guān)系進行分析。
• 平臺監(jiān)控部分一是對任務(wù)運行狀態(tài)進行監(jiān)控，對異常的任務(wù)進行報警并根據(jù)設(shè)定的參數(shù)對任務(wù)進行自動拉起與恢復(fù)，二是針對 Flink 任務(wù)要對 Kafka 消費處理延遲進行監(jiān)控并實時報警。

數(shù)據(jù)據(jù)監(jiān)控則分為兩個部分：

• 首先流式 ETL 是整個實時數(shù)據(jù)流轉(zhuǎn)過程中重要的一環(huán)，ETL 的過程中會關(guān)聯(lián)各種維表，實時關(guān)聯(lián)時，定時對沒有關(guān)聯(lián)上的記錄上報異常日志到監(jiān)控平臺，當數(shù)量達到一定閾值時觸發(fā)報警;
• 其次，部分關(guān)鍵實時指標采用了 lambda 架構(gòu)，因此需要對歷史的實時指標與離線 hive 計算的數(shù)據(jù)定時做對比，提供實時數(shù)據(jù)的數(shù)據(jù)質(zhì)量監(jiān)控，對超過閾值的指標數(shù)據(jù)進行報警。

為了配合數(shù)據(jù)監(jiān)控，需要做實時數(shù)據(jù)血緣，主要是梳理實時數(shù)據(jù)體系中數(shù)據(jù)依賴關(guān)系，以及實時任務(wù)的依賴關(guān)系，從底層ODS 到 DW 再到 DM，以及 DM 層被哪些模型用到， 將整個鏈條串聯(lián)起來，這樣做在數(shù)據(jù)/任務(wù)主動調(diào)整時可以通知關(guān)聯(lián)的下游，指標異常時借助血緣定位問題，同時基于血緣關(guān)系的分析，我們也能評估數(shù)據(jù)的應(yīng)用價值，核算數(shù)據(jù)的計算成本。

2、實時數(shù)據(jù)模型分層 

離線數(shù)倉考慮到效率問題，一般會采取空間換時間的方式，層級劃分會比較多;實時數(shù)倉考慮到實時性問題，分層則越少越好，另外也減少了中間流程出錯的可能性，因此將其分為四層。

1)ODS 層

操作數(shù)據(jù)層，保存原始數(shù)據(jù)，對非結(jié)構(gòu)化的數(shù)據(jù)進行結(jié)構(gòu)化處理，輕度清洗，幾乎不刪除原始數(shù)據(jù)。

該層的數(shù)據(jù)主要來自業(yè)務(wù)數(shù)據(jù)庫的 binlog 日志、埋點日志和應(yīng)用程序日志。

對于 binlog 日志通過 canal 監(jiān)聽，寫到消息隊列 Kafka 中，對應(yīng)于埋點和應(yīng)用程序日志，則通過 Filebeat 采集 nginx 和 tomcat 日志，上報到Kafka 中。

除了存儲在 Kafka 中，同時也會對業(yè)務(wù)數(shù)據(jù)庫的 binlog 日志通過 Flink 寫入 HDFS、Kudu 等存儲引擎，落地到 5min Hive 表，供查詢明細數(shù)據(jù)，同時也提供給離線數(shù)倉，做為其原始數(shù)據(jù);另外，對于埋點日志數(shù)據(jù)，由于 ODS 層是非結(jié)構(gòu)化的，則沒有必要落地。

2)DWD 層

實時明細數(shù)據(jù)層，以業(yè)務(wù)過程作為建模驅(qū)動，基于每個具體的業(yè)務(wù)過程特點，構(gòu)建最細粒度的明細層事實表;可以結(jié)合企業(yè)的數(shù)據(jù)使用特點，將明細事實表的某些重要維度屬性字段做適當冗余，也即寬表化處理。

該層的數(shù)據(jù)來源于 ODS 層，通過簡單的 Streaming ETL 后得到，對于 binlog 日志的處理主要進行簡單的數(shù)據(jù)清洗、處理數(shù)據(jù)漂移，以及可能對多個 ODS 層的表進行 Streaming Join，對流量日志主要是做一些通用ETL 處理，將非結(jié)構(gòu)化的數(shù)據(jù)結(jié)構(gòu)化，關(guān)聯(lián)通用的維度字段。

該層的數(shù)據(jù)存儲在消息隊列 Kafka 中，同時也會用 Flink 實時寫入 Hive 5min 表，供查詢明細數(shù)據(jù)，同時要提供給離線數(shù)倉，做為其原始數(shù)據(jù)。

3)DIM 層

公共維度層，基于維度建模理念思想，建立整個業(yè)務(wù)過程的一致性維度，降低數(shù)據(jù)計算口徑和算法不統(tǒng)一風險。

DIM 層數(shù)據(jù)來源于兩部分：一部分是Flink程序?qū)崟r處理ODS層數(shù)據(jù)得到，另外一部分是通過離線任務(wù)出倉得到。

DIM 層維度數(shù)據(jù)主要使用 MySQL、Hbase、Redis 三種存儲引擎，對于維表數(shù)據(jù)比較少的情況可以使用 MySQL，對于單條數(shù)據(jù)大小比較小，查詢 QPS 比較高的情況，可以使用 Redis 存儲，降低機器內(nèi)存資源占用，對于數(shù)據(jù)量比較大，對維表數(shù)據(jù)變化不是特別敏感的場景，可以使用HBase 存儲。

4)DM 層

①數(shù)據(jù)集市層

以數(shù)據(jù)域+業(yè)務(wù)域的理念建設(shè)公共匯總層，對于DM層比較復(fù)雜，需要綜合考慮對于數(shù)據(jù)落地的要求以及具體的查詢引擎來選擇不同的存儲方式，分為輕度匯總層和高度匯總層，同時產(chǎn)出，高度匯總層數(shù)據(jù)用于前端比較簡單的KV查詢， 提升查詢性能，比如實時大屏，實時報表等，數(shù)據(jù)的時效性要求為秒級，輕度匯總層Kafka中寬表實時寫入OLAP存儲引擎，用于前端產(chǎn)品復(fù)雜的OLAP查詢場景，滿足自助分析和產(chǎn)出復(fù)雜報表的需求，對數(shù)據(jù)的時效性要求可容忍到分鐘級;

②輕度匯總層

輕度匯總層由明細層通過Streaming ETL得到，主要以寬表的形式存在，業(yè)務(wù)明細匯總是由業(yè)務(wù)事實明細表和維度表join得到，流量明細匯總是由流量日志按業(yè)務(wù)線拆分和維度表join得到。

輕度匯總層數(shù)據(jù)存儲比較多樣化，首先利用Flink實時消費DWD層Kafka中明細數(shù)據(jù)join業(yè)務(wù)過程需要的維表，實時打?qū)捄髮懭朐搶拥腒afka中，以Json或PB格式存儲。

同時對多維業(yè)務(wù)明細匯總數(shù)據(jù)通過Flink實時寫入Kudu，用于查詢明細數(shù)據(jù)和更復(fù)雜的多維數(shù)據(jù)分析需求，對于流量數(shù)據(jù)通過Flink分別寫入HDFS和ClickHouse用于復(fù)雜的多維數(shù)據(jù)分析， 實時特征數(shù)據(jù)則通過Flink join維表后實時寫入HDFS，用于下游的離線ETL消費。

對于落地Kudu和HDFS的寬表數(shù)據(jù)，可用Spark SQL做分鐘級的預(yù)計算，滿足業(yè)務(wù)方復(fù)雜數(shù)據(jù)分析需求，提供分鐘級延遲的數(shù)據(jù)，從而加速離線ETL過程的延遲， 另外隨著Flink SQL與Hive生態(tài)集成的不斷完善，可嘗試用Flink SQL做離線ETL和OLAP計算任務(wù)(Flink流計算基于內(nèi)存計算的特性，和presto非常類似，這使其也可以成為一個OLAP計算引擎)，用一套計算引擎解決實時離線需求，從而實現(xiàn)批流統(tǒng)一。

對于Kudu中的業(yè)務(wù)明細數(shù)據(jù)、ClickHouse中的流量明細數(shù)據(jù)，也可以滿足業(yè)務(wù)方的個性化數(shù)據(jù)分析需求，利用強大的OLAP計算引擎，實時查詢明細數(shù)據(jù)，在10s量級的響應(yīng)時間內(nèi)給出結(jié)果，這類需求也即是實時OLAP需求，靈活性比較高。

③高度匯總層

高度匯總層由明細數(shù)據(jù)層或輕度匯總層通過聚合計算后寫入到存儲引擎中，產(chǎn)出一部分實時數(shù)據(jù)指標需求，靈活性比較差。

計算引擎使用Flink Datastream API和Flink SQL，指標存儲引擎根據(jù)不同的需求，對于常見的簡單指標匯總模型可直接放在MySQL里面，維度比較多的、寫入更新比較大的模型會放在HBase里面， 還有一種是需要做排序、對查詢QPS、響應(yīng)時間要求非常高、且不需要持久化存儲如大促活動期間在線TopN商品等直接存儲在Redis里面。

在秒級指標需求中，需要混用Lambda和Kappa架構(gòu)，大部分實時指標使用Kappa架構(gòu)完成計算，少量關(guān)鍵指標(如金額相關(guān))使用Lambda架構(gòu)用批處理重新處理計算，增加一次校對過程。

總體來說 DM 層對外提供三種時效性的數(shù)據(jù)：

• 首先是 Flink 等實時計算引擎預(yù)計算好的秒級實時指標，這種需求對數(shù)據(jù)的時效性要求非常高，用于實時大屏、計算維度不復(fù)雜的實時報表需求。
• 其次是 Spark SQL 預(yù)計算的延遲在分鐘級的準實時指標， 該類指標滿足一些比較復(fù)雜但對數(shù)據(jù)時效性要求不太高的數(shù)據(jù)分析場景，可能會涉及到多個事實表的join，如銷售歸因等需求。
• 最后一種則是不需要預(yù)計算，ad-hoc查詢的復(fù)雜多維數(shù)據(jù)分析場景，此類需求比較個性化，靈活性比較高，如果 OLAP 計算引擎性能足夠強大，也可完全滿足秒級計算需求的場景; 對外提供的秒級實時數(shù)據(jù)和另外兩種準實時數(shù)據(jù)的比例大致為 3：7，絕大多數(shù)的業(yè)務(wù)需求都優(yōu)先考慮準實時計算或 ad-hoc 方式，可以降低資源使用、提升數(shù)據(jù)準確性，以更靈活的方式滿足復(fù)雜的業(yè)務(wù)場景。

3、實時數(shù)據(jù)體系建設(shè)方式

整個實時數(shù)據(jù)體系分為兩種建設(shè)方式，即實時和準實時(它們的實現(xiàn)方式分別是基于流計算引擎和 ETL、OLAP 引擎，數(shù)據(jù)時效性則分別是秒級和分鐘級。

1)在調(diào)度開銷方面，準實時數(shù)據(jù)是批處理過程，因此仍然需要調(diào)度系統(tǒng)支持，調(diào)度頻率較高，而實時數(shù)據(jù)卻沒有調(diào)度開銷。

2)在業(yè)務(wù)靈活性方面，因為準實時數(shù)據(jù)是基于 ETL 或 OLAP 引擎實現(xiàn)，靈活性優(yōu)于基于流計算的方式。

3)在對數(shù)據(jù)晚到的容忍度方面，因為準實時數(shù)據(jù)可以基于一個周期內(nèi)的數(shù)據(jù)進行全量計算，因此對于數(shù)據(jù)晚到的容忍度也是比較高的，而實時數(shù)據(jù)使用的是增量計算，對于數(shù)據(jù)晚到的容忍度更低一些。

4)在適用場景方面，準實時數(shù)據(jù)主要用于有實時性要求但不太高、涉及多表關(guān)聯(lián)和業(yè)務(wù)變更頻繁的場景，如交易類型的實時分析，實時數(shù)據(jù)則更適用于實時性要求高、數(shù)據(jù)量大的場景，如實時特征、流量類型實時分析等場景。

4、流批一體實時數(shù)據(jù)架構(gòu)發(fā)展

從1990年 Inmon 提出數(shù)據(jù)倉庫概念到今天，大數(shù)據(jù)架構(gòu)經(jīng)歷了從最初的離線大數(shù)據(jù)架構(gòu)、Lambda 架構(gòu)、Kappa 架構(gòu)以及 Flink 的火熱帶出的流批一體架構(gòu)，數(shù)據(jù)架構(gòu)技術(shù)不斷演進，本質(zhì)是在往流批一體的方向發(fā)展，讓用戶能以最自然、最小的成本完成實時計算。

1)離線大數(shù)據(jù)架構(gòu)：數(shù)據(jù)源通過離線的方式導(dǎo)入到離線數(shù)倉中，下游應(yīng)用根據(jù)業(yè)務(wù)需求選擇直接讀取 DM 或加一層數(shù)據(jù)服務(wù)，比如 MySQL 或 Redis，數(shù)據(jù)存儲引擎是 HDFS/Hive，ETL 工具可以是 MapReduce 腳本或 HiveSQL。數(shù)據(jù)倉庫從模型層面分為操作數(shù)據(jù)層 ODS、數(shù)據(jù)倉庫明細層 DWD、數(shù)據(jù)集市層 DM。

2)Lambda 架構(gòu)：隨著大數(shù)據(jù)應(yīng)用的發(fā)展，人們逐漸對系統(tǒng)的實時性提出了要求，為了計算一些實時指標，就在原來離線數(shù)倉的基礎(chǔ)上增加了一個實時計算的鏈路，并對數(shù)據(jù)源做流式改造(即把數(shù)據(jù)發(fā)送到消息隊列)，實時計算去訂閱消息隊列，直接完成指標增量的計算，推送到下游的數(shù)據(jù)服務(wù)中去，由數(shù)據(jù)服務(wù)層完成離線&實時結(jié)果的合并。

3)Kappa 架構(gòu)：Lambda 架構(gòu)雖然滿足了實時的需求，但帶來了更多的開發(fā)與運維工作，其架構(gòu)背景是流處理引擎還不完善，流處理的結(jié)果只作為臨時的、近似的值提供參考。后來隨著 Flink 等流處理引擎的出現(xiàn)，流處理技術(shù)成熟起來，這時為了解決兩套代碼的問題，LickedIn 的 Jay Kreps 提出了 Kappa 架構(gòu)。

4)流批一體架構(gòu)：流批一體架構(gòu)比較完美的實現(xiàn)方式是采用流計算 + 交互式分析雙引擎架構(gòu)，在這個架構(gòu)中，流計算負責的是基礎(chǔ)數(shù)據(jù)，而交互式分析引擎是中心，流計算引擎對數(shù)據(jù)進行實時 ETL 工作，與離線相比，降低了 ETL 過程的 latency，交互式分析引擎則自帶存儲，通過計算存儲的協(xié)同優(yōu)化， 實現(xiàn)高寫入 TPS、高查詢 QPS 和低查詢 latency ，從而做到全鏈路的實時化和 SQL 化，這樣就可以用批的方式實現(xiàn)實時分析和按需分析，并能快速的響應(yīng)業(yè)務(wù)的變化，兩者配合，實現(xiàn) 1 + 1 > 2 的效果;該架構(gòu)對交互式分析引擎的要求非常高，也許是未來大數(shù)據(jù)庫技術(shù)發(fā)展的一個重點和方向。

為了應(yīng)對業(yè)務(wù)方更復(fù)雜的多維實時數(shù)據(jù)分析需求，筆者目前在數(shù)據(jù)開發(fā)中引入 Kudu這個 OLAP 存儲引擎，對訂單等業(yè)務(wù)數(shù)據(jù)使用 Presto + Kudu 的計算方案也是在探索流批一體架構(gòu)在實時數(shù)據(jù)分析領(lǐng)域的可行性。此外，目前比較熱的數(shù)據(jù)湖技術(shù)，如 Delta lake、Hudi 等支持在 HDFS 上進行 upsert 更新，隨著其流式寫入、SQL 引擎支持的成熟，未來可以用一套存儲引擎解決實時、離線數(shù)據(jù)需求，從而減少多引擎運維開發(fā)成本。

三、Flink SQL 實時計算 UV 指標

上一部分從宏觀層面介紹了如何建設(shè)實時數(shù)據(jù)體系，非常不接地氣，可能大家需要的只是一個具體的 case 來了解一下該怎么做，那么接下來用一個接地氣的案例來介紹如何實時計算 UV 數(shù)據(jù)。

大家都知道，在 ToC 的互聯(lián)網(wǎng)公司，UV 是一個很重要的指標，對于老板、商務(wù)、運營的及時決策會產(chǎn)生很大的影響，筆者在電商公司，目前主要的工作就是計算 UV、銷售等各類實時數(shù)據(jù)，體驗就特別深刻， 因此就用一個簡單demo 演示如何用 Flink SQL 消費 Kafka 中的 PV 數(shù)據(jù)，實時計算出 UV 指標后寫入 Hbase。

1、Kafka 源數(shù)據(jù)解析

PV 數(shù)據(jù)來源于埋點數(shù)據(jù)經(jīng) FileBeat 上報清洗后，以 ProtoBuffer 格式寫入下游 Kafka，消費時第一步要先反序列化 PB 格式的數(shù)據(jù)為 Flink 能識別的 Row 類型，因此也就需要自定義實現(xiàn) DeserializationSchema 接口，具體如下代碼， 這里只抽取計算用到的 PV 的 mid、事件時間 time_local，并從其解析得到 log_date 字段： 

public class PageViewDeserializationSchema implements DeserializationSchema {  
public static final Logger LOG = LoggerFactory.getLogger(PageViewDeserializationSchema.class);  
protected SimpleDateFormat dayFormatter;  
private final RowTypeInfo rowTypeInfo;  
public PageViewDeserializationSchema(RowTypeInfo rowTypeInfo){  
dayFormatter = new SimpleDateFormat("yyyyMMdd", Locale.UK);  
this.rowTypeInfo = rowTypeInfo;  
} 
 
@Override 
 
public Row deserialize(byte[] message) throws IOException {  
Row row = new Row(rowTypeInfo.getArity());  
MobilePage mobilePage = null;  
try { 
 
mobilePage = MobilePage.parseFrom(message);  
String mid = mobilePage.getMid();  
row.setField(0, mid);  
Long timeLocal = mobilePage.getTimeLocal();  
String logDate = dayFormatter.format(timeLocal);  
row.setField(1, logDate);  
row.setField(2, timeLocal);  
}catch (Exception e){  
String mobilePageError = (mobilePage != null) ? mobilePage.toString() : "";  
LOG.error("error parse bytes payload is {}, pageview error is {}", message.toString(), mobilePageError, e);  
} 
 return null;  
} 

2、編寫 Flink Job 主程序

將 PV 數(shù)據(jù)解析為 Flink 的 Row 類型后，接下來就很簡單了，編寫主函數(shù)，寫 SQL 就能統(tǒng)計 UV 指標了，代碼如下： 

public class RealtimeUV {  
public static void main(String[] args) throws Exception {  
//step1 從properties配置文件中解析出需要的Kakfa、Hbase配置信息、checkpoint參數(shù)信息  
Map config = PropertiesUtil.loadConfFromFile(args[0]);  
String topic = config.get("source.kafka.topic");  
String groupId = config.get("source.group.id");  
String sourceBootStrapServers = config.get("source.bootstrap.servers");  
String hbaseTable = config.get("hbase.table.name");  
String hbaseZkQuorum = config.get("hbase.zk.quorum");  
String hbaseZkParent = config.get("hbase.zk.parent");  
int checkPointPeriod = Integer.parseInt(config.get("checkpoint.period"));  
int checkPointTimeout = Integer.parseInt(config.get("checkpoint.timeout"));  
StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment(); 
 
//step2 設(shè)置Checkpoint相關(guān)參數(shù)，用于Failover容錯 
 
sEnv.getConfig().registerTypeWithKryoSerializer(MobilePage.class, 
 
ProtobufSerializer.class);  
sEnv.getCheckpointConfig().setFailOnCheckpointingErrors(false);  
sEnv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);  
sEnv.enableCheckpointing(checkPointPeriod,CheckpointingMode.EXACTLY_ONCE);  
sEnv.getCheckpointConfig().setCheckpointTimeout(checkPointTimeout);  
sEnv.getCheckpointConfig().enableExternalizedCheckpoints(  
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); 
 
//step3 使用Blink planner、創(chuàng)建TableEnvironment,并且設(shè)置狀態(tài)過期時間，避免Job OOM 
 
EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance()  
.useBlinkPlanner()  
.inStreamingMode()  
.build();  
StreamTableEnvironment tEnv = StreamTableEnvironment.create(sEnv, environmentSettings); 
tEnv.getConfig().setIdleStateRetentionTime(Time.days(1), Time.days(2));  
Properties sourceProperties = new Properties(); 
 sourceProperties.setProperty("bootstrap.servers", sourceBootStrapServers);  
sourceProperties.setProperty("auto.commit.interval.ms", "3000");  
sourceProperties.setProperty("group.id", groupId); 
 
//step4 初始化KafkaTableSource的Schema信息，筆者這里使用register TableSource的方式將源表注冊到Flink中，而沒有用register DataStream方式，也是因為想熟悉一下如何注冊KafkaTableSource到Flink中 
 
TableSchema schema = TableSchemaUtil.getAppPageViewTableSchema();  
Optional proctimeAttribute = Optional.empty();  
List rowtimeAttributeDescriptors = Collections.emptyList();  
Map fieldMapping = new HashMap<>();  
List columnNames = new ArrayList<>(); 
 
RowTypeInfo rowTypeInfo = new RowTypeInfo(schema.getFieldTypes(), schema.getFieldNames()); 
 
columnNames.addAll(Arrays.asList(schema.getFieldNames()));  
columnNames.forEach(name -> fieldMapping.put(name, name));  
PageViewDeserializationSchema deserializationSchema = new  
PageViewDeserializationSchema(  
rowTypeInfo);  
Map specificOffsets = new HashMap<>();  
Kafka011TableSource kafkaTableSource = new Kafka011TableSource( 
 
schema, 
 
proctimeAttribute,  
rowtimeAttributeDescriptors,  
Optional.of(fieldMapping),  
topic, 
 
sourceProperties,  
deserializationSchema,  
StartupMode.EARLIEST,  
specificOffsets);  
tEnv.registerTableSource("pageview", kafkaTableSource); 
 
//step5 初始化Hbase TableSchema、寫入?yún)?shù)，并將其注冊到Flink中 
 
HBaseTableSchema hBaseTableSchema = new HBaseTableSchema(); 
hBaseTableSchema.setRowKey("log_date", String.class);  
hBaseTableSchema.addColumn("f", "UV", Long.class);  
HBaseOptions hBaseOptions = HBaseOptions.builder()  
.setTableName(hbaseTable)  
.setZkQuorum(hbaseZkQuorum)  
.setZkNodeParent(hbaseZkParent)  
.build(); 
 
HBaseWriteOptions hBaseWriteOptions = HBaseWriteOptions.builder() 
.setBufferFlushMaxRows(1000)  
.setBufferFlushIntervalMillis(1000)  
.build();  
HBaseUpsertTableSink hBaseSink = new HBaseUpsertTableSink(hBaseTableSchema, hBaseOptions, hBaseWriteOptions);  
tEnv.registerTableSink("uv_index", hBaseSink); 
 
//step6 實時計算當天UV指標sql, 這里使用最簡單的group by agg，沒有使用minibatch或窗口，在大數(shù)據(jù)量優(yōu)化時最好使用后兩種方式 
 
String uvQuery = "insert into uv_index "  
+ "select log_date,\n"  
+ "ROW(count(distinct mid) as UV)\n"  
+ "from pageview\n"  
+ "group by log_date";  
tEnv.sqlUpdate(uvQuery);  
//step7 執(zhí)行Job  
sEnv.execute("UV Job");  
}  
} 

以上就是一個簡單的使用 Flink SQL 統(tǒng)計 UV 的 case, 代碼非常簡單，只需要理清楚如何解析 Kafka 中數(shù)據(jù)，如何初始化 Table Schema，以及如何將表注冊到 Flink中，即可使用 Flink SQL 完成各種復(fù)雜的實時數(shù)據(jù)統(tǒng)計類的業(yè)務(wù)需求，學(xué)習(xí)成本比API 的方式低很多。

說明一下，筆者這個 demo 是基于目前業(yè)務(wù)場景而開發(fā)的，在生產(chǎn)環(huán)境中可以真實運行起來，可能不能拆箱即用，你需要結(jié)合自己的業(yè)務(wù)場景自定義相應(yīng)的 kafka 數(shù)據(jù)解析類。