?大部分數(shù)據(jù)團隊在進行實時業(yè)務建設的初期，都會出現(xiàn)煙囪式開發(fā)、一個任務搞定全部數(shù)據(jù)加工環(huán)節(jié)等問題，缺乏實時數(shù)據(jù)的管理和實時數(shù)倉分層建設的規(guī)范意識。隨著實時場景的進一步豐富，出現(xiàn)了實時數(shù)據(jù)復用、業(yè)務方自助進行實時取數(shù)等需求，因此要求數(shù)據(jù)團隊要像管理離線數(shù)據(jù)一樣對實時數(shù)據(jù)進行有規(guī)范的實時數(shù)倉管理。

本文將從一個實際業(yè)務場景和一個模擬數(shù)倉構建的案例來說明如何利用 EasyData 實時開發(fā)平臺來建設實時數(shù)倉。

1、實際業(yè)務場景

1.1 背景介紹

業(yè)務方是某移動 APP 的運營團隊，需求是要實時監(jiān)控各類運營活動的 ABtest 的實驗效果，以便業(yè)務方根據(jù)實驗效果隨時調(diào)整運營投放策略、投放目標用戶和投放比例。

1.2 業(yè)務數(shù)據(jù)分層

通常業(yè)務方的實時數(shù)據(jù)流轉鏈路包含實時數(shù)據(jù)采集、實時數(shù)據(jù)加工處理、實時數(shù)據(jù)落庫查詢等步驟，在此用戶的ABtest 場景中，數(shù)據(jù)加工鏈路如下圖所示：

第一步：用戶的日志數(shù)據(jù)經(jīng)過實時采集寫入 ODS 層的 Kafka 中。ODS 層數(shù)據(jù)為原始未加工的業(yè)務數(shù)據(jù)，保存在 Kafka，7 天后自動清理。

第二步：ODS 層數(shù)據(jù)經(jīng)過 Flink 任務處理寫入 DWD 層 Kafka 中。DWD 層數(shù)據(jù)為經(jīng)過清洗的數(shù)據(jù)。

第三步：DWD 層數(shù)據(jù)經(jīng)過 Flink 任務處理寫入 DWS 層 KUDU 數(shù)據(jù)庫中落庫。DWS 層數(shù)據(jù)為經(jīng)過聚合、過濾等加工步驟，可以向業(yè)務方提供的數(shù)據(jù)。

第四步：業(yè)務方在需要時通過 Impala 查詢 KUDU 數(shù)據(jù)庫中的數(shù)據(jù)生成報表。

通過以上實時數(shù)據(jù)加工鏈路，業(yè)務方可實現(xiàn)實時報表展示，時效性較離線加工鏈路大大提高，可以滿足業(yè)務方要求數(shù)據(jù)實時更新的需求。

1.3 業(yè)務痛點

在這個業(yè)務場景中 ODS 層、DWD 層的 Kafka 數(shù)據(jù)在其他加工鏈路中也需要被復用，但在其他鏈路加工過程中，同樣的 Topic 需要不斷重復在不同任務中進行 Flink Table 的定義，每次定義用戶均需要使用 DDL 語句定義字段、表配置等，重復工作很多，同時在任務中進行表定義時，數(shù)據(jù)管理者無法感知哪些數(shù)據(jù)已被使用，也無法判斷是否有可以優(yōu)化的數(shù)據(jù)流轉鏈路。

1.4 產(chǎn)品方案

EasyData 實時開發(fā)模塊中為用戶提供了實時流表登記和管理的功能，輔助用戶進行實時數(shù)倉的建設。實時流表是 EasyData 實時開發(fā)模塊中的特有概念。流表的內(nèi)容為 Kafka，Rocketmq 等沒有明確 schema 的消息中間件的元數(shù)據(jù)。在平臺通過登記流表并在任務中直接引用流表的方式即可將這部分元數(shù)據(jù)進行復用。

同時在流表管理模塊中，用戶可以查看流表的定義。此外，按照業(yè)務方的數(shù)倉規(guī)范中的表命名規(guī)范登記流表后，可以根據(jù)流表的表名判斷流表的分層歸屬。在接下來的規(guī)劃中，數(shù)倉流表模塊將支持數(shù)據(jù)血緣查看、數(shù)據(jù)預覽、使用數(shù)據(jù)模型建表等功能，基于流表元數(shù)據(jù)進行更完整更易用的實時數(shù)倉管理。

