作者｜徐榜江 余文兵 趙紅梅

隨著大數(shù)據(jù)的迅猛發(fā)展，企業(yè)越來(lái)越重視數(shù)據(jù)的價(jià)值，這就意味著需要數(shù)據(jù)盡快到達(dá)企業(yè)分析決策人員，以最大化發(fā)揮數(shù)據(jù)價(jià)值。企業(yè)最常見(jiàn)的做法就是通過(guò)構(gòu)建實(shí)時(shí)數(shù)倉(cāng)來(lái)滿足對(duì)數(shù)據(jù)的快速探索。在業(yè)務(wù)建設(shè)過(guò)程中，實(shí)時(shí)數(shù)倉(cāng)需要支持?jǐn)?shù)據(jù)實(shí)時(shí)寫(xiě)入與更新、業(yè)務(wù)敏捷快速響應(yīng)、數(shù)據(jù)自助分析、運(yùn)維操作便捷、云原生彈性擴(kuò)縮容等一系列需求，而這就依賴一個(gè)強(qiáng)大的實(shí)時(shí)數(shù)倉(cāng)解決方案。阿里云實(shí)時(shí)計(jì)算 Flink 版（以下簡(jiǎn)稱“阿里云 Flink”）提供全增量一體化數(shù)據(jù)同步技術(shù)、強(qiáng)大的流式 ETL 等能力，支持海量數(shù)據(jù)實(shí)時(shí)入倉(cāng)入湖。阿里云 Hologres 作為新一代實(shí)時(shí)數(shù)倉(cāng)引擎能同時(shí)解決 OLAP 多維分析、在線服務(wù)、離線數(shù)據(jù)加速等多個(gè)業(yè)務(wù)查詢場(chǎng)景，通過(guò)阿里云 Flink 與 Hologres 的強(qiáng)強(qiáng)結(jié)合，實(shí)現(xiàn)全鏈路的數(shù)據(jù)探索實(shí)時(shí)化、數(shù)據(jù)分析敏捷化，快速助力業(yè)務(wù)構(gòu)建企業(yè)級(jí)一站式實(shí)時(shí)數(shù)倉(cāng)，實(shí)現(xiàn)更具時(shí)效更智能的業(yè)務(wù)決策。

在本文中，我們將會(huì)介紹阿里云 Flink、阿里云 Hologres 在構(gòu)建實(shí)時(shí)數(shù)倉(cāng)上所具備的核心能力以及二者結(jié)合的最佳解決方案，用戶通過(guò)阿里云 Flink+Hologres 實(shí)時(shí)數(shù)倉(cāng)解決方案，可以顯著降低數(shù)倉(cāng)建設(shè)門(mén)檻，讓數(shù)據(jù)發(fā)揮更大的價(jià)值，助力各行各業(yè)實(shí)現(xiàn)數(shù)字化升級(jí)。

Flink CDC 核心能力

Apache Flink 是開(kāi)源的大數(shù)據(jù)流式計(jì)算引擎，支持處理數(shù)據(jù)庫(kù)、Binlog、在線日志等多種實(shí)時(shí)數(shù)據(jù)，提供端到端亞秒級(jí)實(shí)時(shí)數(shù)據(jù)分析能力，并通過(guò)標(biāo)準(zhǔn) SQL 降低實(shí)時(shí)業(yè)務(wù)開(kāi)發(fā)門(mén)檻。伴隨著實(shí)時(shí)化浪潮的發(fā)展和深化，F(xiàn)link 已逐步演進(jìn)為流處理的領(lǐng)軍角色和事實(shí)標(biāo)準(zhǔn)，并蟬聯(lián) Apache 社區(qū)最活躍項(xiàng)目。

Flink CDC 是阿里云計(jì)算平臺(tái)事業(yè)部 2020 年 7 月開(kāi)源的一款數(shù)據(jù)集成框架，與 Flink 生態(tài)深度融合，具有全增量一體化、無(wú)鎖讀取、并發(fā)讀取、分布式架構(gòu)等技術(shù)優(yōu)勢(shì)，既可以替代傳統(tǒng)的 DataX 和 Canal 工具做數(shù)據(jù)同步，也支持?jǐn)?shù)據(jù)庫(kù)數(shù)據(jù)實(shí)時(shí)入湖入倉(cāng)，同時(shí)還具備強(qiáng)大的數(shù)據(jù)加工能力。

在構(gòu)建實(shí)時(shí)數(shù)倉(cāng)的過(guò)程中，數(shù)據(jù)采集是必需的組件。在傳統(tǒng)的 ETL 架構(gòu)里，采集層國(guó)外用戶通常選擇 Debezium，國(guó)內(nèi)用戶則習(xí)慣用 DataX 和 Canal，采集工具負(fù)責(zé)采集數(shù)據(jù)庫(kù)的全量數(shù)據(jù)和增量數(shù)據(jù)。采集到的數(shù)據(jù)會(huì)輸出到消息中間件如 Kafka，然后通過(guò) Flink 計(jì)算引擎實(shí)時(shí)消費(fèi)消息中間件數(shù)據(jù)做計(jì)算層的數(shù)據(jù)清洗和數(shù)據(jù)加工，加工完成后再寫(xiě)入目的端（裝載層），通常是各種數(shù)據(jù)庫(kù)、數(shù)據(jù)湖和數(shù)據(jù)倉(cāng)庫(kù)。在傳統(tǒng) ETL 鏈路中，數(shù)據(jù)采集工具與消息隊(duì)列是比較重的組件，可能維護(hù)在不同的團(tuán)隊(duì)，在上游的數(shù)據(jù)源有業(yè)務(wù)變更或者這些組件需要升級(jí)維護(hù)時(shí)，整個(gè)鏈路的維護(hù)成本會(huì)非常大。

通過(guò)使用 Flink CDC 去替換上圖中的數(shù)據(jù)采集組件與消息隊(duì)列，將采集層（Extraction）和計(jì)算層（Transformation）合并，簡(jiǎn)化了整個(gè) ETL 分析鏈路，用戶可以使用更少的組件完成數(shù)據(jù)鏈路的搭建，整體架構(gòu)帶來(lái)更低的運(yùn)維開(kāi)銷和更少的硬件成本、更好的數(shù)據(jù)鏈路穩(wěn)定性、以及降低端到端的數(shù)據(jù)延遲。除了穩(wěn)定性的提升，F(xiàn)link CDC 的另一個(gè)優(yōu)勢(shì)就是用戶只需要寫(xiě) SQL 腳本就能完成 CDC 數(shù)據(jù)的清洗，加工和同步，極大地降低了用戶使用門(mén)檻。

