目錄

• 01 SeaTunnel簡(jiǎn)介
• 1. Apache SeaTunnel整體介紹
• 2. Apache SeaTunnel技術(shù)特性
• 3. Apache SeaTunnel工作流程
• 4. Apache SeaTunnel環(huán)境依賴(lài)
• 5. Apache SeaTunnel用戶(hù)使用情況
• 02 SeaTunnel應(yīng)用場(chǎng)景
• 1. 交管行業(yè)數(shù)據(jù)簡(jiǎn)介
• 2. 交管行業(yè)數(shù)據(jù)特點(diǎn)
• 03 相關(guān)業(yè)務(wù)痛點(diǎn)
• 1. 數(shù)據(jù)抽取限制較多
• 04 選擇SeaTunnel的原因
• 1. SeaTunnel的優(yōu)勢(shì)
• 2. SeaTunnel的安裝部署
• 3. SeaTunnel配置文件
• 4. SeaTunnel插件支持
• 05 具體實(shí)現(xiàn)方案
• 1. 數(shù)據(jù)增量更新具體實(shí)現(xiàn)
• 2. 具體方法
• 06 具體實(shí)現(xiàn)流程
• 1. 確定運(yùn)算資源
• 2. 確定數(shù)據(jù)來(lái)源
• 3. 數(shù)據(jù)轉(zhuǎn)換
• 4. 數(shù)據(jù)輸出
• 5. 腳本和調(diào)度執(zhí)行

導(dǎo)讀： 隨著全球數(shù)據(jù)量的不斷增長(zhǎng)，越來(lái)越多的業(yè)務(wù)需要支撐高并發(fā)、高可用、可擴(kuò)展、以及海量的數(shù)據(jù)存儲(chǔ)，在這種情況下，適應(yīng)各種場(chǎng)景的數(shù)據(jù)存儲(chǔ)技術(shù)也不斷的產(chǎn)生和發(fā)展。與此同時(shí)，各種數(shù)據(jù)庫(kù)之間的同步與轉(zhuǎn)化的需求也不斷增多，數(shù)據(jù)集成成為大數(shù)據(jù)領(lǐng)域的熱門(mén)方向，于是SeaTunnel應(yīng)運(yùn)而生。SeaTunnel是一個(gè)分布式、高性能、易擴(kuò)展、易使用、用于海量數(shù)據(jù)（支持實(shí)時(shí)流式和離線(xiàn)批處理）同步和轉(zhuǎn)化的數(shù)據(jù)集成平臺(tái)，架構(gòu)于Apache Spark和Apache Flink之上。本文主要介紹SeaTunnel 1.X在交管行業(yè)中的應(yīng)用，以及其中如何實(shí)現(xiàn)從Oracle數(shù)據(jù)庫(kù)把數(shù)據(jù)增量導(dǎo)入數(shù)倉(cāng)這樣一個(gè)具體的場(chǎng)景。

今天的介紹會(huì)圍繞下面六點(diǎn)展開(kāi)：

• SeaTunnel簡(jiǎn)介
• SeaTunnel應(yīng)用場(chǎng)景
• 相關(guān)業(yè)務(wù)痛點(diǎn)
• 選擇SeaTunnel的原因
• 具體實(shí)現(xiàn)方案
• 具體實(shí)現(xiàn)流程

01 SeaTunnel簡(jiǎn)介

下面對(duì)SeaTunnel從產(chǎn)品功能，技術(shù)特性、工作流程、環(huán)境依賴(lài)、用戶(hù)使用等方面做一個(gè)總體的介紹。

1. Apache SeaTunnel整體介紹

互聯(lián)網(wǎng)行業(yè)數(shù)據(jù)量非常大，對(duì)性能還有其他各方面的技術(shù)要求都非常高，在筆者所在的交管行業(yè)中，情況就不太一樣，各方面的要求也沒(méi)有互聯(lián)網(wǎng)行業(yè)那么高，在具體的數(shù)據(jù)集成應(yīng)用中，主要是使用SeaTunnel1.X版本。

上圖所示內(nèi)容引用了Apache SeaTunnel官網(wǎng)中的介紹。

Apache Spark對(duì)于分布式數(shù)據(jù)處理來(lái)說(shuō)是一個(gè)偉大的進(jìn)步，但是直接使用Spark框架還是有一定門(mén)檻的，SeaTunnel這個(gè)產(chǎn)品把業(yè)界使用Spark的優(yōu)質(zhì)經(jīng)驗(yàn)固化到了其中，明顯降低了學(xué)習(xí)成本，加快分布式數(shù)據(jù)處理能力在生產(chǎn)環(huán)境中落地。在SeaTunnel2.X版本中，除了Spark，也增加了對(duì)Flink的支持。

