一、準(zhǔn)實時數(shù)據(jù)平臺

光大銀行數(shù)據(jù)資產(chǎn)部以提升服務(wù)和經(jīng)營效率為核心目標(biāo)，負(fù)責(zé)光大銀行的數(shù)據(jù)管理和數(shù)字化轉(zhuǎn)型工作，提供關(guān)鍵性的底層技術(shù)支撐。在整個數(shù)字化轉(zhuǎn)型和業(yè)務(wù)升級過程中，數(shù)據(jù)資產(chǎn)部支撐業(yè)務(wù)關(guān)鍵資源、保障關(guān)鍵基礎(chǔ)設(shè)施運行，在我行扮演著重要角色。

準(zhǔn)實時數(shù)據(jù)平臺則是光大銀行數(shù)據(jù)資產(chǎn)部-大數(shù)據(jù)平臺團隊下的重要項目。

準(zhǔn)實時數(shù)據(jù)平臺始于2016年研發(fā)，至今已有6-7年的演進，上圖是平臺的整體架構(gòu)圖，數(shù)據(jù)流向從左到右。

初期平臺的定位是準(zhǔn)實時數(shù)據(jù)采集與流式計算，向外提供準(zhǔn)實時技術(shù)支持和按需開發(fā)的流式數(shù)據(jù)加工等服務(wù)。基于此，平臺分為三個模塊：數(shù)據(jù)采集、數(shù)據(jù)標(biāo)準(zhǔn)化和數(shù)據(jù)發(fā)布。

簡單來說，我們的數(shù)據(jù)采集接近于數(shù)據(jù)貼源層的邏輯，整個流程更像一個實時的數(shù)據(jù)倉庫。數(shù)據(jù)源通過兩種方式被傳送到我們的Kafka貼源層：

• 第一種是傳統(tǒng)關(guān)系型數(shù)據(jù)庫的CDC數(shù)據(jù)，這些數(shù)據(jù)會通過我們的OGG工具進行同步并存儲于貼源層，另外，業(yè)務(wù)系統(tǒng)中的有異步導(dǎo)出需求且對延時敏感的數(shù)據(jù)通過Kafka API直接寫入到貼源層。這些數(shù)據(jù)大多不可被直接使用于下游業(yè)務(wù)應(yīng)用。因此在標(biāo)準(zhǔn)化模塊進行公共邏輯處理，數(shù)據(jù)會被傳遞到SparkStreaming或Flink實時流式任務(wù)，隨后被存儲于Kafka標(biāo)準(zhǔn)層。在標(biāo)準(zhǔn)層進行的處理后，其中的一部分?jǐn)?shù)據(jù)可以被直接用于下游應(yīng)用的消費。

• 另一種數(shù)據(jù)則通過各種相對復(fù)雜的計算邏輯（flink/spark）或批處理技術(shù)寫入到我們最終的發(fā)布層。與前兩層處理不同，第三層著重處理業(yè)務(wù)邏輯并將最終的數(shù)據(jù)提供給下游系統(tǒng)進行實時訂閱。

整個架構(gòu)經(jīng)過多年演進，展現(xiàn)出如下三個特點：

• 具有清晰的分層邏輯，并且基于實時數(shù)倉的設(shè)計邏輯；
• 該架構(gòu)涵蓋廣泛的組件，包括明顯的業(yè)務(wù)邏輯處理，初衷是提供流式數(shù)據(jù)加工架構(gòu)相關(guān)的服務(wù)；
• 兼具準(zhǔn)時數(shù)據(jù)的存儲平臺和計算資源服務(wù)提供平臺。

隨著大數(shù)據(jù)領(lǐng)域多年的發(fā)展，組件分工也更加細(xì)化。組件細(xì)分為大數(shù)據(jù)架構(gòu)提供更高的可維護性、可擴展性、安全性、穩(wěn)定性、性能、效率和易用性，使得整個架構(gòu)更加靈活和強大，能夠更好地滿足不斷增長的數(shù)據(jù)處理需求。

