Apache SeaTunnel實現 非CDC數據抽取實踐記錄
目錄
- 01 SeaTunnel簡介
- 1. Apache SeaTunnel整體介紹
- 2. Apache SeaTunnel技術特性
- 3. Apache SeaTunnel工作流程
- 4. Apache SeaTunnel環境依賴
- 5. Apache SeaTunnel用戶使用情況
- 02 SeaTunnel應用場景
- 1. 交管行業數據簡介
- 2. 交管行業數據特點
- 03 相關業務痛點
- 1. 數據抽取限制較多
- 04 選擇SeaTunnel的原因
- 1. SeaTunnel的優勢
- 2. SeaTunnel的安裝部署
- 3. SeaTunnel配置文件
- 4. SeaTunnel插件支持
- 05 具體實現方案
- 1. 數據增量更新具體實現
- 2. 具體方法
- 06 具體實現流程
- 1. 確定運算資源
- 2. 確定數據來源
- 3. 數據轉換
- 4. 數據輸出
- 5. 腳本和調度執行
導讀： 隨著全球數據量的不斷增長，越來越多的業務需要支撐高并發、高可用、可擴展、以及海量的數據存儲，在這種情況下，適應各種場景的數據存儲技術也不斷的產生和發展。與此同時，各種數據庫之間的同步與轉化的需求也不斷增多，數據集成成為大數據領域的熱門方向，于是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數據抽取內容請搜索以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持！
網公網安備