Apache SeaTunnel實現 非CDC資料抽取實踐記錄

導讀： 隨著全球資料量的不斷增長，越來越多的業務需要支撐高並行、高可用、可延伸、以及海量的資料儲存，在這種情況下，適應各種場景的資料儲存技術也不斷的產生和發展。與此同時，各種資料庫之間的同步與轉化的需求也不斷增多，資料整合成為巨量資料領域的熱門方向，於是SeaTunnel應運而生。SeaTunnel是一個分散式、高效能、易擴充套件、易使用、用於海量資料（支援實時流式和離線批次處理）同步和轉化的資料整合平臺，架構於Apache Spark和Apache Flink之上。本文主要介紹SeaTunnel 1.X在交管行業中的應用，以及其中如何實現從Oracle資料庫把資料增量匯入數倉這樣一個具體的場景。

今天的介紹會圍繞下面六點展開：

• SeaTunnel簡介
• SeaTunnel應用場景
• 相關業務痛點
• 選擇SeaTunnel的原因
• 具體實現方案
• 具體實現流程

01 SeaTunnel簡介

下面對SeaTunnel從產品功能，技術特性、工作流程、環境依賴、使用者使用等方面做一個總體的介紹。

1. Apache SeaTunnel整體介紹

網際網路行業資料量非常大，對效能還有其他各方面的技術要求都非常高，在筆者所在的交管行業中，情況就不太一樣，各方面的要求也沒有網際網路行業那麼高，在具體的資料整合應用中，主要是使用SeaTunnel1.X版本。

上圖所示內容參照了Apache SeaTunnel官網中的介紹。

Apache Spark對於分散式資料處理來說是一個偉大的進步，但是直接使用Spark框架還是有一定門檻的，SeaTunnel這個產品把業界使用Spark的優質經驗固化到了其中，明顯降低了學習成本，加快分散式資料處理能力在生產環境中落地。在SeaTunnel2.X版本中，除了Spark，也增加了對Flink的支援。

除此之外，SeaTunnel還可以較好的解決實際業務場景中碰到的下列問題：

• 資料丟失與重複
• 資料整合中任務堆積與延遲
• 資料同步較低的吞吐量
• Spark/Flink應用到生產環境週期較長、複雜度較高
• 缺少應用執行狀態的監控

2. Apache SeaTunnel技術特性

SeaTunnel具備如上圖所示的技術特性：

• 簡單易用，開發設定簡單、靈活，無需編碼開發，支援通過SQL進行資料處理和聚合，使用成本低
• 分散式，高效能，經歷大規模生產環境使用和海量資料檢驗，成熟穩定
• 模組化和外掛化，內建豐富外掛，並且可以開發客製化個性化外掛，支援熱插拔，具備高擴充套件性
• 使用Spark/Flink作為底層資料同步引擎使其具備分散式執行能力

3. Apache SeaTunnel工作流程

SeaTunnel的架構和整個工作流程如下圖所示，Input/Source [資料來源輸入] -> Filter/Transform [資料處理] -> Output/Sink [結果輸出]，資料處理流水線由多個過濾器構成，以滿足多種資料處理需求。如果使用者習慣了SQL，也可以直接使用SQL構建資料處理管道，更加簡單高效。目前，SeaTunnel支援的過濾器列表也在擴充套件中。

在外掛方面，SeaTunnel已支援多種Input/Sink外掛，同時也支援多種Filter/Transform處理外掛，整體上基於系統非常易於擴充套件，使用者還可以自行開發資料處理外掛，具體如下：

• Input/Source 外掛

Fake, File, Hive/Hdfs, Kafka, Jdbc, ClickHouse, TiDB, HBase, Kudu, S3, Socket, 自行開發的Input外掛

• Filter/Transform 外掛

Add, Checksum, Convert, Date, Drop, Grok, Json, Kv, Lowercase, Remove, Rename, Repartition, Replace, Sample, Split, Sql, Table, Truncate, Uppercase, Uuid, 自行開發的Filter/Transform外掛

• Output/Sink 外掛

Elasticsearch, File, Hdfs, Jdbc, Kafka, Mysql, ClickHouse, Stdout, 自行開發的Output 外掛

4. Apache SeaTunnel環境依賴

SeaTunnel1.X支援Spark計算引擎，SeaTunnel2.X目前支援Spark/Flink兩種計算引擎，在筆者的實際專案中使用的是SeaTunnel1.X版本。

