摘要：本文整理自阿里云資深技術(shù)專家，阿里云Hologres負(fù)責(zé)人姜偉華，在FFA實時湖倉專場的分享。點擊查看>>本篇內(nèi)容主要分為四個部分：

一、實時數(shù)倉分層的技術(shù)需求

二、阿里云一站式實時數(shù)倉Hologres介紹

三、Flink x Hologres：天作之合

四、基于Flink Catalog的Streaming Warehouse實踐

點擊查看視頻回放

一、實時數(shù)倉分層的技術(shù)需求

首先，我們講一講數(shù)倉的分層技術(shù)以及分層技術(shù)的現(xiàn)狀。

1、實時數(shù)倉分層技術(shù)現(xiàn)狀

大數(shù)據(jù)現(xiàn)在越來越講究實時化，在各種場景下都需要實時，例如春晚實時直播大屏，雙11 GMV實時大屏、實時個性化推薦等場景，都對數(shù)據(jù)的實時性有著非常高的要求。為了滿足業(yè)務(wù)的實時性需求，大數(shù)據(jù)技術(shù)也開始逐步發(fā)展出實時數(shù)倉。

但如何構(gòu)建實時數(shù)倉呢？

相比離線數(shù)倉，實時數(shù)倉沒有明確的方法論體系。因此在實踐中，有各種各樣的方法，但沒有一個方法是萬能。最近行業(yè)內(nèi)提出了Streaming Warehouse的概念。Streaming Warehouse的本質(zhì)是分層之間能夠做到實時數(shù)據(jù)的流動，從而解決實時數(shù)倉分層的問題。

下面，我們先來了解下實時數(shù)倉的主流分層方案。

2、實時數(shù)倉主流分層方案

實時數(shù)倉的主流分層方案主要有4個。

方案1：流式ETL

ETL（Extract- Transform-Load）是比較傳統(tǒng)的數(shù)據(jù)倉庫建設(shè)方法，而流式ETL就是指：實時數(shù)據(jù)經(jīng)過Flink實時ETL處理之后，將結(jié)果寫入到KV引擎中，供應(yīng)用查詢。而為了解決中間層不方便排查的問題，也需要將中間層數(shù)據(jù)同步到實時數(shù)倉中供分析之用。最常見的做法就是數(shù)據(jù)通過Flink清洗后，寫到Kafka形成ODS層。再從Kafka消費，經(jīng)過加工形成DWD層。然后Flink加工成DWS層，最后通過加工形成ADS層的數(shù)據(jù)寫到KV引擎并對接上層應(yīng)用。因為直接使用Kafka數(shù)據(jù)進(jìn)行分析和探查很麻煩，所以也會同步一份Kafka數(shù)據(jù)到實時數(shù)倉，通過實時數(shù)倉進(jìn)行分析和探查。

這個方案的優(yōu)勢是層次明確，分工明確。但劣勢是需要有大量的同步任務(wù)、數(shù)據(jù)資源消耗很大、數(shù)據(jù)有很多冗余、處理鏈路較復(fù)雜需要很多的組件。除此之外，這個方案構(gòu)建的實時數(shù)倉分層，尤其是Kafka分層，復(fù)用性非常差，也沒辦法響應(yīng)schema的動態(tài)變化。

方案2：流式ELT

而流式ELT則是將計算后置，直接將明細(xì)數(shù)據(jù)寫進(jìn)實時數(shù)倉（EL），不需要嚴(yán)格的數(shù)倉分層，整個架構(gòu)只需要一層，上層應(yīng)用查詢的時候進(jìn)行數(shù)據(jù)的變換（T）或者分層。常見的做法就是把數(shù)據(jù)加工清洗后，寫到實時數(shù)倉里，形成DWD層，所有的查詢都基于DWD層的明細(xì)數(shù)據(jù)進(jìn)行。

這個方案的好處在于，沒有ETL，只有一層；數(shù)據(jù)修訂很方便。但它的弊端有兩個方面：

• 在查詢性能方面，由于是明細(xì)數(shù)據(jù)查詢，所以在某些場景下不能滿足QPS或延遲的要求。
• 因為沒有嚴(yán)格的數(shù)倉分層，所以數(shù)據(jù)復(fù)用很困難，很難兼顧各方面的訴求。