除此之外，SeaTunnel還可以較好的解決實(shí)際業(yè)務(wù)場(chǎng)景中碰到的下列問(wèn)題：

• 數(shù)據(jù)丟失與重復(fù)
• 數(shù)據(jù)集成中任務(wù)堆積與延遲
• 數(shù)據(jù)同步較低的吞吐量
• Spark/Flink應(yīng)用到生產(chǎn)環(huán)境周期較長(zhǎng)、復(fù)雜度較高
• 缺少應(yīng)用運(yùn)行狀態(tài)的監(jiān)控

2. Apache SeaTunnel技術(shù)特性

SeaTunnel具備如上圖所示的技術(shù)特性：

• 簡(jiǎn)單易用，開(kāi)發(fā)配置簡(jiǎn)單、靈活，無(wú)需編碼開(kāi)發(fā)，支持通過(guò)SQL進(jìn)行數(shù)據(jù)處理和聚合，使用成本低
• 分布式，高性能，經(jīng)歷大規(guī)模生產(chǎn)環(huán)境使用和海量數(shù)據(jù)檢驗(yàn)，成熟穩(wěn)定
• 模塊化和插件化，內(nèi)置豐富插件，并且可以開(kāi)發(fā)定制個(gè)性化插件，支持熱插拔，具備高擴(kuò)展性
• 使用Spark/Flink作為底層數(shù)據(jù)同步引擎使其具備分布式執(zhí)行能力

3. Apache SeaTunnel工作流程

SeaTunnel的架構(gòu)和整個(gè)工作流程如下圖所示，Input/Source [數(shù)據(jù)源輸入] -> Filter/Transform [數(shù)據(jù)處理] -> Output/Sink [結(jié)果輸出]，數(shù)據(jù)處理流水線(xiàn)由多個(gè)過(guò)濾器構(gòu)成，以滿(mǎn)足多種數(shù)據(jù)處理需求。如果用戶(hù)習(xí)慣了SQL，也可以直接使用SQL構(gòu)建數(shù)據(jù)處理管道，更加簡(jiǎn)單高效。目前，SeaTunnel支持的過(guò)濾器列表也在擴(kuò)展中。

在插件方面，SeaTunnel已支持多種Input/Sink插件，同時(shí)也支持多種Filter/Transform處理插件，整體上基于系統(tǒng)非常易于擴(kuò)展，用戶(hù)還可以自行開(kāi)發(fā)數(shù)據(jù)處理插件，具體如下：

• Input/Source 插件

Fake, File, Hive/Hdfs, Kafka, Jdbc, ClickHouse, TiDB, HBase, Kudu, S3, Socket, 自行開(kāi)發(fā)的Input插件

• Filter/Transform 插件

Add, Checksum, Convert, Date, Drop, Grok, Json, Kv, Lowercase, Remove, Rename, Repartition, Replace, Sample, Split, Sql, Table, Truncate, Uppercase, Uuid, 自行開(kāi)發(fā)的Filter/Transform插件

• Output/Sink 插件

Elasticsearch, File, Hdfs, Jdbc, Kafka, Mysql, ClickHouse, Stdout, 自行開(kāi)發(fā)的Output 插件

4. Apache SeaTunnel環(huán)境依賴(lài)

SeaTunnel1.X支持Spark計(jì)算引擎，SeaTunnel2.X目前支持Spark/Flink兩種計(jì)算引擎，在筆者的實(shí)際項(xiàng)目中使用的是SeaTunnel1.X版本。

5. Apache SeaTunnel用戶(hù)使用情況

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

02 SeaTunnel應(yīng)用場(chǎng)景

SeaTunnel特別適合以下場(chǎng)景使用：

• 海量數(shù)據(jù)集成和ETL
• 海量數(shù)據(jù)聚合
• 多源數(shù)據(jù)處理

下面主要介紹SeaTunnel在交管行業(yè)中的應(yīng)用。

1. 交管行業(yè)數(shù)據(jù)簡(jiǎn)介

