簡介： 本次內容將會介紹使用Flink和Hologres，實現可擴展的、高效的、云原生實時數倉。

一、Hologres生態

從前面幾篇的內容，相信大家已經了解到Hologres是一款兼容PostgreSQL協議的實時交互式分析產品。在生態的兼容性上，Hologres有著非常龐大的生態家族，如下圖所示，

• 對于開源大數據領域，Hologres支持當下最流行的大數據開源組件，其中包括
• 對于埋點類數據，支持Blink/Flink/Spark/數據集成等大數據工具進行高性能的實時導入和批量導入
• 對于數據庫類的數據，通過和Dataworks數據集成(DataX和StreamX)共建實現方便高效的數據庫整庫實時鏡像到Hologres中，并滿足金融企業對可管理性、監控、網絡等的需求

無論是實時數據，還是離線數據接入Hologres之后，接下來就能使用Hologres對數據進行分析。最常見的就是使用JDBC或者ODBC對數據進行查詢、分析、監控，然后承接上游的業務，比如說大屏、報表、應用等各種場景。

同時再為大家介紹一下DataWorks，它是阿里云的一個數據開發平臺，提供了數據集成、數據地圖、數據服務等功能。數據集成主要功能可以將數據庫的數據導入Hologres，其中同步的方式包括離線同步和實時同步，離線同步支持幾十種異構數據源同步至Hologres，而實時同步當前主要支持以下幾種：

• Mysql Binlog：通過訂閱Biblog的方式將mysql數據實時寫入Hologres
• Oracle CDC：全稱是Change Data Capture，也是一個類似Mysql Binlog的用來獲取Oracle表change log的方式
• Datahub：是阿里巴巴自研的一個分布式高性能消息隊列，值得一提的是，Datahub自身也提供了直接將數據實時導入至Hologres的功能，無需經過Dataworks
• PolarDB：是阿里巴巴自主研發的關系型分布式云原生數據庫

二、Hologres實時導入接口介紹

接下來為大家介紹一下Hologres提供的一個實時導入的接口，以及接口的技術原理。

1)實時導入接口

Hologres實時導入接口的具備以下特性：

• 行存&列存都支持
• 支持根據主鍵去重 (Exactly once)
• 支持整行數據局部更新
• 導入即可見，毫秒級延遲
• 單Core 2W+ RPS (TPCH PartSupp表)
• 性能隨資源線性擴展
• 支持分區表寫入

2)實時導入原理

實時導入的原理如下圖所示，首先我們看一下該圖的最上面的幾個節點，代表了數據的上游，也就是業務層。如何將數據導入Hologres，主要有兩種場景：

• 使用SQL進行數據的導入(最常見)

例如使用JDBC執行insert語句，該insert語句會經過一個負載均衡服務器路由分發至我們的Frontend節點，對該insert語句進行SQL的解析優化，然后生成一個優化后的執行計劃，并將該執行計劃分發至后端的worker節點。worker節點收到該執行計劃之后，就會將該數據完成寫入。

• Connector寫入

另外一條鏈路為左邊的Private API鏈路，也就是當前Apache Flink或者Apache Spark Connector所使用的Hologres的實時導入接口。該Private API提供的數據接口和普通sql請求不一樣，而是我們稱之為Fixed Plan的請求接口，這些請求被分發至負載均衡服務器之后，負載均衡服務器會將數據路由分發至一個叫做Private API Service的節點。該節點將數據寫入請求分發至worker節點，也就是后端的節點。當worker節點收到，無論是Fixed Plan，還是執行計劃之后，會對數據進行持久化，最終數據完成寫入。

接著來更進一步理解Private API Service的一個數據分發功能。如下圖所示，一張表的數據分布在多個Shard上，一條記錄只會屬于一個Shard，根據Distribution key屬性進行Hash。

當實時寫入的數據請求到達后端的worker節點之后，worker節點是怎么處理的。如下圖所示，這一塊有如下特點：

• Log Structured Merge Tree(LSM)
• 全異步框架，協程(Coroutine)
• 基于Masstree的Memtable