除全增量一體化同步能力外，阿里云 Flink CDC 還提供了表結(jié)構(gòu)變更自動(dòng)同步、整庫(kù)同步、分庫(kù)分表合并同步等諸多企業(yè)級(jí)特性，方便用戶快速打通數(shù)據(jù)孤島，實(shí)現(xiàn)業(yè)務(wù)價(jià)值。

1.1 全增量一體化同步

Flink CDC 通過(guò)增量快照讀取算法在開(kāi)源數(shù)據(jù)集成領(lǐng)域率先支持了無(wú)鎖讀取、并行讀取、斷點(diǎn)續(xù)傳、不丟不重四個(gè)重要特性。其中無(wú)鎖讀取徹底解決了數(shù)據(jù)同步對(duì)上游業(yè)務(wù)數(shù)據(jù)庫(kù)的死鎖風(fēng)險(xiǎn)，并行讀取很好地滿足了海量數(shù)據(jù)同步的需求，斷點(diǎn)續(xù)傳和不丟不重特性則是提升了同步鏈路的穩(wěn)定性和可靠性。

增量快照讀取算法的核心思路就是在全量讀取階段把表分成一個(gè)個(gè) chunk 進(jìn)行并發(fā)讀取，在進(jìn)入增量階段后只需要一個(gè) task 進(jìn)行單并發(fā)讀取 Binlog 日志，在全量和增量自動(dòng)切換時(shí)，通過(guò)無(wú)鎖算法保障一致性。這種設(shè)計(jì)在提高讀取效率的同時(shí)，進(jìn)一步節(jié)約了資源，實(shí)現(xiàn)了全增量一體化的數(shù)據(jù)同步。配合阿里云實(shí)時(shí)計(jì)算產(chǎn)品提供的資源自動(dòng)調(diào)優(yōu)特性，F(xiàn)link CDC 作業(yè)的資源可以做到自動(dòng)擴(kuò)縮容，無(wú)需手動(dòng)介入。

1.2 表結(jié)構(gòu)變更自動(dòng)同步

隨著業(yè)務(wù)的迭代和發(fā)展，數(shù)據(jù)源的表結(jié)構(gòu)變更是經(jīng)常會(huì)發(fā)生的操作。用戶需要及時(shí)地去修改數(shù)據(jù)同步作業(yè)以適配最新的表結(jié)構(gòu)，這一方面帶來(lái)了較大的運(yùn)維成本，也影響了同步管道的穩(wěn)定性和數(shù)據(jù)的時(shí)效性。阿里云 Flink 支持通過(guò) Catalog 來(lái)實(shí)現(xiàn)元數(shù)據(jù)的自動(dòng)發(fā)現(xiàn)和管理，配合 CTAS （Create Table AS）語(yǔ)法，用戶可以通過(guò)一行 SQL 實(shí)現(xiàn)數(shù)據(jù)的同步和表結(jié)構(gòu)變更自動(dòng)同步。

Flink SQL> USE CATALOG holo;

Flink SQL> CREATE TABLE user AS TABLE mysql.`order_db`.`user`;

CTAS 語(yǔ)句會(huì)解析成一個(gè) Flink 作業(yè)執(zhí)行，這個(gè) Flink 作業(yè)源頭支持讀取數(shù)據(jù)變更和表結(jié)構(gòu)變更并同步到下游，數(shù)據(jù)和表結(jié)構(gòu)變更都可以保證順序，上述 CTAS 語(yǔ)句運(yùn)行時(shí)結(jié)構(gòu)變更同步的效果如下圖所示。

示例如果在上游 MySQL 的 user 表中新增一列 age，并插入一條 id 為 27，年齡為 30 的記錄。

MySQL> ALTER TABLE `user` ADD COLUMN `age` INT;

MySQL> INSERT INTO `user` (id, name, age) VALUES (27, 'Tony', 30);

user 表上的數(shù)據(jù)和結(jié)構(gòu)變更都能實(shí)時(shí)地自動(dòng)同步到下游 Hologres 的 user 表中，id 為 12,16 和 19 的歷史數(shù)據(jù)，新增的列會(huì)自動(dòng)補(bǔ) NULL 值。

1.3 整庫(kù)同步

在實(shí)時(shí)數(shù)倉(cāng)構(gòu)建中，用戶經(jīng)常需要將整個(gè)數(shù)據(jù)庫(kù)同步到數(shù)倉(cāng)中做進(jìn)一步的分析，一張表一個(gè)同步作業(yè)的方式不但浪費(fèi)資源，也會(huì)給上游數(shù)據(jù)庫(kù)產(chǎn)生較大的壓力。針對(duì)這類用戶痛點(diǎn)，阿里云 Flink CDC 提供了整庫(kù)同步特性。整庫(kù)同步功能通過(guò) CDAS (Create Database AS) 語(yǔ)法配合 Catalog 實(shí)現(xiàn)。

Flink SQL> USE CATALOG holo;

Flink SQL> CREATE DATABASE holo_order AS DATABASE

mysql.`order_db` INCLUDING ALL TABLES;

例如 MySQL Catalog 和 Hologres Catalog 配合 CDAS 語(yǔ)法，可以完成 MySQL 到 Hologres 的全增量數(shù)據(jù)同步。CDAS 語(yǔ)句會(huì)解析成一個(gè) Flink 作業(yè)執(zhí)行，這個(gè) Flink 作業(yè)自動(dòng)解析源表的表結(jié)構(gòu)及相應(yīng)的參數(shù)，并將指定的一個(gè)或多個(gè)數(shù)據(jù)庫(kù)同步到下游 Hologres 數(shù)倉(cāng)中，整個(gè)過(guò)程用戶無(wú)需手寫(xiě) DDL 語(yǔ)句，無(wú)需用戶在 Hologres 提前創(chuàng)建表，就能快速實(shí)現(xiàn)數(shù)據(jù)的整庫(kù)同步。