然而，反觀我們的系統(tǒng)，存在兩個問題：

• 首先，我們必須維護大量業(yè)務(wù)處理邏輯，隨著時間推移，這對我們平臺的資源分配和能力優(yōu)化帶來了巨大壓力；
• 其次，我們維護了大量組件，從導(dǎo)入層到存儲層、計算層、上層調(diào)度層，使得該平臺的定位相對較模糊，需要更加明確和細(xì)化。

因此，我們需要加強對架構(gòu)的管理，以縮小各個組件之間的差異，同時優(yōu)化架構(gòu)的組織和協(xié)調(diào)，使之更加適應(yīng)未來的需求。

二、架構(gòu)演進實踐

針對上述問題，我們結(jié)合整個準(zhǔn)實時數(shù)據(jù)平臺的組件架構(gòu)特點，將平臺拆分成了兩個子平臺：

• 實時數(shù)據(jù)湖平臺
• 數(shù)據(jù)服務(wù)總線平臺

整個準(zhǔn)實時數(shù)據(jù)平臺維護的相關(guān)組件從左到右，包括數(shù)據(jù)接入和導(dǎo)出層、消息中間件層、數(shù)據(jù)計算層以及數(shù)據(jù)存儲層。最上層是調(diào)度層，分為資源調(diào)度和任務(wù)調(diào)度，資源調(diào)度也由我們完成。

這樣的拆分更好地解決平臺定位不清晰的問題，同時也提高了平臺組件之間的協(xié)作效率。

數(shù)據(jù)服務(wù)總線平臺拆分后以Kafka為中心的部分，該部分包括前端封裝的SDK和OGG，以及隨后引入的Schema Registry組件，致力于提供更完備的流式數(shù)據(jù)供給能力。

同時，我們引入了Apache Hudi，并結(jié)合原有的流式計算組件生態(tài)，構(gòu)建出實時數(shù)據(jù)湖平臺，旨在實現(xiàn)實時數(shù)據(jù)的高效處理和存儲。該平臺致力于實時貼源數(shù)據(jù)存儲，在數(shù)據(jù)的管理、查詢和可視化分析等方面提供優(yōu)化的解決方案。

1、實時數(shù)據(jù)湖

下圖是實時數(shù)據(jù)湖的數(shù)據(jù)處理鏈路，中間是綠色的部分是數(shù)據(jù)湖的存儲層，由Hudi和HDFS實現(xiàn)。

左邊是數(shù)據(jù)導(dǎo)入層，分為兩部分：

• 一是我們存量的貼源數(shù)據(jù)，基于Spark的批量任務(wù)從數(shù)據(jù)湖導(dǎo)入；

• 二是實時貼源數(shù)據(jù)，基于Flink實時流任務(wù)，從數(shù)據(jù)服務(wù)總線貼源層導(dǎo)入。

這兩部分共同匯總在實時數(shù)據(jù)湖，實現(xiàn)了一個完整的實時貼源數(shù)據(jù)存儲，對外提供分鐘級的一個貼源的數(shù)據(jù)的可見性。

右邊是數(shù)據(jù)導(dǎo)出層，分三種場景：

• 一是基于 Spark/Hive的這種批處理的導(dǎo)出的能力，給下游的數(shù)倉和其他業(yè)務(wù)系統(tǒng)提供更穩(wěn)定、低時延的數(shù)據(jù)可用性；

• 二是基于Flink去提供秒級的實時數(shù)據(jù)的消費的能力；

• 三是基于內(nèi)部的多源分析查詢平臺(Presto)，提供實時數(shù)據(jù)的OLAP查詢能力。

