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

本文將從一個(gè)實(shí)際業(yè)務(wù)場(chǎng)景和一個(gè)模擬數(shù)倉(cāng)構(gòu)建的案例來(lái)說(shuō)明如何利用 EasyData 實(shí)時(shí)開(kāi)發(fā)平臺(tái)來(lái)建設(shè)實(shí)時(shí)數(shù)倉(cāng)。

1、實(shí)際業(yè)務(wù)場(chǎng)景

1.1 背景介紹

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

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

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

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

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

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

第四步：業(yè)務(wù)方在需要時(shí)通過(guò) Impala 查詢(xún) KUDU 數(shù)據(jù)庫(kù)中的數(shù)據(jù)生成報(bào)表。

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

1.3 業(yè)務(wù)痛點(diǎn)

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

1.4 產(chǎn)品方案

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

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

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

2、案例場(chǎng)景介紹

業(yè)務(wù)目標(biāo)：統(tǒng)計(jì)某 APP 實(shí)時(shí)訪(fǎng)問(wèn)的 DAU，需要統(tǒng)計(jì)的值包括總 DAU，各設(shè)備類(lèi)型 DAU（iPhone、華為、OPPO、其他）。

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

第一步：通過(guò) CDC 任務(wù)采集用戶(hù)訪(fǎng)問(wèn)數(shù)據(jù)數(shù)據(jù)實(shí)時(shí)變更至 Kafka（對(duì)應(yīng)案例步驟第一步）

第二步：將 ODS 層 Kafka 數(shù)據(jù)通過(guò) Flink 任務(wù)進(jìn)行清洗和聚合，寫(xiě)入 MySQL 落庫(kù)（對(duì)應(yīng)案例步驟第二、三步）

第三步：將 MySQL 數(shù)據(jù)通過(guò) BI 報(bào)表展示（對(duì)應(yīng)案例最終結(jié)果）

3、案例操作步驟

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

3.1.1 數(shù)據(jù)源準(zhǔn)備

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

3.1.2 數(shù)據(jù)表準(zhǔn)備

（1）準(zhǔn)備 DS 層源端表：DAU_DS

此表用于記錄用戶(hù)訪(fǎng)問(wèn)數(shù)據(jù)。表結(jié)構(gòu)與樣例數(shù)據(jù)如下：

（2）準(zhǔn)備 DWD 層結(jié)果表：DAU_FINAL

此表用于統(tǒng)計(jì)最終結(jié)果。表結(jié)構(gòu)如下：

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

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

3.2.1 創(chuàng)建 CDC 任務(wù)

任務(wù)名稱(chēng)可自行命名，任務(wù)保存位置可選擇根目錄或創(chuàng)建以自己名字命名的目錄。任務(wù)環(huán)境和任務(wù)類(lèi)別為任務(wù)標(biāo)簽信息，選擇測(cè)試和默認(rèn)標(biāo)簽即可，不影響任務(wù)實(shí)際運(yùn)行。

3.2.2 編輯 CDC 任務(wù)

源端配置：

表：DAU_DS

傳輸起始位點(diǎn)：

若只想消費(fèi)新增數(shù)據(jù)，請(qǐng)選擇最新數(shù)據(jù)，最終結(jié)果報(bào)表中將僅有體驗(yàn)當(dāng)日的數(shù)據(jù)。

若想先消費(fèi)歷史存量數(shù)據(jù)，之后再消費(fèi)最新數(shù)據(jù)，請(qǐng)選擇全量初始化，最終結(jié)果報(bào)表中將有歷史數(shù)據(jù)。

目標(biāo)端配置：

類(lèi)型：kafka

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

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

序列化方式：canal-json

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

點(diǎn)擊頁(yè)面上方的 保存 和 一鍵發(fā)布 按鈕，填寫(xiě)任意提交描述，將任務(wù)發(fā)布至實(shí)時(shí)運(yùn)維列表。

成功發(fā)布后可點(diǎn)擊 運(yùn)維 按鈕前往任務(wù)運(yùn)維頁(yè)面。

3.2.4 啟動(dòng)任務(wù)

在運(yùn)維頁(yè)面找到對(duì)應(yīng)任務(wù)后點(diǎn)擊 啟動(dòng) 按鈕啟動(dòng)任務(wù)。

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

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

點(diǎn)擊實(shí)時(shí)開(kāi)發(fā)頁(yè)面左側(cè)目錄第四項(xiàng)流表，打開(kāi)流表管理頁(yè)面。點(diǎn)擊頁(yè)面右上角創(chuàng)建表按鈕，開(kāi)始創(chuàng)建流表。

表名：自行命名

topic：填寫(xiě)上一步 CDC 任務(wù)的目標(biāo) Topic 名稱(chēng)

序列化方式：canal-json

填寫(xiě)完以上信息后可開(kāi)始進(jìn)行字段自動(dòng)解析。

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

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

字段信息確認(rèn)無(wú)誤后即可保存流表。

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

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

3.4.1 創(chuàng)建 SQL 任務(wù)

引擎請(qǐng)選擇 FLINK-1.14。其他類(lèi)似創(chuàng)建 CDC 任務(wù)步驟。

3.4.2 編輯 SQL 代碼

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

在代碼中引用流表時(shí)，直接使用 [庫(kù)].[表] 二元組的寫(xiě)法即可使用對(duì)應(yīng)的流表。

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

Kafka消費(fèi)者組id配置，需要更改配置中的流表名稱(chēng)為自己的流表名稱(chēng)

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

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

具體要修改的內(nèi)容請(qǐng)見(jiàn)代碼中的標(biāo)注。

-- 設(shè)置Kafka消費(fèi)者組id，需要更改自己創(chuàng)建的流表名稱(chēng), 配置方式：'{流表名稱(chēng)}.connections.group.id' = '{groupid-name}'set '這里填流表名稱(chēng).connections.group.id' = 'example001'; -- 設(shè)置讀取消息隊(duì)列的位置, 配置方式：'{流表名稱(chēng)}.scan.startup.mode' = '{earliest-offset / latest-offset}'set '這里填流表名稱(chēng).scan.startup.mode' = 'earliest-offset'; -- 設(shè)置流表主鍵，配置方式：'{流表名稱(chēng)}.primary.keys' = '{primary key name}'set '這里填流表名稱(chēng).primary.keys' = 'id';--設(shè)置源端表表名CREATE view v1 ASSELECT     SUBSTRING(visitTime, 0, 10) AS `date`,     equipment from 這里填流表庫(kù)名稱(chēng).這里填流表名稱(chēng);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ù)源標(biāo)識(shí).MySQL數(shù)據(jù)庫(kù)名稱(chēng).DAU_FINALSELECT * FROM v4;

3.4.3 發(fā)布 SQL 任務(wù)并啟動(dòng)任務(wù)

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

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

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

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

（1）折線(xiàn)圖：

（2）報(bào)表