方案3：定時調(diào)度

既然實時流式無法完成數(shù)據(jù)的實時數(shù)倉分層，我們可以將數(shù)據(jù)實時寫入實時數(shù)倉的DWD層。DWS層、ADS層用離線的高頻調(diào)度方法，實現(xiàn)分鐘級的調(diào)度，從而借用離線數(shù)倉，進(jìn)行分層構(gòu)造。這個也就是業(yè)界常用的方案3。

這個方案的好處在于可以復(fù)用很多離線經(jīng)驗，方案成本低且成熟。但方案也存在如下缺點：

• 延遲大：每一層的延遲都跟調(diào)度相關(guān)，隨著層次越多，調(diào)度延遲越大，實時數(shù)倉也變成了準(zhǔn)實時數(shù)倉。
• 不能完全復(fù)用離線方案：離線調(diào)度一般是小時級或天級，我們可以使用全量計算。但在分鐘級調(diào)度時，必須做增量計算，否則無法及時調(diào)度。

方案4：實時物化視圖

第4種方案就是通過實時數(shù)倉的物化視圖能力實現(xiàn)數(shù)倉分層。常見的做法就是Flink實時加工后，將數(shù)據(jù)寫到實時數(shù)倉形成DWD層，DWS層或ADS層的構(gòu)造依賴于實時數(shù)倉的實時物化視圖能力。

現(xiàn)在主流實時數(shù)倉都開始提供物化視圖的能力，但本質(zhì)上都是提供了一些簡單的聚合類物化視圖。如果物化視圖的需求比較簡單，可以利用實時數(shù)倉里的實時物化視圖能力，將DWS層到ADS層的構(gòu)建自動化，從而讓物化視圖的查詢保證較高的QPS。但這個方案最大的缺點在于，現(xiàn)在的實時物化視圖技術(shù)都還不成熟，能力有限，支持的場景也比較有限。

二、阿里云一站式實時數(shù)倉Hologres介紹

接下來，先介紹一下阿里云一站式實時數(shù)倉Hologres產(chǎn)品。Hologres是阿里云自研的一站式實時數(shù)倉，它同時包含三種能力：

• OLAP能力：同傳統(tǒng)的實時數(shù)倉一樣，可以支持?jǐn)?shù)據(jù)的實時寫入、以及復(fù)雜OLAP實時多維分析快速響應(yīng)，滿足業(yè)務(wù)的極致數(shù)據(jù)探索能力。
• 在線服務(wù)Serving(KV)：可以支持KV查詢場景，提供非常高的QPS和毫秒級的低延遲。
• 湖倉一體：能夠直接查詢數(shù)據(jù)湖的數(shù)據(jù)，以及能夠加速阿里云離線數(shù)倉MaxCompute，助力業(yè)務(wù)更低成本實現(xiàn)湖倉一體。

下面為具體介紹：