實時數(shù)據(jù)湖提供的是分鐘級貼源數(shù)據(jù)存儲，填補了數(shù)據(jù)湖天級延遲場景和數(shù)據(jù)服務(wù)總線秒級延遲場景之間的空白。實時數(shù)據(jù)湖不僅是一種存儲方式，更提供了快速響應(yīng)業(yè)務(wù)需求能力，實現(xiàn)了現(xiàn)代化的數(shù)據(jù)處理，推動了企業(yè)數(shù)字化轉(zhuǎn)型進程。

此外，對于光大內(nèi)部許多業(yè)務(wù)而言，對數(shù)據(jù)的實時性并不需要完全做到秒級，分鐘級的延遲就已能夠滿足當(dāng)前六成的業(yè)務(wù)需求。

下圖是我們內(nèi)部一個核心的業(yè)務(wù)運營系統(tǒng)的數(shù)據(jù)處理鏈路，在引入實時數(shù)據(jù)湖之前和之后的演進路線：

未使用實時數(shù)據(jù)湖之前，數(shù)據(jù)來源于Kafka的ods層，通過Spark Streaming的流處理寫入Hbase，然后經(jīng)過Hive的Tez批處理，最終通過基于Hive on Hbase表提供給業(yè)務(wù)查詢。整條數(shù)據(jù)鏈路的延遲時間非常長，達到了小時級別，這種延時對運營系統(tǒng)的性能影響非常大，明顯減緩了業(yè)務(wù)觸達的時效性。

演進后是基于Hudi的改進的數(shù)據(jù)處理鏈路，可以達到分鐘級的一個延遲。數(shù)據(jù)來源還是Kafka貼源層，會基于Flink的流處理，再去實時導(dǎo)入到Hudi，最終通過 Presto提供給業(yè)務(wù)實時貼源數(shù)據(jù)的交互查詢的能力。

整條鏈路經(jīng)過改造后，數(shù)據(jù)可見性延遲從小時級別下降到了分鐘級別，處理環(huán)節(jié)相對簡化，同時Hudi提供了更穩(wěn)定的數(shù)據(jù)來源。

2、數(shù)據(jù)服務(wù)總線

數(shù)據(jù)服務(wù)總線方面的實踐，主要涉及三個方面：

• 基于Confluent開源的Schema Registry實現(xiàn)了schema解耦；
• 對Kafka的原生API進行了封裝，以便更好地對外提供服務(wù)；
• 提升了服務(wù)的可觀測性和可視化水平。

1）Schema的解耦

下圖是優(yōu)化前的數(shù)據(jù)流轉(zhuǎn)圖。

黑色實線表示數(shù)據(jù)流轉(zhuǎn)的方向，黑色虛線表示元數(shù)據(jù)流轉(zhuǎn)的方向。數(shù)據(jù)來源于Oracle數(shù)據(jù)庫，通過OGG進行導(dǎo)入，以Avro的格式，通過Kafka做數(shù)據(jù)層的解耦，由下游基于Schema去做解析，下游解析數(shù)據(jù)的Schema來源于上游OGG接收到數(shù)據(jù)變更自動生成（即圖中藍色的.avsc文件）。

所以，在上游表結(jié)構(gòu)發(fā)生變更時，游必須同步去更新這個Schema文件才能完成數(shù)據(jù)的解析，這個Schema文件導(dǎo)致上下游系統(tǒng)在數(shù)據(jù)層是未完全解耦的狀態(tài)；當(dāng)上游系統(tǒng)發(fā)生變更時，下游系統(tǒng)需要先基于舊的Schema完成舊數(shù)據(jù)的消費，然后再把新的Schema同步過來，重啟服務(wù)使用新的Schema消費新的數(shù)據(jù)。

總的來說，這個變更流程比較復(fù)雜，且下游對上游存在依賴，并且如果上游在投產(chǎn)的時候，沒有來得及通知下游，當(dāng)Schema變更以后，下游系統(tǒng)會使用舊的Schema去解析新的數(shù)據(jù)，可能就會導(dǎo)致無法預(yù)知的程序宕機，影響下游服務(wù)的穩(wěn)定運行。

為解決這一問題，我們引入了Confluent的Schema Registry來實現(xiàn)Schema的解耦。