在交管行業(yè)中，數(shù)據(jù)主要包括駕駛?cè)恕④?chē)輛相關(guān)的數(shù)據(jù)，平時(shí)在道路上發(fā)生的一些交通警情數(shù)據(jù)，交通違法數(shù)據(jù)，機(jī)動(dòng)車(chē)登記信息，執(zhí)勤執(zhí)法的數(shù)據(jù)，交通事故以及其他一些互聯(lián)網(wǎng)數(shù)據(jù)，這些數(shù)據(jù)的量不是很大，另外還有卡口過(guò)車(chē)、車(chē)輛GPS數(shù)據(jù)，這兩種數(shù)據(jù)的數(shù)據(jù)量都比較大，例如一些省會(huì)城市，每秒鐘至少有幾千條過(guò)車(chē)數(shù)據(jù)，這些數(shù)據(jù)都是屬于交管行業(yè)內(nèi)的數(shù)據(jù)。

2. 交管行業(yè)數(shù)據(jù)特點(diǎn)

交管行業(yè)數(shù)據(jù)，跟互聯(lián)網(wǎng)行業(yè)的數(shù)據(jù)還是有很大區(qū)別的，首先這些數(shù)據(jù)的體量大小不一，并且分布在內(nèi)部的公安網(wǎng)以及智能專(zhuān)網(wǎng)，這兩個(gè)網(wǎng)之間是物理隔離的，我們需要把這些數(shù)據(jù)在兩個(gè)網(wǎng)絡(luò)之間轉(zhuǎn)移，在這個(gè)過(guò)程中，還要做一些數(shù)據(jù)處理。其次，在數(shù)據(jù)處理實(shí)時(shí)性方面的要求，并不是非常高，數(shù)據(jù)的更新頻率也不是很高。然后，在數(shù)據(jù)安全方面，要求比較高，數(shù)據(jù)是不能丟的，同時(shí)對(duì)保密性要求也比較高，所以具體的數(shù)據(jù)也不能展示出來(lái)。

03 相關(guān)業(yè)務(wù)痛點(diǎn)

1. 數(shù)據(jù)抽取限制較多

在做業(yè)務(wù)的過(guò)程中，會(huì)有一些業(yè)務(wù)痛點(diǎn)，首先因?yàn)榻还苄袠I(yè)是政府行業(yè)，基本各個(gè)子平臺(tái)的數(shù)據(jù)都是存儲(chǔ)在Oracle數(shù)據(jù)庫(kù)中的，我們需要把數(shù)據(jù)從Oracle數(shù)據(jù)庫(kù)中抽取到我們的數(shù)倉(cāng)里面，出于安全性的考慮，無(wú)法得到用戶(hù)級(jí)別的權(quán)限，我們只能通過(guò)一些視圖級(jí)別的用戶(hù)權(quán)限去處理數(shù)據(jù)，對(duì)于數(shù)據(jù)源表結(jié)構(gòu)的變更也無(wú)法及時(shí)知曉。其次，會(huì)話(huà)數(shù)是受到限制的，多線(xiàn)程抽取數(shù)據(jù)的話(huà)，如果會(huì)話(huà)數(shù)達(dá)到上限，連接就會(huì)受到影響，而且這個(gè)分配的用戶(hù)也同時(shí)會(huì)用于其他用途。最后，我們?cè)谔幚硪恍┰隽繑?shù)據(jù)的時(shí)候，一般情況下需要一個(gè)增量列，用于保持一個(gè)增量更新，很多時(shí)候，是沒(méi)辦法確定哪些列可以作為增量列的。以上就是在做業(yè)務(wù)的過(guò)程中，經(jīng)常會(huì)遇到的一些問(wèn)題，下圖也把這些問(wèn)題列舉了出來(lái)。

04 選擇SeaTunnel的原因

最初的時(shí)候，做數(shù)據(jù)處理、數(shù)據(jù)抽取的時(shí)候，并沒(méi)有使用SeaTunnel，而是使用Apache NiFi，這個(gè)工具功能比較強(qiáng)大而且全面，但是NiFi中用于數(shù)據(jù)處理的處理器比較多，而且數(shù)據(jù)處理鏈路中要做很多轉(zhuǎn)換，所以需要對(duì)NiFi里面的各種組件要非常熟悉，對(duì)使用者的要求也比較高。

1. SeaTunnel的優(yōu)勢(shì)

我們一開(kāi)始也用Spark程序做數(shù)據(jù)處理，對(duì)大數(shù)據(jù)相關(guān)人員的要求比較高，我們這邊大數(shù)據(jù)人員比較少，有時(shí)處理一些新的需求的時(shí)候，會(huì)比較繁忙。如果不需要通過(guò)編碼，而是直接使用工具，進(jìn)行簡(jiǎn)單的配置就能實(shí)現(xiàn)的話(huà)，會(huì)帶來(lái)較大的便利和效率的提高。