• 首先，大家可以把Hologres當(dāng)做一個常見的實時數(shù)倉。它的特點在于寫入側(cè)支持百萬RPS的實時寫入，寫入即可查，沒有延遲。同時也支持高性能的實時整行更新和局部更新。其中，整行更新是把整行替換掉，局部更新可以更新一行中的局部字段，二者都是實時更新。
• 在查詢側(cè)，一方面支持復(fù)雜的OLAP多維分析，可以非常好的支持實時大屏、實時報表等場景。近期Hologres拿到了TPC-H 30TB的性能世界第一的TPC官方認(rèn)證成績，見>>阿里云 ODPS-Hologres刷新世界紀(jì)錄，領(lǐng)先第二名23%。其次，Hologres也支持在線服務(wù)查詢，不僅支持百萬QPS KV點查，而且也支持阿里云達(dá)摩院的Proxima向量檢索引擎，可以支持非常高效的向量檢索能力。同時這些能力在Hologres中是全用SQL表達(dá)，對用戶使用非常友好。此外，為了兼顧數(shù)據(jù)服務(wù)和實時數(shù)倉的需求，Hologres在行存、列存的數(shù)據(jù)格式基礎(chǔ)上，也支持行列共存，即行列共存的表即一份行存，又有一份列存，并且系統(tǒng)保證這兩份數(shù)據(jù)是強(qiáng)一致的，對于OLAP分析，優(yōu)化器會自動選擇列存，對于線上服務(wù)，會自動選擇行存，通過行列共存可以非常友好的實現(xiàn)一份數(shù)據(jù)支撐多個應(yīng)用場景。
• 因為Hologres同時支持OLAP分析和線上服務(wù)，其中線上服務(wù)要求非常高的穩(wěn)定性和SLA。為了保證OLAP分析和線上服務(wù)時不會發(fā)生沖突，我們支持了讀寫分離，從而實現(xiàn)OLAP與數(shù)據(jù)服務(wù)的強(qiáng)隔離。
• 最后，在湖倉數(shù)據(jù)交互式分析方面，Hologres對阿里云MaxCompute離線數(shù)倉里的數(shù)據(jù)，數(shù)據(jù)湖中的數(shù)據(jù)都可以秒級交互式分析，且不需要做任何的數(shù)據(jù)搬遷。
• 除此之外，Hologres的定位是一站式的企業(yè)級實時數(shù)倉，所以除了上述能力，我們還有很多其他能力。包括數(shù)據(jù)的治理、成本治理、數(shù)據(jù)血緣、數(shù)據(jù)脫敏、數(shù)據(jù)加密、IP白名單、數(shù)據(jù)的備份和恢復(fù)等等。

三、Flink x Hologres：天作之合

1、Hologres與Flink深度集成

Flink對于實時數(shù)倉能夠提供非常豐富的數(shù)據(jù)處理、數(shù)據(jù)入湖倉的能力。Hologres與Flink有些非常深度的整合能力，具體包括：

• Hologres可以作為Flink的維表：在實時計算的場景下，Flink對維表的需求很強(qiáng)，Hologres支持百萬級至千萬級RPS的KV點查能力，可以直接當(dāng)做Flink維表使用，且可以做到實時更新，對于像實時特征存儲等維表關(guān)聯(lián)場景就也可以非常高效的支持。
• Hologres可以作為Flink的結(jié)果表：Hologres支持高性能的實時寫入和整行實時更新的能力，可以結(jié)合Flink，輸出需要強(qiáng)大的Update能力，滿足數(shù)倉場景下的實時更新、覆蓋等需求。與此同時，Hologres還有很強(qiáng)的局部更新能力。局部更新能力在很多場景下，可以替代Flink的多流Join，為客戶節(jié)省成本。
• Hologres可以作為Flink的源表：Hologres支持Binlog能力，一張表的任何變化，比如insert、update、delete等等，都會產(chǎn)生Binlog事件。Flink可以訂閱Hologres Binlog，進(jìn)行驅(qū)動計算。由于Flink支持Hologres的整表讀取，二者結(jié)合構(gòu)成了Flink全增量一體化的讀取能力。并且，Hologres也對了接Flink CDC，它可以驅(qū)動Flink CDC的計算。
• 支持Hologres Catalog：通過Hologres Catalog的任何操作，都會直接實時反映到Hologres里，用戶也不需要在Flink建Hologres表，這樣就使得Flink+Hologres就具備了整庫同步、Schema Evolution的能力。

2、基于Flink+Hologres的Streaming Warehouse方案

那Flink和Hologres如何構(gòu)建Streaming Warehouse？

Streaming Warehouse：數(shù)據(jù)能在數(shù)倉之間實時的流動，本質(zhì)上就是解決實時數(shù)倉分層的問題

最開始我們介紹了常見的數(shù)倉分層方案，Flink+Hologres的Streaming Warehouse方案則是可以完全將Flink+Kafka替換。具體做法如下：

1. 將Flink寫到Hologres里，形成ODS層。Flink訂閱ODS層的Hologres Binlog進(jìn)行加工，將Flink從DWD層再次寫入Hologres里。
2. Flink再訂閱DWD層的Hologres Binlog，通過計算形成DWS層，將其再次寫入Hologres里。
3. 最后，由Hologres對外提供應(yīng)用查詢。

該方案相比Kafka有如下優(yōu)點：