CDAS 作業(yè)默認(rèn)提供表結(jié)構(gòu)變更同步能力，所有表的結(jié)構(gòu)變更都會(huì)按照發(fā)生順序同步至下游 Hologres 實(shí)時(shí)數(shù)倉(cāng)，CDAS 語(yǔ)法也支持過(guò)濾不需要同步的表。

1.4 分庫(kù)分表合并同步

分庫(kù)分表是高并發(fā)業(yè)務(wù)系統(tǒng)采用的經(jīng)典數(shù)據(jù)庫(kù)設(shè)計(jì)，通常我們需要將分庫(kù)分表的業(yè)務(wù)數(shù)據(jù)匯聚到一張數(shù)倉(cāng)中的大表，方便后續(xù)的數(shù)據(jù)分析，即分庫(kù)分表合并同步的場(chǎng)景。針對(duì)這種場(chǎng)景，阿里云 Flink CDC 提供了分庫(kù)分表合并同步特性，通過(guò)在 CTAS 語(yǔ)法支持源庫(kù)和源表的正則表達(dá)式，源數(shù)據(jù)庫(kù)的分表可以高效地合并同步到下游 Hologres 數(shù)倉(cāng)中。

Flink SQL> USE CATALOG holo;

Flink SQL> CREATE TABLE order AS TABLE mysql.`order_db.*`.`order_.*`;

述 CTAS 語(yǔ)句中的源庫(kù)名 order_db.* 是個(gè)正則表達(dá)式，可以匹配當(dāng)前 MySQL 實(shí)例下的 order_db01，order_db02 和 order_db03 三個(gè)庫(kù)，源表名 order* 也是個(gè)正則表達(dá)式，可以匹配三個(gè)庫(kù)下所有以 order打頭的表。

針對(duì)分庫(kù)分表同步場(chǎng)景，用戶只需要提供分庫(kù)分表的正則表達(dá)式就可以將這多張分庫(kù)分表合并同步到下游 Hologres 數(shù)倉(cāng)的 ordder 表中。與其他 CDAS 語(yǔ)句一樣，分庫(kù)分表同步場(chǎng)景默認(rèn)提供表結(jié)構(gòu)變更自動(dòng)同步特性，下游 Hologres 表的 schema 為所有分表合并后的最寬 schema。分庫(kù)分表同步時(shí)每行記錄所屬的庫(kù)名和表名會(huì)作為額外的兩個(gè)字段自動(dòng)寫(xiě)入到 user 表中，庫(kù)名(上圖中 db 列)、表名(上圖中 tbl 列)和原主鍵(上圖中 id 列) 會(huì)一起作為下游 Hologres user 表的聯(lián)合主鍵，保證 Hologres user 表上主鍵的唯一性。

Hologres 核心能力

阿里云 Hologres 是自研的一站式實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)引擎，支持海量數(shù)據(jù)實(shí)時(shí)寫(xiě)入、實(shí)時(shí)更新、實(shí)時(shí)分析，支持標(biāo)準(zhǔn) SQL（兼容 PostgreSQL 協(xié)議），提供 PB 級(jí)數(shù)據(jù)多維分析（OLAP）與即席分析以及高并發(fā)低延遲的在線數(shù)據(jù)服務(wù)（Serving），與阿里云 Flink、MaxCompute、DataWorks 等深度融合，為企業(yè)提供離在線一體化全棧數(shù)倉(cāng)解決方案。

2.1 高性能實(shí)時(shí)寫(xiě)入與更新

數(shù)據(jù)寫(xiě)入的時(shí)效性是實(shí)時(shí)數(shù)倉(cāng)的重要能力之一。對(duì)于 BI 類等延遲不敏感的業(yè)務(wù)查詢，如果寫(xiě)入時(shí)延幾秒甚至幾分鐘可能是可以接受的。而對(duì)于很多生產(chǎn)系統(tǒng)，如實(shí)時(shí)風(fēng)控、實(shí)時(shí)大屏等場(chǎng)景，要求數(shù)據(jù)寫(xiě)入即可見(jiàn)。如果寫(xiě)入出現(xiàn)延遲，就會(huì)查詢不到最新的數(shù)據(jù)，嚴(yán)重影響線上業(yè)務(wù)決策。在實(shí)時(shí)數(shù)倉(cāng)整個(gè)數(shù)據(jù)處理鏈路中，Hologres 作為一站式實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)引擎，提供海量數(shù)據(jù)高性能的實(shí)時(shí)寫(xiě)入，數(shù)據(jù)寫(xiě)入即可查詢，無(wú)延遲。

同時(shí)在數(shù)倉(cāng)場(chǎng)景上，數(shù)據(jù)來(lái)源復(fù)雜，會(huì)涉及到非常多的數(shù)據(jù)更新、修正的場(chǎng)景，Hologres 可以通過(guò)主鍵（Primary Key, PK）提供高性能的 Upsert 能力，整個(gè)寫(xiě)入和更新過(guò)程確保 Exactly Once，滿足對(duì)對(duì)數(shù)據(jù)的合并、更新等需求。

下圖為 Hologres 128C 實(shí)例下，10 個(gè)并發(fā)實(shí)時(shí)寫(xiě)入 20 列的列存表的測(cè)試結(jié)果。其中豎軸表示每秒寫(xiě)入記錄數(shù)，橫軸為 4 個(gè)寫(xiě)入場(chǎng)景：

●Append Only：寫(xiě)入表無(wú)主鍵，寫(xiě)入能力 230 萬(wàn)+的 RPS。

●INSERT：寫(xiě)入表有主鍵，如果主鍵沖突就丟棄新行，寫(xiě)入能力 200 萬(wàn) RPS。

●UPDATE-1：寫(xiě)入表有主鍵，表中原始數(shù)據(jù)量為 2 億，按照主鍵 Upsert，寫(xiě)入能力 80 萬(wàn)的 RPS。

●UPDATE-2：寫(xiě)入表有主鍵，表中數(shù)據(jù)量為 20 億，按照主鍵做 Upsert，寫(xiě)入能力 70 萬(wàn)的 RPS。

2.2 實(shí)時(shí) OLAP 分析