前面在SeaTunnel的介紹中，已經(jīng)講到SeaTunnel是比較易于使用的，安裝部署方便，開(kāi)箱即用，執(zhí)行效率很高，因?yàn)樗欠植际降模梢詰?yīng)用整個(gè)集群資源來(lái)做數(shù)據(jù)處理工作。

SeaTunnel無(wú)需編程，只要做簡(jiǎn)單的配置，并且它的Source和Sink都比較豐富，并且可以自己根據(jù)接口開(kāi)發(fā)需要的插件，對(duì)數(shù)據(jù)源的權(quán)限要求也不高。

更加重要的是，SeaTunnel是首個(gè)進(jìn)入Apache孵化的國(guó)人開(kāi)源數(shù)據(jù)集成平臺(tái)。

2. SeaTunnel的安裝部署

如上圖所示是SeaTunnel官方部署文檔，只需要簡(jiǎn)單幾步，就可以把SeaTunnel安裝到我們的環(huán)境之中，然后就可以使用了。

3. SeaTunnel配置文件

下圖所示是一個(gè)配置文件的示例，這個(gè)配置文件是SeaTunnel1.X版本的一個(gè)配置，一個(gè)完整的SeaTunnel配置包含spark, input, filter, output四部分，其中spark是spark相關(guān)的配置，例如，啟動(dòng)多少個(gè)executor，每個(gè) executor使用多少核數(shù)的CPU，多少內(nèi)存等，input可配置任意的input插件及其參數(shù)，具體參數(shù)隨不同的input插件而變化，filter可配置任意的filter插件及其參數(shù)，具體參數(shù)隨不同的filter插件而變化，filter中的多個(gè)插件按配置順序形成了數(shù)據(jù)處理的pipeline, 上一個(gè)filter的輸出是下一個(gè)filter的輸入，通過(guò)input插件把數(shù)據(jù)取出，成為了spark里面的一個(gè)數(shù)據(jù)集，然后filter插件會(huì)對(duì)這個(gè)數(shù)據(jù)集做一些轉(zhuǎn)換操作，output可配置任意的output插件及其參數(shù)，具體參數(shù)隨不同的output插件而變化，filter處理完的數(shù)據(jù)，會(huì)發(fā)送給output中配置的每個(gè)插件

4. SeaTunnel插件支持

如下圖所示，SeaTunnel支持的插件非常豐富，日常所能用到的基本都有。

這里面著重介紹一下filter插件中的sql插件，這個(gè)插件非常靈活，在用sql插件做轉(zhuǎn)換操作時(shí)，只要是sparksql里面支持的函數(shù)等內(nèi)容，都可以在這里使用，然后再output到目標(biāo)數(shù)據(jù)存儲(chǔ)，例如HDFS、Kafka、ES、Clickhouse等。

05 具體實(shí)現(xiàn)方案

接下來(lái)講一下具體的實(shí)現(xiàn)方案，在我們具體的業(yè)務(wù)中，如何把這些行業(yè)數(shù)據(jù)從智能專(zhuān)網(wǎng)直接抽取到公安網(wǎng)中，這里會(huì)涉及到數(shù)據(jù)的增量更新。

1. 數(shù)據(jù)增量更新具體實(shí)現(xiàn)

當(dāng)需要實(shí)現(xiàn)一個(gè)增量更新的時(shí)候，首先就是增量列的選擇，之前提到原先是用NiFi來(lái)做增量更新，但是對(duì)增量列的支持不是特別好，尤其是對(duì)日期類(lèi)型的支持不是很好。但是SeaTunnel對(duì)增量列的支持不受列的類(lèi)型限制，可以比較靈活的進(jìn)行選擇。

2. 具體方法

實(shí)際業(yè)務(wù)當(dāng)中，選取了記錄的更新時(shí)間列作為增量列，每次數(shù)據(jù)抽取過(guò)來(lái)，會(huì)記錄增量列的最大值，下次數(shù)據(jù)抽取時(shí)，可以從這個(gè)位置繼續(xù)抽取數(shù)據(jù)，這個(gè)也是受以前寫(xiě)spark程序的啟發(fā)，把checkpoint存儲(chǔ)在HDFS里面。當(dāng)然，增量列的選擇，在實(shí)際應(yīng)用中，除了更新時(shí)間，增量ID以外，還有其他業(yè)務(wù)字段可以做為增量列，增量列的選擇一定是根據(jù)真正的業(yè)務(wù)需求，實(shí)時(shí)的程度和粒度來(lái)決定的。

06 具體實(shí)現(xiàn)流程