• 解決了傳統(tǒng)中間層Kafka數(shù)據(jù)不易查、不易更新、不易修正的問題。Hologres的每一層都可查、可更新、可修正。
• Hologres的每一層都可以單獨對外提供服務(wù)。因為每一層的數(shù)據(jù)都是可查的，所以數(shù)據(jù)的復(fù)用會更好，真正實現(xiàn)數(shù)倉分層復(fù)用的目標(biāo)。
• Hologres支持?jǐn)?shù)據(jù)復(fù)用，模型統(tǒng)一，架構(gòu)簡化。通過Flink+Hologres，就能實現(xiàn)實時數(shù)倉分層，簡化架構(gòu)和降低成本。

3、Flink+Hologres核心能力：Binlog、行列共存、資源隔離

上面講的Flink+Hologres的Streaming Warehouse方案，其強(qiáng)依賴于以下三個Hologres核心能力：

1. Binlog：因為實時數(shù)倉一般沒有Binlog，但Hologres提供了Binlog能力，用來驅(qū)動Flink做實時計算，正因為有了Binlog，Hologres才能作為流式計算的上游。
2. 行列共存。一張表既有行存數(shù)據(jù)，又有列存數(shù)據(jù)。這兩份數(shù)據(jù)是強(qiáng)一致的。行列共存的特性讓中間層的每張表，不但能夠給Flink使用，而且可以給其他應(yīng)用（比如OLAP、或者線上服務(wù)）使用。
3. 資源強(qiáng)隔離。實時數(shù)倉一般是弱隔離或軟隔離，通過資源組、資源隊列的方法實現(xiàn)資源隔離。如果Flink的資源消耗很大，可能影響中間層的點查性能。但在Hologres強(qiáng)隔離的能力下，Flink對Hologres Binlog的數(shù)據(jù)拉取，不會影響線上服務(wù)。

通過Binlog、行列共存、資源強(qiáng)隔離的三個特點，不僅能讓Flink+Hologres形成Streaming Warehouse，并且能夠使中間的每層數(shù)據(jù)復(fù)用，被其他應(yīng)用或線上服務(wù)使用，助力企業(yè)構(gòu)建最簡單最完整的實時數(shù)倉。

4、基于Flink+Hologres的多流合并

接下來，講一講基于Flink+Hologres的多流合并。

因為Hologres有特別強(qiáng)大的局部更新能力，基于此我們可以簡化Flink的多流Join。比如在風(fēng)控場景下，我們需要基于用戶ID構(gòu)建用戶的多側(cè)面畫像，用戶畫像來自很多數(shù)據(jù)源，比如客戶的瀏覽行為、成交行為、履約行為等等。把數(shù)據(jù)源的數(shù)據(jù)按照用戶ID，把每個用戶放到一行里，形成不同的字段，形成用戶的完整畫像。

傳統(tǒng)的方式需要用Flink多流Join實現(xiàn)，Flink把上游的多個數(shù)據(jù)源關(guān)聯(lián)到一起，Join后寫到Kafka里，然后驅(qū)動下游的Flink，加工這行完整的數(shù)據(jù)。這就使得多流Join非常耗資源。

所以在Flink+Hologres的Streaming Warehouse方案中，可以利用Hologres的局部更新能力，把一張表定為定義成Hologres的行存表或行列共存表。此時，整個方案就簡化成上游每個數(shù)據(jù)源，同步數(shù)據(jù)到Hologres表的若干個字段里，若干個任務(wù)同時寫入這張表，然后利用Hologres的局部更新能力，把數(shù)據(jù)匯總在一起。

如果打開這張Hologres表的 Binlog，上游任何數(shù)據(jù)源的變化都會更新這張表，使這張表的Binlog中生成行數(shù)據(jù)的最新狀態(tài)，然后驅(qū)動下游的Flink繼續(xù)計算，從而完美匹配常見的風(fēng)控場景。這種用法下，資源消耗、運(yùn)維都得到了極大的簡化。

四、基于Flink Catalog的Streaming Warehouse實踐

Flink+Hologres的Streaming Warehouse方案已經(jīng)非常成熟，但唯一的缺點在于，用戶需要在兩個系統(tǒng)之間切換，過程比較繁瑣。為了讓用戶操作更簡單，我們基于Flink Catalog提供了更加簡單的使用體驗。

