輕量級SaaS化應用數據鏈路構建方案的技術探索及落地實踐

導語

2022騰訊全球數字生態大會已圓滿落幕，大會以“數實創新、產業共進”為主題，聚焦數實融合，探索以全真互聯的數字技術助力實體經濟高質量發展。大會設有29個產品技術主題專場、18個行業主題專場和6個生態主題專場，各業務負責人與客户、合作伙伴共同總結經驗、凝結共識，推動數實融合新發展。

本次大會設立了微服務與中間件專場，本專場從產品研發、運維等最佳落地實踐出發，詳細闡述雲原生時代，企業在開發微服務和構建雲原生中間件過程中應該怎樣少走彎路，聚焦業務需求，助力企業發展創新。

隨着大數據時代的到來，企業在生產和經營活動中產生的各類數據正以前所未有的速度增長，通過對實時及歷史數據的融合分析，及時挖掘業務洞察和輔助決策，已成為企業的普遍行動。在雲原生的浪潮下，企業需要聚焦業務，迫切需要簡單易行，零代碼地配置搭建起自己的可以達到將本增效效果的數據鏈路系統。

本篇文章將從以下幾個方面對圍繞着消息隊列如何快速搭建數據鏈路的落地實踐進行分享。

• 數據鏈路構建的挑戰
• 技術架構體系的建設
• 客户實踐和落地案例

視頻：https://www.zhihu.com/zvideo/1586024427908657152

數據鏈路構建的挑戰與開源生態

數據鏈路構建的挑戰

如下圖所示，這是一張經典的數據鏈路的架構圖，從左到右依次可以分為數據源、數據接入層、數據緩衝層、數據處理層和右邊的數據目標。在這樣一個典型的數據鏈路裏，技術組件非常多，導致整個圖非常複雜，這會增加運維成本。

圖1

接下來看另一張圖，如果把中間部分全部屏蔽掉，這個數據鏈路變為一款SaaS化的數據接入組件，那它就會非常輕量。

圖2

所以在開源生態中，多樣的數據源和數據目標，眾多開源組件的學習成本，數據鏈路的搭建和運維是整個數據鏈路系統主要面對的問題。

企業需要聚焦業務，因此數據鏈路系統需要：SAAS 化、低代碼化、簡單易用、穩定可靠、高性能、按量付費。以達到整體上的的降本增效。

我們再回到圖1，可以看到，它的緩衝層在業界主要都是 Kafka，然後圍繞 Kafka 生態，具有豐富的上下游，那複雜度、學習成本、維護成本這些問題要如何解決呢？繼續往下看。

數據鏈路功能矩陣

圖3

圖4

如上圖3所示，數據鏈路由數據源、數據庫兩部分組成。

• 數據源

文本日誌、CVM、容器、安全等

• 數據庫

數據庫數據、主動上報數據等

這些數據需要處理上報然後發到下游，在業界更多的是 Filebeat、Flink、Logstash 等社區組件。想要達到圖3這張圖的效果，就需要圖4這一堆組件，這就涉及到上面提到過的問題。所以就衍生出了一個 SaaS化 的數據鏈路的方案。

Saas化的數據鏈路方案

CKafka 連接器是騰訊雲上 SaaS 化的數據接入和處理解決方案，一站式提供對數據的接入、處理和分發功能。

提供基於 HTTP/TCP 協議的 SDK 協助客户完成數據上報；基於 CDC 機制訂閲、存儲多款數據庫變更信息；簡單可配置的數據清洗 (ETL) 能力；豐富的數據分發渠道；打通了混合雲/跨雲的豐富的數據源(MQ, 數據庫，事件等)數據接入。

協助客户低成本搭建數據流轉鏈路，構建數據源和數據處理系統間的橋樑。

應用場景

數據鏈路構建

在正常業務當中，用户需要將多種數據源的數據經過客户單採集，實時處理緩衝，傳到下游的搜索，這時就可以通過這套鏈路直接把數據一條鏈路完全打通，直接把數據源打到下游的存儲，這就非常便利了。

在實際業務過程中，用户經常需要將多個數據源的數據彙總到消息隊列中，比如業務客户端數據、業務 DB 數據、業務的運行日誌數據彙總到消息隊列中進行分析處理。正常情況下，需要先將這些數據進行清洗格式化後，再做統一的轉儲、分析或處理。

CKafka 連接器支持將不同環境（騰訊公有云、用户自建 IDC、跨雲、混合雲等）的不同數據源（數據庫、中間件、日誌、應用系統等）的數據集成到公有云的消息隊列服務中，以便進行數據的處理和分發。提供了數據聚合、存儲、處理、轉儲的能力，即 數據集成 的能力，將不同的數據源連接到下游的數據目標中。

數據接入分發