Hologres 采用可擴(kuò)展的 MPP 全并行計(jì)算，支持行存、列存、行列共存等多種存儲(chǔ)模式，同時(shí)支持多種索引類型。通過(guò)分布式處理 SQL 以及向量化的算子，能夠?qū)?CPU 資源發(fā)揮到極致，從而支持海量數(shù)據(jù)亞秒級(jí)分析，無(wú)需預(yù)計(jì)算，就能支持實(shí)時(shí)多維分析、即席分析等多種實(shí)時(shí) OLAP 分析的場(chǎng)景，再直接無(wú)縫對(duì)接上層應(yīng)用/服務(wù)，滿足所見(jiàn)即所得的分析體驗(yàn)。

下圖為 Hologres 128C 實(shí)例下，TPCH 100G 標(biāo)準(zhǔn)數(shù)據(jù)集下的測(cè)試結(jié)果，橫軸表示 query，縱軸是響應(yīng)時(shí)間：

2.3 高性能在線服務(wù)

隨著實(shí)時(shí)數(shù)倉(cāng)的廣泛應(yīng)用，越來(lái)越多的企業(yè)把實(shí)時(shí)數(shù)倉(cāng)作為在線服務(wù)系統(tǒng)提供在線查詢。Hologres 作為 HSAP（Hybrid Serving and Analytics Processing, 服務(wù)與分析一體化）的最佳落地實(shí)踐，除了具備處理分析型 Query 的能力外，還具備十分強(qiáng)大的在線服務(wù) Serving 能力(高 QPS 點(diǎn)查)，例如 KV 點(diǎn)查與向量檢索。在 KV 點(diǎn)查場(chǎng)景中，Holgres 通過(guò) SQL 接口可以支持百萬(wàn)級(jí)的 QPS 吞吐與極低的延時(shí)。通過(guò) Hologres 能夠做到一套系統(tǒng)、一份數(shù)據(jù)支持同時(shí) OLAP 分析和在線服務(wù)兩種場(chǎng)景，簡(jiǎn)化數(shù)據(jù)架構(gòu)。

下圖為 Hologres 128C 實(shí)例下，CPU 消耗 25%的點(diǎn)查測(cè)試性能：

2.4 讀寫(xiě)分離高可用

實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù) Hologres 提供高 QPS 低延遲的寫(xiě)入能力，支持在線服務(wù)的查詢場(chǎng)景，還支持復(fù)雜的多維分析 OLAP 查詢。當(dāng)不同類型，不同復(fù)雜的任務(wù)請(qǐng)求到 Hologres 實(shí)例上時(shí)，Hologres 不僅需要確保任務(wù)的正常運(yùn)行，還要確保系統(tǒng)的穩(wěn)定性。當(dāng)前 Hologres 支持通過(guò)共享存儲(chǔ)的一主多從子實(shí)例的高可用架構(gòu)，實(shí)現(xiàn)了完整的讀寫(xiě)分離功能，保障 不同業(yè)務(wù)場(chǎng)景的 SLA。

1.讀寫(xiě)分離：實(shí)現(xiàn)了完整的讀寫(xiě)分離功能，保障不同業(yè)務(wù)場(chǎng)景的 SLA，在高吞吐的數(shù)據(jù)寫(xiě)入和復(fù)雜的 ETL 作業(yè)、OLAP 查詢、AdHoc 查詢、在線服務(wù)等場(chǎng)景中，系統(tǒng)負(fù)載物理上完全隔離，不會(huì)因?qū)懭肴蝿?wù)產(chǎn)生了查詢?nèi)蝿?wù)的抖動(dòng)。

2.多類型負(fù)載資源隔離：一個(gè)主實(shí)例可以配置四個(gè)只讀實(shí)例，實(shí)例之間可以根據(jù)業(yè)務(wù)情況配置不同規(guī)格，系統(tǒng)負(fù)載物理上完全隔離，避免相互影響而帶來(lái)抖動(dòng)。

3.實(shí)例間數(shù)據(jù)毫秒級(jí)異步同步延遲：P99 5ms 內(nèi)。

2.5 Binlog 訂閱

類似于傳統(tǒng)數(shù)據(jù)庫(kù) MySQL 中的 Binlog 概念，Binlog 用來(lái)記錄數(shù)據(jù)庫(kù)中表數(shù)據(jù)的修改記錄，比如 Insert/Delete/Update 的操作。在 Hologres 中，表的 Binlog 是一種強(qiáng) Schema 格式的數(shù)據(jù)，Binlog 記錄的序列號(hào)（BigInt），在單 shard 內(nèi)單調(diào)遞增，類似于 Kafka 中的 Offset 概念。通過(guò)阿里云 Flink 消費(fèi) Hologres Binlog，可以實(shí)現(xiàn)數(shù)倉(cāng)分層間的全鏈路實(shí)時(shí)開(kāi)發(fā)，在分層治理的前提下，縮短數(shù)據(jù)加工端到端延遲，同時(shí)提升實(shí)時(shí)數(shù)倉(cāng)分層的開(kāi)發(fā)效率。

阿里云 Flink x Hologres 一站式企業(yè)級(jí)實(shí)時(shí)數(shù)倉(cāng)解決方案

3.1 實(shí)時(shí)數(shù)倉(cāng) ETL

ETL( Extract-Transform-Load)是比較傳統(tǒng)的數(shù)據(jù)倉(cāng)庫(kù)建設(shè)方法，業(yè)務(wù)庫(kù)的數(shù)據(jù) Binlog 經(jīng)過(guò)阿里云 Flink 的 ETL 處理之后，數(shù)據(jù)寫(xiě)入到實(shí)時(shí)數(shù)倉(cāng) Hologres 中，然后進(jìn)行各類數(shù)據(jù)查詢分析。ETL 的方法核心是需要在數(shù)倉(cāng)中具備完善的數(shù)倉(cāng)模型分層，通常按照 ODS（Operational Data Source）> DWD（Data Warehouse Detail）> DWS（Data Warehouse Summary）> ADS（Application Data Service）分層，整個(gè)數(shù)倉(cāng)鏈路比較完善。