在下方模擬案例介紹中，將為大家講解如何定義和使用流表，以及如何通過登記流表進行數(shù)倉建設。

2、案例場景介紹

業(yè)務目標：統(tǒng)計某 APP 實時訪問的 DAU，需要統(tǒng)計的值包括總 DAU，各設備類型 DAU（iPhone、華為、OPPO、其他）。

業(yè)務數(shù)據(jù)鏈路：

第一步：通過 CDC 任務采集用戶訪問數(shù)據(jù)數(shù)據(jù)實時變更至 Kafka（對應案例步驟第一步）

第二步：將 ODS 層 Kafka 數(shù)據(jù)通過 Flink 任務進行清洗和聚合，寫入 MySQL 落庫（對應案例步驟第二、三步）

第三步：將 MySQL 數(shù)據(jù)通過 BI 報表展示（對應案例最終結果）

3、案例操作步驟

3.1 準備階段：準備模擬數(shù)據(jù)

3.1.1 數(shù)據(jù)源準備

需準備用于實踐的 MySQL 數(shù)據(jù)源和 Kafka 數(shù)據(jù)源。

3.1.2 數(shù)據(jù)表準備

（1）準備 DS 層源端表：DAU_DS

此表用于記錄用戶訪問數(shù)據(jù)。表結構與樣例數(shù)據(jù)如下：

（2）準備 DWD 層結果表：DAU_FINAL

此表用于統(tǒng)計最終結果。表結構如下：

注意：由于模擬案例最終希望直接展示不同用戶的計算結果，故需要向同一張已提前制作好對應 BI 報表的表內(nèi)寫數(shù)據(jù)，每人更新一行數(shù)據(jù)。正常業(yè)務場景下根據(jù)業(yè)務需求決定結果表結構和數(shù)量。

3.2 第一步：創(chuàng)建 CDC 任務

3.2.1 創(chuàng)建 CDC 任務

任務名稱可自行命名，任務保存位置可選擇根目錄或創(chuàng)建以自己名字命名的目錄。任務環(huán)境和任務類別為任務標簽信息，選擇測試和默認標簽即可，不影響任務實際運行。

3.2.2 編輯 CDC 任務

源端配置：

表：DAU_DS

傳輸起始位點：

若只想消費新增數(shù)據(jù)，請選擇最新數(shù)據(jù)，最終結果報表中將僅有體驗當日的數(shù)據(jù)。

若想先消費歷史存量數(shù)據(jù)，之后再消費最新數(shù)據(jù)，請選擇全量初始化，最終結果報表中將有歷史數(shù)據(jù)。

目標端配置：

類型：kafka

數(shù)據(jù)源：poc_kafka

Topic：自行命名，可通過目標 Topic 生成規(guī)則生成，也可在目標 Topic 中手動修改，建議修改目標 Topic 名稱為自己的名稱，方便下一步新建流表時使用。此處選擇不存在的 Topic，在任務運行后對應 Topic 將被自動創(chuàng)建。

序列化方式：canal-json

3.2.3 保存并一鍵發(fā)布任務

點擊頁面上方的 保存 和 一鍵發(fā)布 按鈕，填寫任意提交描述，將任務發(fā)布至實時運維列表。

成功發(fā)布后可點擊 運維 按鈕前往任務運維頁面。

3.2.4 啟動任務

在運維頁面找到對應任務后點擊 啟動 按鈕啟動任務。

任務成功啟動，任務狀態(tài)變?yōu)檫\行中時，創(chuàng)建 CDC 任務步驟操作完成。

3.3 第二步：創(chuàng)建 ODS 層流表

點擊實時開發(fā)頁面左側目錄第四項流表，打開流表管理頁面。點擊頁面右上角創(chuàng)建表按鈕，開始創(chuàng)建流表。

表名：自行命名

topic：填寫上一步 CDC 任務的目標 Topic 名稱

序列化方式：canal-json

填寫完以上信息后可開始進行字段自動解析。