做數(shù)據(jù)增量更新，最重要的是實(shí)現(xiàn)的思路，接下來(lái)詳細(xì)描述一下具體實(shí)現(xiàn)過(guò)程。

1. 確定運(yùn)算資源

首先，如下圖所示，先要確定計(jì)算資源，這里使用了spark，并且針對(duì)spark做了相關(guān)的配置。

2. 確定數(shù)據(jù)來(lái)源

選擇一個(gè)增量列，對(duì)增量列每次產(chǎn)生的最大值（checkpoint），保存在HDFS一個(gè)具體的目錄下。這里input插件選擇HDFS，每次產(chǎn)生的那個(gè)增量數(shù)據(jù)，指向HDFS的一個(gè)具體路徑下面，input插件有個(gè)通用參數(shù)叫做result_table_name，當(dāng)指定result_table_name時(shí)，處理后的數(shù)據(jù)，會(huì)被注冊(cè)為一個(gè)可供其他插件直接訪(fǎng)問(wèn)的數(shù)據(jù)集，或者被稱(chēng)為臨時(shí)表。當(dāng)增量列的最大值保存到HDFS之后，需要取出時(shí)，會(huì)保存在result_table_name指定的表中。接下來(lái)因?yàn)槭菑腛racle數(shù)據(jù)庫(kù)中取數(shù)據(jù)，所以設(shè)置相應(yīng)的Jdbc。當(dāng)數(shù)據(jù)量比較大的時(shí)候，還可以指定分區(qū)列，這樣的話(huà)，數(shù)據(jù)處理的效率會(huì)提高，詳細(xì)配置，如下圖所示。

3. 數(shù)據(jù)轉(zhuǎn)換

下圖所示是必要的數(shù)據(jù)轉(zhuǎn)換，在實(shí)際業(yè)務(wù)中，需要做一個(gè)過(guò)濾操作，取出大于最大更新時(shí)間的數(shù)據(jù)，convert插件里面做的是中間的一些數(shù)據(jù)類(lèi)型轉(zhuǎn)換操作，最后使用了一個(gè)sql插件，用于記錄本次取到的數(shù)據(jù)的一個(gè)最大值，用于下次取數(shù)的比較。

4. 數(shù)據(jù)輸出

下圖所示的是數(shù)據(jù)處理后的輸出，也就是output插件對(duì)應(yīng)的配置，具體是把數(shù)據(jù)抽取到Clickhouse里面。然后數(shù)據(jù)集里面，那個(gè)更新列的最大值，通過(guò)追加模式，寫(xiě)回到HDFS中，供下次使用。

5. 腳本和調(diào)度執(zhí)行

整個(gè)過(guò)程是通過(guò)下圖所示的shell腳本來(lái)做的，通過(guò)nohup后臺(tái)執(zhí)行的方式，利用Crontab進(jìn)行調(diào)度執(zhí)行，因?yàn)樵谖覀儗?shí)際的業(yè)務(wù)中，對(duì)定時(shí)調(diào)度的要求不是很高，所以可以采用Crontab或者開(kāi)源的Dolphin Scheduler都是可以滿(mǎn)足的。

下面的截圖，是實(shí)際運(yùn)行過(guò)程中，產(chǎn)生在HDFS上的增量文件，Crontab調(diào)度腳本，以及執(zhí)行過(guò)程中產(chǎn)生的一些Yarn任務(wù)列表。

在上述整體數(shù)據(jù)處理過(guò)程中，由于實(shí)際情況的限制，尤其我們的數(shù)據(jù)源是高度受限的Oracle數(shù)據(jù)庫(kù)。但是對(duì)于很多傳統(tǒng)公司，如果老系統(tǒng)是以O(shè)racle為主，并且掌控力度比較大的話(huà)，現(xiàn)在想做數(shù)據(jù)架構(gòu)升級(jí)，需要遷移Oracle中的數(shù)據(jù)，那么可以采用CDC讀取日志或者觸發(fā)器的方式，把數(shù)據(jù)變化寫(xiě)入到消息隊(duì)列里面，通過(guò)SeaTunnel就可以很容易的把數(shù)據(jù)實(shí)時(shí)寫(xiě)入到其他異構(gòu)的數(shù)據(jù)庫(kù)。

到此這篇關(guān)于Apache SeaTunnel實(shí)現(xiàn)非CDC數(shù)據(jù)抽取實(shí)踐的文章就介紹到這了,更多相關(guān)Apache SeaTunnel數(shù)據(jù)抽取內(nèi)容請(qǐng)搜索以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持！