在這個(gè)鏈路中，需要將數(shù)據(jù)源比如 MySQL 的 Binlog 數(shù)據(jù)通過(guò)阿里云 Flink CDC 同步到消息隊(duì)列 Kafka，再通過(guò)阿里云 Flink 將 ODS 的數(shù)據(jù)進(jìn)行過(guò)濾，清洗，邏輯轉(zhuǎn)化等操作，形成對(duì)不同的業(yè)務(wù)主題模型的 DWD 數(shù)據(jù)明細(xì)層，同時(shí)將數(shù)據(jù)發(fā)送到 Kafka 集群，之后再通過(guò)阿里云 Flink 將 DWD 的數(shù)據(jù)進(jìn)行輕度的匯總操作，形成業(yè)務(wù)上更加方便查詢的 DWS 輕度匯總層數(shù)據(jù)，再將數(shù)據(jù)寫(xiě)入 Kafka 集群。最后再面向業(yè)務(wù)具體的應(yīng)用層的需求，在 DWS 層基礎(chǔ)上通過(guò)阿里云 Flink 實(shí)時(shí)處理形成 ADS 數(shù)據(jù)應(yīng)用層，寫(xiě)入實(shí)時(shí)數(shù)倉(cāng) Hologres 進(jìn)行存儲(chǔ)和分析，支持業(yè)務(wù)各種不同類型的報(bào)表，畫(huà)像等業(yè)務(wù)場(chǎng)景。

實(shí)時(shí)數(shù)倉(cāng) ETL 的處理優(yōu)點(diǎn)是數(shù)倉(cāng)各種層次比較完備，職責(zé)清晰，但是缺點(diǎn)是 Flink 結(jié)合 Kafka 集群維護(hù)復(fù)雜，處理鏈路比較長(zhǎng)，歷史數(shù)據(jù)修正復(fù)雜，ADS 應(yīng)用層的數(shù)據(jù)實(shí)時(shí)性會(huì)弱，其次數(shù)據(jù)在各個(gè) Kafka 中不便于查詢，不便于檢查數(shù)據(jù)質(zhì)量，也不便于實(shí)現(xiàn) schema 的動(dòng)態(tài)變化。

3.2 實(shí)時(shí)數(shù)倉(cāng) ELT

隨著業(yè)務(wù)對(duì)數(shù)據(jù)的時(shí)效性要求越來(lái)越高時(shí)，相較于 ETL 復(fù)雜繁雜的處理鏈路，業(yè)務(wù)需要更快速的將數(shù)據(jù)實(shí)時(shí)入倉(cāng)，因此 ELT 變成了比較流行的處理方法。ELT 是英文 Extract-Load-Transform 的縮寫(xiě)，我們可將 ELT 理解為一個(gè)數(shù)據(jù)遷移集成的過(guò)程。在這個(gè)過(guò)程中，我們可以對(duì)數(shù)據(jù)源關(guān)系型數(shù)據(jù)庫(kù)比如 MySQL、PostgresSQL 和非關(guān)系型數(shù)據(jù)庫(kù)比如 HBase、Cassandra 等業(yè)務(wù)庫(kù)的 Binlog，消息隊(duì)列比如 Datahub、Kafka 中的埋點(diǎn)采集日志等數(shù)據(jù)，經(jīng)過(guò)阿里云 Flink 實(shí)時(shí)抽取，然后加載到 Hologres 中進(jìn)行相關(guān)的 OLAP 分析和在線服務(wù)。

在這個(gè)鏈路中，阿里云 Flink 負(fù)責(zé)數(shù)據(jù)的實(shí)時(shí)入倉(cāng)以及數(shù)據(jù)的清洗關(guān)聯(lián)，清洗后的數(shù)據(jù)實(shí)時(shí)寫(xiě)入 Hologres，由 Hologres 直接存儲(chǔ)明細(xì)數(shù)據(jù)。在 Hologres 中可以簡(jiǎn)化分層，以明細(xì)層為主，按需搭配其他匯總層，通過(guò) Hologres 強(qiáng)大的數(shù)據(jù)處理能力直接對(duì)接報(bào)表、應(yīng)用等上層查詢服務(wù)。上層的分析 SQL 無(wú)法固化，通常在 ADS 層以邏輯視圖（View）封裝 SQL 邏輯，上層應(yīng)用直接查詢封裝好的 View，實(shí)現(xiàn)即席查詢。

實(shí)時(shí)數(shù)倉(cāng)中采取 ELT 的方式進(jìn)行建設(shè)，會(huì)給數(shù)據(jù)和業(yè)務(wù)帶來(lái)比較大的收益，詳細(xì)如下：

●靈活性：將原始的業(yè)務(wù)數(shù)據(jù)直接入倉(cāng)，形成 ODS 層的數(shù)據(jù)，在數(shù)倉(cāng)中通過(guò) View 可以靈活地對(duì)數(shù)據(jù)進(jìn)行轉(zhuǎn)換(Transformation)的處理，View 可以隨時(shí)根據(jù)業(yè)務(wù)進(jìn)行調(diào)整。

●成本低：數(shù)據(jù)倉(cāng)庫(kù)的架構(gòu)比較清晰，鏈路比較短，運(yùn)維成本比較低。

●指標(biāo)修正簡(jiǎn)單：上層都是 View 邏輯封裝，只需要更新底表的數(shù)據(jù)即可，無(wú)需逐層修正數(shù)據(jù)。

但是該方案也存在一些缺點(diǎn)，當(dāng) View 的邏輯較為復(fù)雜，數(shù)據(jù)量較多時(shí)，查詢性能較低。因此比較適合于數(shù)據(jù)來(lái)源于數(shù)據(jù)庫(kù)和埋點(diǎn)系統(tǒng)，對(duì) QPS 要求不高，對(duì)靈活性要求比較高，且計(jì)算資源較為充足的場(chǎng)景。

3.3 實(shí)時(shí)數(shù)倉(cāng)分層（Streaming Warehouse 方案）

按照傳統(tǒng)數(shù)倉(cāng)的開(kāi)發(fā)方法論，采用 ODS>DWD>DWS>ADS 開(kāi)發(fā)的方法，通過(guò)阿里云 Flink 和 Hologres Binlog 的組合關(guān)系，支持層與層之間有狀態(tài)的全鏈路事件實(shí)時(shí)驅(qū)動(dòng)。在該方案中，數(shù)據(jù)通過(guò)阿里云 Flink CDC 實(shí)時(shí)入倉(cāng)至 Hologres，再通過(guò)阿里云 Flink 訂閱 Hologres Binlog，實(shí)現(xiàn)數(shù)據(jù)在不同層次之間的連續(xù)加工，最后寫(xiě)入 Hologres 對(duì)接應(yīng)用查詢。