5. Apache SeaTunnel使用者使用情況

目前有很多公司都在使用SeaTunnel，其中不乏大型公司，例如：中國移動、騰訊雲、今日頭條、還有筆者所在的中電科。

02 SeaTunnel應用場景

SeaTunnel特別適合以下場景使用：

• 海量資料整合和ETL
• 海量資料聚合
• 多源資料處理

下面主要介紹SeaTunnel在交管行業中的應用。

1. 交管行業資料簡介

在交管行業中，資料主要包括駕駛人、車輛相關的資料，平時在道路上發生的一些交通警情資料，交通違法資料，機動車登記資訊，執勤執法的資料，交通事故以及其他一些網際網路資料，這些資料的量不是很大，另外還有卡口過車、車輛GPS資料，這兩種資料的資料量都比較大，例如一些省會城市，每秒鐘至少有幾千條過車資料，這些資料都是屬於交管行業內的資料。

2. 交管行業資料特點

交管行業資料，跟網際網路行業的資料還是有很大區別的，首先這些資料的體量大小不一，並且分佈在內部的公安網以及智慧專網，這兩個網之間是物理隔離的，我們需要把這些資料在兩個網路之間轉移，在這個過程中，還要做一些資料處理。其次，在資料處理實時性方面的要求，並不是非常高，資料的更新頻率也不是很高。然後，在資料安全方面，要求比較高，資料是不能丟的，同時對保密性要求也比較高，所以具體的資料也不能展示出來。

03 相關業務痛點

1. 資料抽取限制較多

在做業務的過程中，會有一些業務痛點，首先因為交管行業是政府行業，基本各個子平臺的資料都是儲存在Oracle資料庫中的，我們需要把資料從Oracle資料庫中抽取到我們的數倉裡面，出於安全性的考慮，無法得到使用者級別的許可權，我們只能通過一些檢視級別的使用者許可權去處理資料，對於資料來源表結構的變更也無法及時知曉。其次，對談數是受到限制的，多執行緒抽取資料的話，如果對談數達到上限，連線就會受到影響，而且這個分配的使用者也同時會用於其他用途。最後，我們在處理一些增量資料的時候，一般情況下需要一個增量列，用於保持一個增量更新，很多時候，是沒辦法確定哪些列可以作為增量列的。以上就是在做業務的過程中，經常會遇到的一些問題，下圖也把這些問題列舉了出來。

04 選擇SeaTunnel的原因

最初的時候，做資料處理、資料抽取的時候，並沒有使用SeaTunnel，而是使用Apache NiFi，這個工具功能比較強大而且全面，但是NiFi中用於資料處理的處理器比較多，而且資料處理鏈路中要做很多轉換，所以需要對NiFi裡面的各種元件要非常熟悉，對使用者的要求也比較高。

1. SeaTunnel的優勢

我們一開始也用Spark程式做資料處理，對巨量資料相關人員的要求比較高，我們這邊巨量資料人員比較少，有時處理一些新的需求的時候，會比較繁忙。如果不需要通過編碼，而是直接使用工具，進行簡單的設定就能實現的話，會帶來較大的便利和效率的提高。

前面在SeaTunnel的介紹中，已經講到SeaTunnel是比較易於使用的，安裝部署方便，開箱即用，執行效率很高，因為它是分散式的，可以應用整個叢集資源來做資料處理工作。

SeaTunnel無需程式設計，只要做簡單的設定，並且它的Source和Sink都比較豐富，並且可以自己根據介面開發需要的外掛，對資料來源的許可權要求也不高。

更加重要的是，SeaTunnel是首個進入Apache孵化的國人開源資料整合平臺。

2. SeaTunnel的安裝部署

如上圖所示是SeaTunnel官方部署檔案，只需要簡單幾步，就可以把SeaTunnel安裝到我們的環境之中，然後就可以使用了。

3. SeaTunnel組態檔

下圖所示是一個組態檔的範例，這個組態檔是SeaTunnel1.X版本的一個設定，一個完整的SeaTunnel設定包含spark, input, filter, output四部分，其中spark是spark相關的設定，例如，啟動多少個executor，每個 executor使用多少核數的CPU，多少記憶體等，input可設定任意的input外掛及其引數，具體引數隨不同的input外掛而變化，filter可設定任意的filter外掛及其引數，具體引數隨不同的filter外掛而變化，filter中的多個外掛按設定順序形成了資料處理的pipeline, 上一個filter的輸出是下一個filter的輸入，通過input外掛把資料取出，成為了spark裡面的一個資料集，然後filter外掛會對這個資料集做一些轉換操作，output可設定任意的output外掛及其引數，具體引數隨不同的output外掛而變化，filter處理完的資料，會傳送給output中設定的每個外掛