字段信息獲取方式選擇自動解析，之后點擊獲取數(shù)據(jù)，獲取到數(shù)據(jù)樣例后點擊解析，即可解析出流表的字段信息。

Tip：若 CDC 任務正常運行但此處未獲取到樣例數(shù)據(jù)，可能是因為數(shù)據(jù)暫未寫入，稍等一分鐘后重新嘗試。

字段信息確認無誤后即可保存流表。

保存流表成功，即為此步驟操作完成。

3.4 第三步：創(chuàng)建 SQL 任務

3.4.1 創(chuàng)建 SQL 任務

引擎請選擇 FLINK-1.14。其他類似創(chuàng)建 CDC 任務步驟。

3.4.2 編輯 SQL 代碼

代碼邏輯為：將同一天的用戶方式數(shù)據(jù)按日期聚合，并統(tǒng)計當天的DAU總數(shù)以及各設備類型的DAU。

在代碼中引用流表時，直接使用 [庫].[表] 二元組的寫法即可使用對應的流表。

注意以下內(nèi)容在拷貝代碼后需自行更改：

Kafka消費者組id配置，需要更改配置中的流表名稱為自己的流表名稱

Kafka流表名稱，需要更改為上一步中自己登記的流表名稱

插入結果表的submitter字段值，需要更改為自己的名字

具體要修改的內(nèi)容請見代碼中的標注。

-- 設置Kafka消費者組id，需要更改自己創(chuàng)建的流表名稱, 配置方式：'{流表名稱}.connections.group.id' = '{groupid-name}'set '這里填流表名稱.connections.group.id' = 'example001'; -- 設置讀取消息隊列的位置, 配置方式：'{流表名稱}.scan.startup.mode' = '{earliest-offset / latest-offset}'set '這里填流表名稱.scan.startup.mode' = 'earliest-offset'; -- 設置流表主鍵，配置方式：'{流表名稱}.primary.keys' = '{primary key name}'set '這里填流表名稱.primary.keys' = 'id';--設置源端表表名CREATE view v1 ASSELECT     SUBSTRING(visitTime, 0, 10) AS `date`,     equipment from 這里填流表庫名稱.這里填流表名稱;CREATE view v2 AS SELECT     `date`,    equipment,    COUNT(equipment) AS eq_dauFROM v1GROUP BY `date`, equipment;--將你的名字填入 submitter 字段CREATE view v3 ASSELECT '這里填你的名字' AS submitter,    `date`,    SUM(CASE equipment WHEN 'iPhone' THEN eq_dau ELSE 0 END) AS iPhoneDAU,    SUM(CASE equipment WHEN 'Huawei' THEN eq_dau ELSE 0 END) AS HuaweiDAU,    SUM(CASE equipment WHEN 'OPPO' THEN eq_dau ELSE 0 END) AS oppoDAU,    SUM(CASE equipment WHEN 'other' THEN eq_dau ELSE 0 END) AS otherDAUFROM v2 GROUP BY `date`;CREATE VIEW v4 ASSELECT     (v3.iPhoneDAU + v3.HuaweiDAU + v3.oppoDAU + v3.otherDAU) AS totalDAU,     submitter,     `date`,     iPhoneDAU,     HuaweiDAU,     oppoDAU,     otherDAUFROM v3;set 'DAU_FINAL.primary.keys' = 'submitter,date';INSERT INTO MySQL數(shù)據(jù)源標識.MySQL數(shù)據(jù)庫名稱.DAU_FINALSELECT * FROM v4;

3.4.3 發(fā)布 SQL 任務并啟動任務

保存并發(fā)布 SQL 任務，并啟動任務。操作方法與 CDC 任務的發(fā)布和啟動相同。

SQL 任務成功啟動且狀態(tài)變?yōu)檫\行中，則此步驟操作完成。

3.4.4 創(chuàng)建 BI 報表并展示數(shù)據(jù)結果

在有數(shù) BI 中創(chuàng)建對應報表，查看最終的統(tǒng)計結果即可。刷新報表數(shù)據(jù)即可看到報表數(shù)據(jù)實時更新后的結果。預期效果如下：

（1）折線圖：

（2）報表