上圖展示了在引入了Schema Registry后的數(shù)據(jù)流轉(zhuǎn)圖。

數(shù)據(jù)的流向和之前一樣，主要區(qū)別是：在OGG側(cè)感知到源端的表結(jié)構(gòu)變更時，會主動去Schema Registry注冊新的Schema，同時將獲取到的Schema ID和數(shù)據(jù)一起寫入Kafka。下游消費系統(tǒng)從消息中獲取新的Schema ID，也會去Schema Registry請求獲取對應(yīng)的Schema，并使用這個Schema去解析數(shù)據(jù)。

因此，當(dāng)Schema發(fā)生變化時，下游的自動獲取和更新，免去了非必要的重啟和未通知變更情況下的宕機問題。同時，我們通過Schema Registry將Schema中心化存儲起來，方便后續(xù)進一步的管理。同時，引入Schema Registry以后，上游系統(tǒng)和下游系統(tǒng)都會對它有很強的依賴關(guān)系，這對Schema Registry服務(wù)本身的高可用提出了高要求。

①Schema Registry高可用實踐

上圖右側(cè)是Schema Registry單個實例的內(nèi)部實現(xiàn)原理。最上層是接口層，對外提供Restful形式的訪問接口，注冊請求（寫請求）過來后，會基于內(nèi)部的store層將數(shù)據(jù)以消息的形式寫入Kafka的topic中，也就是說Schema Registry底層的真實存儲實際上是一個Kafka的topic；而Store層會緩存所有的Schema，對于讀請求，會直接從內(nèi)存中獲取。

高可用實踐分為客戶端和服務(wù)端兩部分：

客戶端：高可用的一部分依賴于客戶端本身對Schema的緩存，在這種情況下，服務(wù)端如果暫時不可用，只會影響新增Schema的注冊和解析；同時為了讓服務(wù)端部署多個實例，以此避免單點問題導(dǎo)致的服務(wù)不可用問題，也要求客戶端配置多實例地址。因此，我們在自研的SDK里集成了自動化配置這個功能。

服務(wù)端：除多實例部署以外，每個實例本身也做了獨立的部署，以此避免和其他服務(wù)混部造成影響；第二層是服務(wù)層自身的存儲緩存，在底層存儲出現(xiàn)問題時，可以提供讀服務(wù)。

同時，針對業(yè)務(wù)的需求，我們實現(xiàn)了集群跨域部署和容災(zāi)方案。

在我們內(nèi)部，存在跨域獲取數(shù)據(jù)的需求，自然就會產(chǎn)生跨域獲取Schema的需求。

上圖有兩個域，左邊是DC A，右邊是DC B。在A域里，生產(chǎn)者將相關(guān)數(shù)據(jù)寫入Kafka集群，將Schema注冊到Schema Registry服務(wù)。在消費端，同時在A域和B域都會有對數(shù)據(jù)的消費需求，數(shù)據(jù)的同步一般的方案就是Replicator或MM等工具完成，比如topic1同步到B域?qū)?yīng)A.topic1；在B域，拿到數(shù)據(jù)后，需要獲取Schema ID對應(yīng)的Schema才能去解析出數(shù)據(jù)。因此，我們需要跨域獲取Schema的解決方案。

上圖最下面藍框中的就是一個跨域的Schema Registry集群的實現(xiàn)。橙色是A域的Schema Registry實例，綠色是B域。整個集群會有一個主節(jié)點，通過參數(shù)設(shè)置只能在A域產(chǎn)生。同時，Schema Registry底層的存儲也使用A域的topic（即圖中的_schemas），這是為了保證數(shù)據(jù)沒有跨域?qū)懭耄⒈M量避免因為網(wǎng)絡(luò)問題導(dǎo)致的數(shù)據(jù)重復(fù)或丟失問題。