4. SeaTunnel外掛支援

如下圖所示，SeaTunnel支援的外掛非常豐富，日常所能用到的基本都有。

這裡面著重介紹一下filter外掛中的sql外掛，這個外掛非常靈活，在用sql外掛做轉換操作時，只要是sparksql裡面支援的函數等內容，都可以在這裡使用，然後再output到目標資料儲存，例如HDFS、Kafka、ES、Clickhouse等。

05 具體實現方案

接下來講一下具體的實現方案，在我們具體的業務中，如何把這些行業資料從智慧專網直接抽取到公安網中，這裡會涉及到資料的增量更新。

1. 資料增量更新具體實現

當需要實現一個增量更新的時候，首先就是增量列的選擇，之前提到原先是用NiFi來做增量更新，但是對增量列的支援不是特別好，尤其是對日期型別的支援不是很好。但是SeaTunnel對增量列的支援不受列的型別限制，可以比較靈活的進行選擇。

2. 具體方法

實際業務當中，選取了記錄的更新時間列作為增量列，每次資料抽取過來，會記錄增量列的最大值，下次資料抽取時，可以從這個位置繼續抽取資料，這個也是受以前寫spark程式的啟發，把checkpoint儲存在HDFS裡面。當然，增量列的選擇，在實際應用中，除了更新時間，增量ID以外，還有其他業務欄位可以做為增量列，增量列的選擇一定是根據真正的業務需求，實時的程度和粒度來決定的。

06 具體實現流程

做資料增量更新，最重要的是實現的思路，接下來詳細描述一下具體實現過程。

1. 確定運算資源

首先，如下圖所示，先要確定計算資源，這裡使用了spark，並且針對spark做了相關的設定。

2. 確定資料來源

選擇一個增量列，對增量列每次產生的最大值（checkpoint），儲存在HDFS一個具體的目錄下。這裡input外掛選擇HDFS，每次產生的那個增量資料，指向HDFS的一個具體路徑下面，input外掛有個通用引數叫做result_table_name，當指定result_table_name時，處理後的資料，會被註冊為一個可供其他外掛直接存取的資料集，或者被稱為臨時表。當增量列的最大值儲存到HDFS之後，需要取出時，會儲存在result_table_name指定的表中。接下來因為是從Oracle資料庫中取資料，所以設定相應的Jdbc。當資料量比較大的時候，還可以指定分割區列，這樣的話，資料處理的效率會提高，詳細設定，如下圖所示。

3. 資料轉換

下圖所示是必要的資料轉換，在實際業務中，需要做一個過濾操作，取出大於最大更新時間的資料，convert外掛裡面做的是中間的一些資料型別轉換操作，最後使用了一個sql外掛，用於記錄本次取到的資料的一個最大值，用於下次取數的比較。

4. 資料輸出

下圖所示的是資料處理後的輸出，也就是output外掛對應的設定，具體是把資料抽取到Clickhouse裡面。然後資料集裡面，那個更新列的最大值，通過追加模式，寫回到HDFS中，供下次使用。

5. 指令碼和排程執行

整個過程是通過下圖所示的shell指令碼來做的，通過nohup後臺執行的方式，利用Crontab進行排程執行，因為在我們實際的業務中，對定時排程的要求不是很高，所以可以採用Crontab或者開源的Dolphin Scheduler都是可以滿足的。

下面的截圖，是實際執行過程中，產生在HDFS上的增量檔案，Crontab排程指令碼，以及執行過程中產生的一些Yarn任務列表。

在上述整體資料處理過程中，由於實際情況的限制，尤其我們的資料來源是高度受限的Oracle資料庫。但是對於很多傳統公司，如果老系統是以Oracle為主，並且掌控力度比較大的話，現在想做資料架構升級，需要遷移Oracle中的資料，那麼可以採用CDC讀取紀錄檔或者觸發器的方式，把資料變化寫入到訊息佇列裡面，通過SeaTunnel就可以很容易的把資料實時寫入到其他異構的資料庫。

到此這篇關於Apache SeaTunnel實現非CDC資料抽取實踐的文章就介紹到這了,更多相關Apache SeaTunnel資料抽取內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！