同時上面也提到通過SQL來進行數據的寫入是最常見的場景，Hologres也在后端優化了整個SQL的寫入鏈路。例如對于Insert into values，Insert into on conflict do update，Select from where pk = xxx等場景簡單的SQL，Hologres會進行優化，減少SQL的解析和優化過程，提升整個數據寫入和查詢的性能。

三、Hologres實時讀寫場景

前面介紹了Hologres通過connector寫的原理，下面將會介紹Flink+Hologres最常見的寫入場景。

1)實時寫入場景

最常見的第一種就是實時寫入場景。實時寫入分為幾種。

• 第一種，Hologres的結果表沒有設置主鍵，這樣Flink實時接入就是一種Append Only的模式進行寫入。當上游數據發生重復，或者Flink任務作業失敗，上游數據會需要進行回溯，這時候下游數據錄入到Hologres中就會產生重復的數據。這種情形對于日志型數據是比較合理的，因為用戶并不需要關心數據是否需要進行去重
• 第二種，Hologres的結果表設置了主鍵。Flink或者其它實時寫入就會按照行的主鍵進行更新。主鍵更新的意思就是說對于相同主鍵的兩行數據，后到的數據會完全覆蓋掉之前已經到達的數據。
• 第三種，是按照主鍵去重。就是說后到的數據會被忽略掉，只保留最早到的一條記錄。這種場景用戶并不關心主鍵的更新情況，只需要保證主鍵的去重。

2)寬表Merge場景

例如一個用戶的結果表有非常多的字段，會有上百列，而該表的許多字段可能同時分布在不同的數據上游，例如，Column C和D分布在一個kafka的topic A上面，Column E和F分布在kafka的topic B上面，用戶希望消費兩個kafka topic，并將數據merge成Hologres的一張結構表。最常見的解決辦法是，進行流場景的一個雙流Join。這種實現對于開發人員來說相對比較復雜，需要實現一個雙流Join，而且理論上來說會對計算資源要求非常大，也加劇了運維人員的負擔。

而Hologres針對這種場景是如何實現的呢？
Hologres支持局部更新的功能。如下圖所示，按照這種實現方式，只需要兩個流各自寫入Hologres結果表。第一個流消費ABCD四個字段，將數據寫入到最終的結果表中。第二個流消費ABEF四個字段，最終將數據寫入到結果表，并不需要進行雙流的Join，最終Hologres會自己進行一個數據的組裝。第一個流寫入ABCD的時候并不會去更新已經存在的EF字段，同樣，第二個流寫入ABEF字段的時候，C和D字段已經存在，不會被更新，最終達到完整的一個數據Merge的功能。使用這種功能，可以大大提升流作業的開發效率，以及減少流作業所需要的資源消耗，也能夠更容易的維護各個流作業。

3)實時維表Join場景

除了寫場景，Hologres也支持讀場景，最常見的是使用Hologres的行存表來進行點查。如下圖所示，是一個實時維表的Join場景。主要邏輯是生成一個數據源，會不停的生成一個數據流，和Hologres的維表進行Join，打寬數據流，最終將數據寫入到一個結果表中。在實際業務中，這種使用場景通常會用來替換HBase，以達到更好的性能和更低的成本。

4)Hologres Binlog場景
如下圖所示，以消息隊列方式讀取Hologres數據的Change log。 其中：

• Binlog系統字段，表示Binlog序號，Shard內部單調遞增不保證連續，不同Shard之間不保證唯一和有序
• Binlog系統字段，表示當前 Record 所表示的修改類型
• UPDATE操作會產生兩條Binlog記錄，一條更新前，一條更新后的。訂閱Binlog功能會保證這兩條記錄是連續的且更新前的Binlog記錄在前，更新后的Binlog記錄在后

原文鏈接：https://developer.aliyun.com/article/778798

總結

以上是生活随笔為你收集整理的怎样不停请求接口实现实时刷新_Hologres+Flink实时数仓详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。