對于寫請求，都會轉(zhuǎn)發(fā)給主節(jié)點實現(xiàn)，這要求B域中的Schema Registry實例開通和A域中的Schema Registry實例之間的網(wǎng)絡(luò)關(guān)系；對于讀請求，節(jié)點要緩存Schema數(shù)據(jù)，需要和底層的存儲通信，這要求B域中的Schema Registry實例開通和A域中的Kafka Broker節(jié)點之間的網(wǎng)絡(luò)關(guān)系，讀請求是增量獲取，跨域請求對網(wǎng)絡(luò)流量的影響也會比較小。

針對單個域的Kafka集群故障或機房故障，數(shù)據(jù)本身的備份可以通過同步工具完成，由于Schema Registry底層的存儲也是一個topic，它的數(shù)據(jù)也可以通過這種方式備份。在主節(jié)點集群出現(xiàn)問題時，我們可以通過腳本去實現(xiàn)參數(shù)的一鍵切換，從而保證服務(wù)的可用性。

以上是整個Schema Registry的跨域災(zāi)備方案。

我們使用的Schema Registry是Confluent的開源版本，在光大銀行的系統(tǒng)整合過程中，進行了一些改造，主要包括兩個方面：

• 安全&權(quán)限：這是將開源系統(tǒng)集成到企業(yè)內(nèi)部所必需的改進之一，所以Schema Registry的實現(xiàn)充分考慮到這點，為使用者提供了標(biāo)準(zhǔn)的插件接口。我們基于這個接口實現(xiàn)了一個RBAC鑒權(quán)機制，確保服務(wù)的安全性，同時還增加了審計日志，以便進行用戶請求的跟蹤。

• 運維：我們先將服務(wù)與Kerberos整合，再把服務(wù)的內(nèi)部指標(biāo)和行內(nèi)監(jiān)控系統(tǒng)進行了打通。上圖是測試環(huán)境的示例，由于Schema Registry服務(wù)本身對高可用性要求較高，我們在監(jiān)控和報警方面前期做了很多工作，未來也將不斷優(yōu)化。

2）SDK封裝

我們對Kafka客戶端所做的二次開發(fā)做了SDK封裝，提供統(tǒng)一的客戶端接口，解決了以下問題：

• 減少不同客戶端版本帶來的性能差異和穩(wěn)定性問題；
• 方便升級管理、災(zāi)備切換等；
• 更好地規(guī)范客戶端行為，增強對客戶端的數(shù)據(jù)面控制。

在上圖右側(cè)，我們展示了SDK已經(jīng)實現(xiàn)及計劃實現(xiàn)的特性，下面重點介紹下消費定制化這一特性。

如圖所示，在消費定制化的使用場景中，某個生產(chǎn)者寫入的某個數(shù)據(jù)，在下游可能并不是關(guān)心的，以訂單數(shù)據(jù)為例，有些業(yè)務(wù)可能只關(guān)心訂單的流轉(zhuǎn)狀態(tài)，有些業(yè)務(wù)只關(guān)心下單后的物流信息。因此，在大部分情況下，下游消費數(shù)據(jù)只是上游寫入數(shù)據(jù)的一個子集，不同的消費者子集也可能不同。

基于這些場景，我們實現(xiàn)了消費定制化訂閱的特性，將業(yè)務(wù)真正關(guān)心的schema托管到我們自研的Schema Manager服務(wù)中，在SDK消費時，可以提供給下游業(yè)務(wù)真正關(guān)心的數(shù)據(jù)，從而滿足不同消費者的不同數(shù)據(jù)需求。

該特性的對外暴露由SDK封裝實現(xiàn)，底層依賴于Schema Registry和Schema Manager，其中Schema Registry實現(xiàn)了用于中心化托管和對外服務(wù)的schema寫入，而Schema Manager實現(xiàn)了用戶自定義schema的中心化托管和對外服務(wù)，在SDK側(cè)，我們實現(xiàn)了和這兩個服務(wù)之間的交互，及自定義schema的獲取，并完成底層異常情況的處理以及消費數(shù)據(jù)的抽取。