另外三個場景分別是數據上報、數據庫訂閲和數據的清理和分發。

客户、業務端或者運維端可能有很多數據需要上報，需要自己搭建一個上報的 Server，但如果使用 Sass 化數據接入產品，它就可以很輕量化的完成數據上報。

數據庫訂閲和數據的清理分發等功能是一樣的原理，需要做的就是把數據從各種數據源很 Saas 化的接進來，然後簡單輕量的清洗出去。

數據上報

數據庫數據訂閲

數據庫清洗和分發

接下來分享如何從技術上實現輕量級 Saas 化數據鏈路搭建，會遇到什麼問題，業界有什麼通用的做法。

技術架構體系的建設

系統架構

從上圖可知，數據鏈路整體分為4個層面：接入層、緩衝層、數據處理層和數據分發層。

從左到右，在數據面可以看到數據源、客户端、APP，會通過訂閲、上報等接口把數據上報到接入層裏面；然後接入層會把數據緩衝到緩衝層，緩衝層一般是 MQ，比如 Kafka、Pulsar 等消息隊列產品；接着在數據處理層，會處理消費緩存層的數據，把數據經過簡單的 ETL 重組、重裝、裁剪等等分發到下游的各種數據目標。

控制面會提供一些 API 控制調度監控、擴縮容、管理、運維、遷移等等這些管控面的能力，這時會提供 API 給大家調用，這就是控制面和數據面的大體架構。如果自己去搭建這麼一套數據鏈路的產品也是需要這麼多的工作的。

界面化的ETL引擎

在數據處理層一般是通過編碼，比如 Logstash 的語法，或者 Python 和 Flink 的 代碼，或者 ETL 函數的語法等處理方式。但對用户來説，他可能不需要這麼多的功能，也不想投入這麼多的學習成本，用户就可以使用 CKafka 連接器，在通過 CKafka 連接器組件處理數據流入流出任務時，通常需要對數據進行簡單的清洗操作，比如格式化原始數據，格式化解析特定字段，數據格式轉換等。開發者往往需要自己搭建一套數據清洗的服務（ETL）。

如下圖所示，從數據進來以後會經過多層的轉換存在緩衝層然後再消費到下游，這是數據處理一個體系化的鏈路圖。我們可以提供一個完全界面化的處理引擎來支持 JSON 的簡易操作、JSON 的格式化解析、數據的裁剪替換等通用的 ETL 的行為。這個界面化的 ETL 引擎底層是基於 Transform 接口、Interface 等機制來實現的。

多引擎架構 — Kafka Connector

怎麼樣來解決整個數據流的連接和接入呢？從研發層面來講，從進程或者線程的層面，從數據研發數據寫到緩衝層再打到下游，整個不同任務的維度是需要調度的，當前的業界沒有一種通用的引擎去解決所有問題，所以CKafka連接器方案底層實現的是多引擎的一套架構，那相當於有多套引擎同時並行的提供服務、調度、分佈式的遷移和啟動、停止、變更等行為。

首先來看引擎1：Kafka Connector，它是 Kafka 社區提供的一款計算調度的產品。這款產品主要解決的問題就是它提供了一個分佈式的任務調度的框架，會同時開放出很多 Interface 的接口，會從數據源提供很多插件，比如 JDBC、Syslog、MQTT、MongoDB 等，這些插件會把數據從源端不斷的拉到 Kafka 裏面來，然後在下游再對接 HBRSE、S3、Elastic、Cassandra 等一些 Sink 的服務。Kafka Connector 分為兩個層面，一個是調度層面，調度層面就整個框架，會提供分佈式的部署，分佈式的容災。另一個是跨可用區的部署、跨可用區容災等，提供各種不同的插件，Source、Sink 等，形成一套數據流。Kafka 引擎一個打通一個引擎，如果開發者自建，可以自己去搭建的，這時候更多要關注穩定性、擴縮容，以及內核問題的及時修復等。

多引擎架構 – Flink Connector

接着看引擎2：Flink Connector，Flink 大家都用的非常熟，其實 Flink Connector 也非常強大，它會提供很多計算框架，其實跟 Kafka Connector 類似，它也提供了很多分佈式計算層的服務，也提供了很多 Connector 和 Extract 函數、UTF 等操作，它的 Connector 會對接各種數據源，也會對接各種 ES，它在數據源會定個數據庫的 CDC，更多的是服務類的，比如數據源是 Kafka、DFS、Cassandra 等，這時它會通過內部的分佈式調度和處理把數據源打到下游的 ES，這裏是一個 Load 的過程，裏面有很多算子等的概念。如果用户想要自己去搭建的話是比較複雜的。多引擎架構是為了解決兩款技術體系 Flink 和 Connector 具有的不足之處，將兩款技術體系融合在一起，進行不同的調度和遷移。從數據源來看，它執行的就是為不同的數據源拿數據，沒有緩衝層，直接到下游的 ES，區別在於，如果你需要存或者不需要存，任務的數據量、並行度這些都是我們控制的。