下面我們來看看怎么樣基于Flink Catalog去構(gòu)建基于Flink+Hologres的Streaming Warehouse。我們會發(fā)現(xiàn)，有了Flink Catalog后，整個使用體驗會很簡單，并能充分發(fā)揮Flink和Hologres兩個產(chǎn)品的強(qiáng)大能力。

下圖是一個典型的Flink+Hologres實時ETL鏈路：

1. ODS層、DWD層、ODS層的數(shù)據(jù)都存在Hologres中。
2. 鏈路中所有的數(shù)據(jù)加工都是通過Flink SQL完成。在整個ETL鏈路中，用戶不需要任何Hologres SQL，直接寫Flink SQL即可。
3. Flink用戶可以通過Flink SQL對每層中的Hologres數(shù)據(jù)進(jìn)行數(shù)據(jù)探查（流模式和批模式都可以）。比如：當(dāng)我們發(fā)現(xiàn)DWS層的數(shù)據(jù)結(jié)果出現(xiàn)問題，需要查看哪層的結(jié)果有問題或邏輯有錯誤。此時，我們可以復(fù)用原來的Flink SQL來進(jìn)行探查、定位或者數(shù)據(jù)重新消費。
4. Hologres中的每層數(shù)據(jù)都可以對外提供查詢和服務(wù)（通過Hologres SQL）。

接下來，以某個電商場景為例，演示一下基于Flink Catalog的Streaming Warehouse。如下圖所示，有一個MySQL數(shù)據(jù)庫作為訂單庫，里面有訂單表orders、訂單支付表orders_pay、以及產(chǎn)品品類表product_catalog。

1. 第一步，我們通過Flink的實時數(shù)倉，把數(shù)據(jù)實時同步到Hologres里，形成ODS層。
2. 第二步，加工DWD層。將DWD層的數(shù)據(jù)寫到Hologres里。在這個過程中，我們需要把訂單表和訂單支付表，合并成一張表，實現(xiàn)多路合并。與此同時，我們希望orders表關(guān)聯(lián)商品品類表product_catalog。
3. 第三步，驅(qū)動下游計算，構(gòu)建DWS層。以用戶維度和商店維度，收集統(tǒng)計數(shù)據(jù)。比如用戶每天的訂單金額和商店每天的訂單金額，從而形成一條完整的鏈路。
4. 第四步，將DWS層的表推薦給系統(tǒng)使用。作為用戶和商店的特征，用做推薦用途。
5. 第五步，DWD層的表能夠直接用來做實時統(tǒng)計分析、統(tǒng)計產(chǎn)品、實時大屏、實時報表。

上圖中的綠色鏈路，全部使用Flink SQL完成。橙色鏈路對外提供服務(wù)，由Hologres SQL完成。

接下來，講一講每個步驟是如何運(yùn)行的。

第一步，在Flink實時數(shù)倉，形成ODS層。首先，創(chuàng)建一個Hologres的Catalog。MySQL中存儲訂單、支付以及商品信息3張表，通過Flink Catalog功能，將MySQL整庫的數(shù)據(jù)實時同步至Hologres，形成ODS。相關(guān)代碼如下所示。我們可以看到，MySQL整庫同步到Hologres，通過Flink SQL來表達(dá)是非常簡單的。

-- 創(chuàng)建Hologres Catalog
CREATE CATALOG holo WITH ( ‘type’ = ‘hologres’ … );-- MySQL整庫同步到Hologres
CREATE DATABASE IF NOT EXISTS holo.order_dw 
AS DATABASE mysql.sw INCLUDING all tables;

第二步，DWD實時構(gòu)建。數(shù)據(jù)實時寫入ODS層后，Flink讀取Hologres Binlog，并用多流合并、維表關(guān)聯(lián)將訂單、交易、商品3個表打成一個大寬表，實時寫入至Hologres的訂單匯總表中，形成DWD層。

如下SQL是DWD層表的建表語句。這張目標(biāo)表包含了來自orders、orders_pay、product_catalog的字段，關(guān)聯(lián)了相關(guān)的用戶信息、商戶信息、訂單信息、商品品類信息等等，形成了一張寬表。