通過(guò)這個(gè)方案，Hologres 可以達(dá)到像 Kafka、Datahub 等消息隊(duì)列同等的能力，增加數(shù)據(jù)復(fù)用的能力，一個(gè) Table 的數(shù)據(jù)既可以提供給下游阿里云 Flink 任務(wù)消費(fèi)，還可以對(duì)接上游 OLAP/在線服務(wù)查詢，不僅節(jié)省了成本，還簡(jiǎn)化數(shù)倉(cāng)架構(gòu)，同時(shí)也讓數(shù)倉(cāng)中的每一個(gè)層次都可以實(shí)時(shí)構(gòu)建、實(shí)時(shí)查詢，提升數(shù)據(jù)的流轉(zhuǎn)效率。

3.4 流批一體數(shù)倉(cāng)

在實(shí)時(shí)數(shù)倉(cāng)中，流計(jì)算任務(wù)和批處理任務(wù)都是分兩條工作流進(jìn)行開(kāi)發(fā)的，也即是 Kappa 架構(gòu)模式。在這套數(shù)倉(cāng)架構(gòu)中，會(huì)存在人力成本過(guò)高，數(shù)據(jù)鏈路冗余，數(shù)據(jù)口徑不一致，開(kāi)發(fā)效率低下的一些問(wèn)題。

為了解決這些問(wèn)題，阿里云 Flink+Hologres 提供了流批一體的能力。在該場(chǎng)景中，將輸入層統(tǒng)一變成 Hologres，通過(guò)一套業(yè)務(wù)邏輯代碼達(dá)到流和批處理的能力，其中 Flink SQL 的 Stream 任務(wù)消費(fèi) Hologres Binlog 提供流式處理，F(xiàn)link SQL 的 Batch 任務(wù)讀取 Hologres 表的原始數(shù)據(jù)達(dá)到批處理能力，經(jīng)過(guò) Flink 統(tǒng)一的計(jì)算處理之后，統(tǒng)一寫(xiě)入存儲(chǔ)至 Hologres。

阿里云 Flink 結(jié)合 Hologres 的流批一體技術(shù)，統(tǒng)一了數(shù)據(jù)輸入層、實(shí)時(shí)離線計(jì)算層和數(shù)據(jù)分析存儲(chǔ)層，極大的提升了數(shù)據(jù)開(kāi)發(fā)的效率，保證了數(shù)據(jù)的質(zhì)量。

典型應(yīng)用場(chǎng)景

阿里云 Flink 與 Hologres 深度集成，助力企業(yè)快速構(gòu)建一站式實(shí)時(shí)數(shù)倉(cāng)：

●可通過(guò)阿里云 Flink 實(shí)時(shí)寫(xiě)入 Hologres，高性能寫(xiě)入與更新，數(shù)據(jù)寫(xiě)入即可見(jiàn)，無(wú)延遲，滿足實(shí)時(shí)數(shù)倉(cāng)高性能低延遲寫(xiě)入需求；

●可通過(guò)阿里云 Flink 的全量讀取、Binlog 讀取、CDC 讀取、全增量一體化等多種方式，讀取 Hologres 源表數(shù)據(jù)，無(wú)需額外組件，統(tǒng)一計(jì)算和存儲(chǔ)，加速數(shù)據(jù)流轉(zhuǎn)效率；

●可通過(guò)阿里云 Flink 讀取 Hologres 維表，助力高性能維表關(guān)聯(lián)、數(shù)據(jù)打?qū)挼榷喾N應(yīng)用場(chǎng)景；

●阿里云 Flink 與 Hologres 元數(shù)據(jù)打通，通過(guò) Hologres Catalog，實(shí)現(xiàn)元數(shù)據(jù)自動(dòng)發(fā)現(xiàn)，極大提升作業(yè)開(kāi)發(fā)效率和正確性。

通過(guò)阿里云 Flink 與 Hologres 的實(shí)時(shí)數(shù)倉(cāng)標(biāo)準(zhǔn)解決方案，能夠支撐多種實(shí)時(shí)數(shù)倉(cāng)應(yīng)用場(chǎng)景，如實(shí)時(shí)推薦、實(shí)時(shí)風(fēng)控等，滿足企業(yè)的實(shí)時(shí)分析需求。下面我們將會(huì)介紹阿里云 Flink + Hologres 的典型應(yīng)用場(chǎng)景，助力業(yè)務(wù)更加高效的搭建實(shí)時(shí)數(shù)倉(cāng)。

4.1 海量數(shù)據(jù)實(shí)時(shí)入倉(cāng)

實(shí)時(shí)數(shù)倉(cāng)搭建的第一步便是海量數(shù)據(jù)的實(shí)時(shí)入倉(cāng)，基于阿里云 Flink CDC 可以簡(jiǎn)單高效地將海量數(shù)據(jù)同步到實(shí)時(shí)數(shù)倉(cāng)中，并能將增量數(shù)據(jù)以及表結(jié)構(gòu)變更實(shí)時(shí)同步到數(shù)倉(cāng)中。而整個(gè)流程只需在阿里云 Flink 上定義一條 CREATE DATABASE AS DATABASE 的 SQL 即可（詳細(xì)步驟可參考 實(shí)時(shí)入倉(cāng)快速入門(mén)[4]）。經(jīng)測(cè)試，對(duì)于 MySQL 中的 TPC-DS 1T 數(shù)據(jù)集，使用阿里云 Flink 64 并發(fā)，只需 5 小時(shí)便能完全同步到 Hologres，TPS 約 30 萬(wàn)條/秒。在增量 Binlog 同步階段，使用阿里云 Flink 單并發(fā)，同步性能達(dá)到 10 萬(wàn)條/秒。

4.2 雙流 Join

數(shù)據(jù)實(shí)時(shí)入倉(cāng)形成了 ODS 層的數(shù)據(jù)后，通常需要將事實(shí)數(shù)據(jù)與維度數(shù)據(jù)利用 Flink 多流 Join 的能力實(shí)時(shí)地打平成寬表，結(jié)合 Hologres 寬表極佳的多維分析性能，助力上層業(yè)務(wù)查詢提速。阿里云 Flink 支持以全增量一體化的模式讀取 Hologres 表，即先讀取全量數(shù)據(jù)再平滑切換到讀取 CDC 數(shù)據(jù)，整個(gè)過(guò)程保證數(shù)據(jù)的不重不丟。因此基于阿里云 Flink 可以非常方便地實(shí)時(shí)加工和打?qū)?Hologres 的 ODS 層數(shù)據(jù)，完成 DWD 層的寬表模型構(gòu)建。