多引擎架構 – MQTT 協議接入

接下來看引擎3：MQTT 協議接入，MQTT 協議是指數據接入平台會提供整個 MQTT 的軟件層，各種 Connector 端會連接到 MQTT 的整個 Proxy 層，它會提供 MQTT 3、MQTT 5的一流量控制、語音版消息服務等一個體系，也會支持 QS 1、QS 2等，也支持通過 MQTT 把消息打到下游的 Bridge 這些數據橋階層，轉發到 Kafka 或者其他 MQ。

多引擎架構 – HTTP 協議接入

最後看多引擎架構4：支持 HTTP 協議接入，數據能夠通過 HTTP 協議從數據源導進來。

如下圖所示，看一下 HTTP 協議的架構，第一層是網關，它有各種 Report，通過接收數據在內部維護 API 連接池，把數據分發到 Database、Monitor、Report 等，最終是把數據存到各種 MQ 裏面。

從總體來看，CKafka 連接器會提供多種數據流的引擎，Kafka Connector、Flink Connector等，這些對用户都完全屏蔽了，用户用到的只是一個 Saas 化的輕量級組件方案，還可以提供MQTT 協議和 HTTP 協議，用户可以直接接入，接入後用户就可以非常輕量的解決問題。

客户實踐

場景1 – 數據入湖

數據入湖的概念現在非常火，就是把屏蔽底層的各種 HDFS、COS 等持久存儲的數據或者異構的數據進行統一查詢分析。

客户業務數大部分都存在 MongoDB 裏面。有一部分客户行為數據，需要上報後進行分析。客户希望將這些數據統一到數據湖(iceberg)進行分析。

自建鏈路遇到的問題，鏈路太長，涉及的組件非常多。大多數組件是分佈式部署，擴縮容複雜，維護鏈路的穩定性，透明監控需要花費大量精力。使用連接器組件後，只需要簡單配置，SAAS 化，鏈路的穩定性，擴縮容依託平台處理。

看下面的架構圖，有 Mongo 的數據源，在接入層通過 Mongo 的 Connector 去 Mongo 裏拿數據，訂閲 MongoStream 的數據，需要先把數據存到 Kafka 的 Topic 裏，因為原始訂閲數據是有 Schema 規範的，這時在 Iceberg 裏，是一個存儲一個解析的層，所以需要簡單的處理，通過Kafka Connector 的 Sink 把數據存到 DLC 裏面去。

場景2 – 數據上報和多協議接入

數據接入

某教育客户需要將直播課學生上下課、簽到、瀏覽等一些行為信息上傳到後台進行分析、處理和檢索。數據在後台主要有兩種業務邏輯：

     1.  自定義代碼拿到上報數據，進行對應業務邏輯處理

     2. 原始數據進入 Elasticsearch 進行檢索分析

因開發人力有限，希望有一種方便的數據接入服務，簡單快速地完成數據的上報、存儲。

這個客户的數據源是各種客户端，通過數據上報接入到 HTTP 接入層中，然後通過連接器存儲，數據分發到ES，然後客户自己的代碼去消費。

多協議接入

某保險客户的中台團隊遷移上雲，因下游團隊眾多，使用多款MQ產品（Kafka，RocketMQ，RabbitMQ）。各個MQ都是 TCP 協議接入，有各自的 SDK。SDK 學習、使用、以及後續切換成本較高。

基於中台考慮，希望上雲後能夠通過簡單的HTTP協議進行接入，屏蔽底層的具體引擎細節。

有三個要求： 1.  簡化客户端的使用，最好是HTTP協議。 2. 底層MQ引擎切換對業務無感知。 3. 最好有現成的支持HTTP協議的SDK.

使用連接器組件就解決了非常實際的上報、訂閲和分發的場景。

場景3 – 數據庫訂閲

某迅銷平台內部多有多套系統並行運行，某套系統存儲引擎為 PGSQL。 需要將 PGSQL 的變更數據存量導入到 Elasticsearch 裏面進行查詢。有如下幾個需求：1.  數據寫入 ES 的時候需要根據時間分索引  2. 因為某個數據量大，希望在某個時間區間內只保留某個唯一 ID標識的最新數據（update）。3. 需要根據不同的表將數據分發到不同的索引裏面。

自建的架構： PGSQL + DebeziumPGSQL+KafkaConnector+Kafka+Logstash+ Elasticsearch

CKafka連接器架構：      PGSQL + 連接器 + Elasticsearch

從上面的架構可以看的出來，使用連接器方案可以將數據鏈路中的很多細節直接屏蔽，直接打到下游，非常輕量化。