CREATE TABLE holo.order_dw.dwd_orders (order_id bigint not null primary key,--字段來自order 表order_user_id bigint,order_shop_id bigint,order_product_id string,order_fee numeric(20,2),order_create_time timestamp_ltz,order_update_time timestamp_ltz,order_state int,--字段來自product_catalog表order_product_catalog_name string,--字段來自orders_pay表pay_id bigint,pay_platfrom int,pay_create_time timestamp_ltz
) ;

下面的SQL是真正的計算邏輯，這里包含兩個INSERT語句：

• 第一個INSERT語句是從orders表實時打?qū)捄髮懭?。這里用到了Hologres的維表關(guān)聯(lián)能力。實時打?qū)捄?#xff0c;寫入目標(biāo)表的部分字段。
• 第二個INSERT語句是從orders_pay表實時同步到同一張目標(biāo)表，更新另外一些字段。

這兩個INSERT語句最大的關(guān)聯(lián)在于，它們寫的是同一張表，會自動利用目標(biāo)表的主鍵ID進(jìn)行關(guān)聯(lián)。每個INSERT都是做了目標(biāo)表的局部更新，兩者的合力結(jié)果是實時更新的目標(biāo)寬表。

BEGIN STATEMENT SET;
-- 從orders表實時打?qū)捄髮懭?INSERT INTO holo.order_dw.dwd_orders (
order_id,
order_user_id,
order_shop_id,
order_product_id,
order_fee,
order_create_time,
order_update_time,
order_state,
order_product_catalog_name
)
SELECT 
o.*,
dim.catalog_name
FROM
holo.order_dw.orders o 
LEFT JOIN holo.order_dw.product_catalog 
FOR SYSTEM_TIME AS OF proctime () AS dim
ON
o.product_id = dim.product_id;-- 從order_pays表實時寫入
INSERT INTO holo.order_dw.dwd_orders (
pay_id,
order_id,
pay_platform,
pay_create_time
)
SELECT *
FROM
holo.order_dw.orders_pay;
END;

第三步，DWS層的實時聚合。在DWD的基礎(chǔ)上，通過Flink讀取Hologres DWD的Binlog數(shù)據(jù)，進(jìn)行實時指標(biāo)聚合計算，比如按照用戶維度聚合，按照商戶維度聚合等，然后實時寫入Hologres，形成DWS層。

• 先是創(chuàng)建對應(yīng)的聚合指標(biāo)表，DDL語句如下

-- 用戶維度聚合指標(biāo)表
CREATE TABLE holo.order_dw.dws_users (user_id bigint not null,ds string not null,-- 當(dāng)日完成支付總金額     payed_buy_fee_sum numeric(20,2) not null,      primary key(user_id,ds) NOT ENFORCED
);-- 商戶維度聚合指標(biāo)表
CREATE TABLE holo.order_dw.dws_shops (shop_id bigint not null,ds string not null,-- 當(dāng)日完成支付總金額payed_buy_fee_sum numeric(20,2) not null,     primary key(shop_id,ds) NOT ENFORCED
);

• 然后將數(shù)據(jù)寫入Hologres中，經(jīng)過簡單的三步后，Flink SQL構(gòu)建了完整的Streaming Warehouse分層體系。

--數(shù)據(jù)寫入Hologres BEGIN STATEMENT SET;
INSERT INTO holo.order_dw.dws_users
SELECT order_user_id,DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,SUM (order_fee)
FROM holo.order_dw.dwd_orders c
WHERE pay_id IS NOT NULL 
AND order_fee IS NOT NULL
GROUP BY order_user_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');INSERT INTO holo.order_dw.dws_shops
SELECT order_shop_id,DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,SUM (order_fee)
FROM holo.order_dw.dwd_orders c
WHERE pay_id IS NOT NULL 
AND order_fee IS NOT NULL
GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
END;

第四步，構(gòu)建應(yīng)用，基于DWS層，對外提供服務(wù)。

數(shù)據(jù)的分層和加工完成后，業(yè)務(wù)就可以通過Hologres查詢數(shù)據(jù)并應(yīng)用。在這個例子里，推薦系統(tǒng)要求非常高的點查性能，所以要求百萬級的QPS檢查能力。Hologres的行存表或者行列共存表完全可以滿足。