4.3 寬表 Merge

數(shù)據(jù)倉(cāng)庫(kù)中我們通常需要關(guān)心的就是建模，數(shù)據(jù)模型通常分為四種：寬表模型、星型模型、雪花模型、星座模型（Hologres 均支持），在這里我們重點(diǎn)要提到的是寬表模型的建設(shè)。寬表模型通常是指將業(yè)務(wù)主體相關(guān)的指標(biāo)、維表、屬性關(guān)聯(lián)在一起的模型表，也可以泛指將多個(gè)事實(shí)表和多個(gè)維度表相關(guān)聯(lián)到一起形成的寬表。

寬表建設(shè)通常的做法就是通過(guò)阿里云 Flink 的雙流 Join 來(lái)實(shí)現(xiàn)，包括 Regular Join，Interval Join，Temporal Join。對(duì)于主鍵關(guān)聯(lián)的場(chǎng)景（即 Join 條件分別是兩條流的主鍵），我們可以將 Join 的工作下沉到 Hologres 去做，通過(guò) Hologres 的局部更新功能來(lái)實(shí)現(xiàn)寬表 Merge，從而省去了 Flink Join 的狀態(tài)維護(hù)成本。比如廣告場(chǎng)景中，一個(gè) Flink 任務(wù)處理廣告曝光數(shù)據(jù)流，統(tǒng)計(jì)每個(gè)產(chǎn)品的曝光量，以產(chǎn)品 ID 作為主鍵，更新到產(chǎn)品指標(biāo)寬表中。同時(shí)，另一個(gè) Flink 任務(wù)處理廣告點(diǎn)擊數(shù)據(jù)流，統(tǒng)計(jì)每個(gè)產(chǎn)品的點(diǎn)擊量，也以產(chǎn)品 ID 作為主鍵，更新到產(chǎn)品指標(biāo)寬表中。整個(gè)過(guò)程不需要進(jìn)行雙流 Join，最終 Hologres 會(huì)自己完成整行數(shù)據(jù)的組裝。基于得到的產(chǎn)品指標(biāo)寬表，用戶可以方便地在 Hologres 進(jìn)行廣告營(yíng)銷的分析，例如計(jì)算產(chǎn)品的 CTR=點(diǎn)擊數(shù)/曝光數(shù)。下圖和代碼示例展示了如何從雙流 Join 改為寬表 Merge。

CREATE TABLE ods_ad_click (

product_id INT,

click_id BIGINT,

click_time TIMESTAMP

) WITH ('connector'='datahub', 'topic'='..');

CREATE TABLE ods_ad_impressions (

product_id INT,

imp_id BIGINT,

imp_time TIMESTAMP

) WITH ('connector'='datahub', 'topic'='..');

CREATE TABLE dws_ad_product (

product_id INT,

click_cnt BIGINT,

imp_cnt BIGINT,

PRIMARY KEY (product_id) NOT ENFORCED

) WITH ('connector'='hologres','insertOrUpdate'='true');

INSERT INTO dws_ad_product (product_id, click_cnt)

SELECT product_id, COUNT(click_id) as click_cnt

FROM ods_ad_click

GROUP BY product_id;

INSERT INTO dws_ad_product (product_id, imp_cnt)

SELECT product_id, COUNT(imp_id) AS imp_cnt

FROM ods_ad_impressions

GROUP BY product_id;

使用 Hologres 寬表的 Merge 能力，不僅可以提升流作業(yè)的開(kāi)發(fā)效率，還能減少流作業(yè)所需要的資源消耗，也能夠更容易的維護(hù)各個(gè)流作業(yè)，讓作業(yè)之間不會(huì)相互影響。但需要注意的是，寬表 Merge 僅限于使用在主鍵關(guān)聯(lián)的場(chǎng)景，并不適用于數(shù)倉(cāng)中常見(jiàn)的星型模型和雪花模型，所以在大部分場(chǎng)景仍需使用 Flink 的雙流 Join 來(lái)完成寬表建模。

4.4 實(shí)時(shí)維表 Lookup

在實(shí)時(shí)數(shù)倉(cāng)中，在構(gòu)建 DWD 層的數(shù)據(jù)過(guò)程中，一般都是通過(guò)阿里云 Flink 來(lái)讀取消息隊(duì)列比如 Datahub 上的 ODS 數(shù)據(jù)，同時(shí)需要關(guān)聯(lián)維表來(lái)形成 DWD 層。在阿里云 Flink 的計(jì)算過(guò)程中，需要高效的讀取維表的能力，Hologres 可以通過(guò)高 QPS 低延遲的點(diǎn)查能力來(lái)滿足實(shí)現(xiàn)這類場(chǎng)景需求。比如我們需要通過(guò) ODS 的數(shù)據(jù)去 Join 維表形成 DWD 層的時(shí)候，就可以利用 Hologres 提供的點(diǎn)查能力，在該模式中，通常使用行存表的主鍵點(diǎn)查模式提高維表的 Lookup 效率。具體的實(shí)現(xiàn)類似如下：

典型用戶案例

依托阿里云 Flink+Hologres 解決方案，企業(yè)可以快速構(gòu)建一站式實(shí)時(shí)數(shù)倉(cāng)，助力實(shí)時(shí)推薦、實(shí)時(shí)風(fēng)控、實(shí)時(shí)大屏等多種業(yè)務(wù)場(chǎng)景，實(shí)現(xiàn)對(duì)數(shù)據(jù)的快速處理，極速探索查詢。目前該方案已在阿里巴巴內(nèi)部、眾多云上企業(yè)生產(chǎn)落地，成為實(shí)時(shí)數(shù)倉(cāng)的最佳解決方案之一。