上圖展示了消費定制化特性的處理流程，其核心實現(xiàn)有兩個方面：

一是實現(xiàn)了和Schema Manager之間交互及容錯的能力，并且提供給客戶端調(diào)整能力，實現(xiàn)了一層緩存以提高效率。在未來，我們將考慮在客戶端中實現(xiàn)checkpoint機制，以提高可用性。

二是實現(xiàn)了基于讀schema解析時的兼容性，對于一些字段的偏差，我們可以自動處理，并提供給業(yè)務(wù)一些常用匹配策略的配置。

3）可觀測性&可視化

最后介紹一些我們在觀測性和可視化方面做的事情。

首先，我們對Kafka進行了全面的監(jiān)控。之前，我們發(fā)現(xiàn)基于Kafka的機制的產(chǎn)品端監(jiān)控粒度比較低，且缺失很多關(guān)鍵指標(biāo)，如Kafka服務(wù)端網(wǎng)絡(luò)請求處理線程的繁忙率，業(yè)務(wù)側(cè)消費延遲等。因此，我們結(jié)合Kafka本身的指標(biāo)以及我們內(nèi)部的云原生監(jiān)控體系，實現(xiàn)了更細(xì)粒度的監(jiān)控報警方案，以更好地掌握Kafka的狀態(tài)。

其次，我們還開發(fā)了服務(wù)控制臺。前面我們提到，通過Schema Registry我們可以將schema的中心化存儲。但是，這種數(shù)據(jù)并不是一種結(jié)構(gòu)化的存儲，這就需要我們通過控制臺進行可視化的管理。自定義的schema也通過控制臺去完成創(chuàng)建、更新和刪除的操作。下面的圖展示了一個租戶自定義schema列表的示例。

三、總結(jié)&未來規(guī)劃

1、總結(jié)

通過結(jié)合大數(shù)據(jù)領(lǐng)域新興技術(shù)，我們對整個平臺進行了拆分，解決了多項問題，當(dāng)前準(zhǔn)實時數(shù)據(jù)平臺實現(xiàn)了以下收益：

1）更清晰的平臺邊界；

2）覆蓋分鐘級實時貼源數(shù)據(jù)場景；

3）數(shù)據(jù)服務(wù)總線生態(tài)建設(shè)；

4）提升了運維和運營能力。

2、未來規(guī)劃

1）實時數(shù)據(jù)湖

實時數(shù)據(jù)湖平臺目前處于持續(xù)落地階段，未來我們將持續(xù)遷移適用于分鐘級實時數(shù)據(jù)庫的全部貼源場景，同時探索基于行業(yè)經(jīng)驗的湖倉一體和流批一體場景。

2）數(shù)據(jù)服務(wù)總線

• 持續(xù)圍繞消息中間件生態(tài)進行拓展，基于SDK實現(xiàn)客戶端的災(zāi)備切換功能，并結(jié)合服務(wù)端的災(zāi)備，共同實現(xiàn)平臺級別的災(zāi)備方案；

• 由于總線的上下游接口的一個重要計算框架是Flink Connector，因此我們將基于它進行二次開發(fā)，實現(xiàn)部分核心功能，為接入的業(yè)務(wù)提供統(tǒng)一的體驗；

• 除增強客戶端接入能力外，我們將積極開發(fā)管理控制臺，將一些重要的Kafka運維操作和服務(wù)部分運營能力整合到我們的控制臺上，以提高整體的運維和運營效率；

• 著手計劃長期的信創(chuàng)集群建設(shè)。

講師介紹

張旭，光大銀行準(zhǔn)實時數(shù)據(jù)平臺技術(shù)負(fù)責(zé)人，專注于分布式系統(tǒng)內(nèi)核研發(fā)。在OLAP & 消息中間件領(lǐng)域有較豐富經(jīng)驗。開源愛好者，Apache RocketMQ committer，Prometheus contributor。