這個方案和傳統(tǒng)的實時數(shù)倉最大的差別是：傳統(tǒng)的實時數(shù)倉只有最后一層的數(shù)據(jù)，可對外提供服務(wù)。而在Hologres里，DWD等中間層數(shù)據(jù)也可以對外提供服務(wù)，進(jìn)行實時報表統(tǒng)計。用戶可以在中間層進(jìn)行查詢操作，對接各種實時應(yīng)用、實時大屏。比如

• 直接查DWD層的數(shù)據(jù)，典型的如根據(jù)用戶ID返回推薦商品（KV場景）

--場景4: 根據(jù)用戶特征推薦商品
SELECT * 
FROM dws_users 
WHERE 
user_id = ? 
AND ds = '2022-11-09’;--場景4: 根據(jù)店鋪特征推薦商品SELECT * 
FROM dws_shops 
WHERE 
shop_id = ? 
AND ds = '2022-11-09’;

• 實時報表查看訂單量和退單量（OLAP）。

--場景6:基于寬表數(shù)據(jù)展示實時報表
-- 最近30天,每個品類的訂單總量和退單總量
SELECT
TO_CHAR(order_create_time, 'YYYYMMDD'),
order_product_catalog_name,
COUNT(*),
COUNT(CASE WHEN refund_id IS NOT NULL THEN 1 ELSE null END)
FROM
dwd_orders
WHERE
order_create_time > now() - '30 day' :: inteval
GROUP BY
1, 2
ORDER BY
1, 2;

第五步，問題排查：Flink數(shù)據(jù)探查。如果某個業(yè)務(wù)指標(biāo)出現(xiàn)異常，Flink可以直接探查每層表的數(shù)據(jù)來快速定位。比如用Flink探查Hologres DWD層的orders表。Hologres支持Flink的流模式和批模式對數(shù)據(jù)的探查。

由于流模式是Flink的默認(rèn)模式，因此我們不需要設(shè)置執(zhí)行模式。它可以直接記錄數(shù)據(jù)變化，從而非常方便的查看數(shù)據(jù)異常。流模式可以探查獲取一段時間范圍內(nèi)的數(shù)據(jù)及其變化情況。

-- 流模式探查 
SELECT * 
FROM holo.order_dw.dwd_orders 
/*+ OPTIONS('cdcMode'='false', 'startTime'='2022-11-09 00:00:00') */ c 
WHERE user_id = 0;

與此同時，批模式探查是獲取當(dāng)前時刻的最新數(shù)據(jù)。Hologres也支持Flink批模式的數(shù)據(jù)探查。批模式和流模式的區(qū)別在于，流模式關(guān)注的是變化，批模式關(guān)注的是表中的最新狀態(tài)。

-- 批模式探查 
set 'execution.runtime-mode'='batch’;SELECT * 
FROM holo.order_dw.dwd_orders 
WHERE user_id = 0AND order_create_time>'2022-11-09 00:00:00';

五、總結(jié)

Hologres跟Flink深度集成。實現(xiàn)完整的Streaming Warehouse方案，該方案有如下明顯優(yōu)勢：

1. 一站式：全鏈路都可以用SQL表示，并且只需要用到Flink和Hologres兩個組件，操作非常方便。實時ETL鏈路、數(shù)據(jù)分層完全可以用Flink SQL實現(xiàn)，Hologres提供對外提供在線服務(wù)和OLAP查詢，每層數(shù)據(jù)可復(fù)用、可查，方便構(gòu)建實時數(shù)倉的數(shù)據(jù)分層和復(fù)用體系。
2. 高性能：這種方案可以使得使得Hologres發(fā)揮極致的實時寫入、實時更新能力和多維OLAP、高并發(fā)點查能力，Flink發(fā)揮實時加工能力。
3. 企業(yè)級：自帶多種企業(yè)級能力，不僅運(yùn)維更簡單，可觀測性更好，安全能力更強(qiáng)，也提供多種高可用能力，從而企業(yè)更加方便的構(gòu)建企業(yè)級的Streaming Warehouse。