以某知名全球 TOP20 游戲公司業(yè)務(wù)為例，其通過(guò)阿里云 Flink+Hologres 實(shí)時(shí)數(shù)倉(cāng)方案，替換開(kāi)源 Flink+Presto+HBase+ClickHouse 架構(gòu)，簡(jiǎn)化數(shù)據(jù)處理鏈路、統(tǒng)一數(shù)倉(cāng)架構(gòu)、統(tǒng)一存儲(chǔ)、查詢性能提升 100%甚至更多，完美支撐數(shù)據(jù)分析、廣告投放、實(shí)時(shí)決策等多個(gè)場(chǎng)景，助力業(yè)務(wù)快速增長(zhǎng)。

5.1 業(yè)務(wù)困難：ETL 鏈路復(fù)雜、OLAP 查詢慢

客戶原數(shù)倉(cāng)架構(gòu)使用全套開(kāi)源組件，架構(gòu)圖如下。其中開(kāi)源 Flink 做 ETL 處理，處理后寫(xiě)入 ClickHouse、Starocks 等 OLAP 引擎。

這套架構(gòu)遇見(jiàn)的主要痛點(diǎn)有：

1、ETL 鏈路復(fù)雜

●為了解決數(shù)據(jù)實(shí)時(shí) ETL，客戶通過(guò) Flink CDC + Hudi 做流批一體。但由于上游業(yè)務(wù)數(shù)據(jù)經(jīng)常變更表結(jié)構(gòu)，而開(kāi)源 Flink CDC 缺乏 Schema Evolution 的能力，每次表結(jié)構(gòu)變更都需要任務(wù)重新啟動(dòng)，操作非常麻煩，浪費(fèi)大量開(kāi)發(fā)時(shí)間。

●Hudi 的查詢性能不滿足業(yè)務(wù)需求，還需要再加一個(gè) Presto 做加速查詢，造成鏈路冗余。

2、OLAP 架構(gòu)冗余，查詢慢

客戶主要是靠買量發(fā)行作為游戲推廣的重要手段，為了解決廣告歸因的實(shí)時(shí)決策場(chǎng)景對(duì)查詢加速的需要，于是部署了開(kāi)源 Presto、ClickHouse、HBase 等多套集群搭建混合 OLAP 平臺(tái)。帶來(lái)的問(wèn)題有：

●平臺(tái)需要維護(hù)多套集群，導(dǎo)致運(yùn)維變得非常復(fù)雜。

●開(kāi)發(fā)需要在各種 SQL 中切換，為開(kāi)發(fā)團(tuán)隊(duì)帶來(lái)了許多困擾。由于 ClickHouse 缺乏主鍵，在歸因分析時(shí)需要使用 Last Click 模型，帶來(lái)了大量的額外工作。

●同時(shí) OLAP 引擎的查詢性能沒(méi)有辦法很好的滿足業(yè)務(wù)需求，沒(méi)辦法根據(jù)數(shù)據(jù)實(shí)時(shí)決策。

●數(shù)據(jù)需要在多個(gè) OLAP 系統(tǒng)中存儲(chǔ)，造成存儲(chǔ)冗余，導(dǎo)致成本壓力劇增。

基于上面的痛點(diǎn)，客戶開(kāi)始重新做技術(shù)選型，并使用阿里云 Flink+Hologres 來(lái)替換現(xiàn)有的開(kāi)源數(shù)倉(cāng)架構(gòu)。

5.2 架構(gòu)升級(jí)：阿里云 Flink+Hologres 統(tǒng)一數(shù)據(jù)存儲(chǔ)與服務(wù)

通過(guò)阿里云 Flink+Hologres 替換后的數(shù)據(jù)鏈路如下：

●數(shù)據(jù)源數(shù)據(jù)通過(guò) Flink CDC 能力寫(xiě)入 Kafka 做前置清洗，清洗后通過(guò)阿里云 Flink 進(jìn)行 ETL 處理。

●阿里云 Flink 經(jīng)過(guò) ETL 后的數(shù)據(jù)實(shí)時(shí)寫(xiě)入 Hologres，通過(guò) Hologres 替換了 Kafka 作為實(shí)時(shí)數(shù)倉(cāng)的中間數(shù)據(jù)層，統(tǒng)一了流批存儲(chǔ)。

●在 Hologres 中根據(jù) ODS > DWD > DWS 層匯總加工。在 ODS 層，阿里云 Flink 訂閱 Hologres Binlog，計(jì)算后寫(xiě)入 Hologres DWD 層，DWD 層在 Hologres 中匯總成 DWS 層，最后 DWS 對(duì)接上層報(bào)表和數(shù)據(jù)服務(wù)等業(yè)務(wù)。

●為了存儲(chǔ)的統(tǒng)一，也將原離線 Hive 數(shù)據(jù)替換成阿里云 MaxCompute，以 MaxCompute 為離線主要鏈路。因 Hologres 與 MaxCompute 的高效互通能力，Hologres 通過(guò)外表離線加速查詢 MaxCompute，并將歷史數(shù)據(jù)定期歸檔至 MaxCompute。

5.3 業(yè)務(wù)收益：架構(gòu)統(tǒng)一，性能提升 100%

通過(guò)架構(gòu)升級(jí)后，客戶的顯著業(yè)務(wù)收益如下：

●依托阿里云 Flink+Hologres，數(shù)據(jù)可以實(shí)時(shí)寫(xiě)入 Hologres，寫(xiě)入即可見(jiàn)，并且 Hologres 有主鍵，能夠支撐高性能的寫(xiě)入更新能力，百萬(wàn)級(jí)更新毫秒級(jí)延遲。

●阿里云 Flink 提供 Schema Evolution 的能力，自動(dòng)感知上游表結(jié)構(gòu)變更并同步 Hologres，改造后的實(shí)時(shí) ETL 鏈路通過(guò)訂閱 Hologres Binlog 日志來(lái)完成，降低鏈路維護(hù)成本。

●通過(guò) Hologres 統(tǒng)一了數(shù)據(jù)查詢出口，經(jīng)過(guò)客戶實(shí)測(cè)，Hologres 可以達(dá)到毫秒級(jí)延遲，相比開(kāi)源 ClickHouse 性能提升 100%甚至更多，JOIN 查詢性能快 10 倍。

●升級(jí)后數(shù)倉(cāng)架構(gòu)變得更加靈活簡(jiǎn)潔，統(tǒng)一了存儲(chǔ)，只需要一套系統(tǒng)就能滿足業(yè)務(wù)需求，降低運(yùn)維壓力和運(yùn)維成本。