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

導語

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

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

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

本篇文章將從以下幾個方面對圍繞著訊息佇列如何快速搭建資料鏈路的落地實踐進行分享。

• 資料鏈路構建的挑戰
• 技術架構體系的建設
• 客戶實踐和落地案例

視訊：http://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

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