陳胡:Apache SeaTunnel實現 非CDC數據抽取實踐

来源:https://www.cnblogs.com/datafuntalk/archive/2022/05/19/16290103.html
-Advertisement-
Play Games

導讀: 隨著全球數據量的不斷增長,越來越多的業務需要支撐高併發、高可用、可擴展、以及海量的數據存儲,在這種情況下,適應各種場景的數據存儲技術也不斷的產生和發展。與此同時,各種資料庫之間的同步與轉化的需求也不斷增多,數據集成成為大數據領域的熱門方向,於是SeaTunnel應運而生。SeaTunnel是 ...


file


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

今天的介紹會圍繞下麵六點展開:

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

--

01 SeaTunnel簡介

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

1. Apache SeaTunnel整體介紹

互聯網行業數據量非常大,對性能還有其他各方面的技術要求都非常高,在筆者所在的交管行業中,情況就不太一樣,各方面的要求也沒有互聯網行業那麼高,在具體的數據集成應用中,主要是使用SeaTunnel1.X版本。

file

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

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

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

  • 數據丟失與重覆
  • 數據集成中任務堆積與延遲
  • 數據同步較低的吞吐量
  • Spark/Flink應用到生產環境周期較長、複雜度較高
  • 缺少應用運行狀態的監控

2. Apache SeaTunnel技術特性

file

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

  • 簡單易用,開發配置簡單、靈活,無需編碼開發,支持通過SQL進行數據處理和聚合,使用成本低
  • 分散式,高性能,經歷大規模生產環境使用和海量數據檢驗,成熟穩定
  • 模塊化和插件化,內置豐富插件,並且可以開發定製個性化插件,支持熱插拔,具備高擴展性
  • 使用Spark/Flink作為底層數據同步引擎使其具備分散式執行能力

3. Apache SeaTunnel工作流程

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

file

在插件方面,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版本。

file

5. Apache SeaTunnel用戶使用情況

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

file

--

02 SeaTunnel應用場景

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

  • 海量數據集成和ETL
  • 海量數據聚合
  • 多源數據處理

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

1. 交管行業數據簡介

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

file

2. 交管行業數據特點

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

file

--

03 相關業務痛點

1. 數據抽取限制較多

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

file

--

04 選擇SeaTunnel的原因

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

1. SeaTunnel的優勢

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

file

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

SeaTunnel無需編程,只要做簡單的配置,並且它的Source和Sink都比較豐富,並且可以自己根據介面開發需要的插件,對數據源的許可權要求也不高。

更加重要的是,SeaTunnel是首個進入Apache孵化的國人開源數據集成平臺。

2. SeaTunnel的安裝部署

file

如上圖所示是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中配置的每個插件

file

4. SeaTunnel插件支持

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

file

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

--

05 具體實現方案

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

1. 數據增量更新具體實現

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

file

2. 具體方法

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

--

06 具體實現流程

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

1. 確定運算資源

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

file

2. 確定數據來源

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

file

3. 數據轉換

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

file

4. 數據輸出

下圖所示的是數據處理後的輸出,也就是output插件對應的配置,具體是把數據抽取到Clickhouse裡面。然後數據集裡面,那個更新列的最大值,通過追加模式,寫回到HDFS中,供下次使用。

file

5. 腳本和調度執行

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

file

下麵的截圖,是實際運行過程中,產生在HDFS上的增量文件,Crontab調度腳本,以及執行過程中產生的一些Yarn任務列表。

file

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


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • sed sed命令 sed全稱是:Stream EDitor(流編輯器。 Linux sed 命令是利用腳本來處理文本文件,sed 可依照腳本的指令來處理、編輯文本文件。Sed 主要用來自動編輯一個或多個文件、簡化對文件的反覆操作、編寫轉換程式等。 當處理數據時,sed 從輸入源一次讀入一行,並將它 ...
  • 常用命令 sudo -i然後輸入密碼登錄root賬戶(群暉預設只能使用admin賬號登陸) vim xxx編輯(編輯是進去之後按i,退出並保存是按esc,然後:wq!再回車) mkdir xx創建文件夾 準備 1. 群暉一般預設安裝docker 我們不需要進行安裝,如果沒有安裝,則去套件中心進行安裝 ...
  • 鏡像下載、功能變數名稱解析、時間同步請點擊 阿裡雲開源鏡像站 你可以在虛擬機環境里運行任何操作系統,不論是測試還是為了某種需要。 對於 Linux 而言,它在虛擬環境下的性能會優於其他操作系統。即便你可能會猶豫是否在物理機(裸金屬)上安裝 Linux 系統,你仍然可以在虛擬機中安裝一個性能幾乎和物理機一樣好 ...
  • 本文例子參考《STM32單片機開發實例——基於Proteus虛擬模擬與HAL/LL庫》 源代碼:https://github.com/LanLinnet/STM33F103R6 項目要求 單片機將由串口收到的1位元組數據存入Flash ROM的指定地址;按下按鈕BTN,單片機將存儲在Flash ROM ...
  • 為什麼要使用Redwood Redwood是一個全棧web框架,旨在幫助你從副業項目發展到創業。Redwood的特色是一個端到端的開發工作流,它將React、GraphQL、Prisma、TypeScript、Jest和Storybook中最好的部分編織在一起。 RedwoodJS 是集成 、Pri ...
  • Termius是微軟的一款SSH終端工具,它支持多平臺。而且操作界面十分ha好看且簡潔,今天分享給大家❤️ 軟體下載 關註下方公眾號,回覆termius獲取下載地址 軟體功能介紹 Termius Mac破解版是一款非常好用而且漂亮的SSH客戶端,能快速遠程式控制制伺服器,可以定製自己喜歡的主題,支持FT ...
  • 一、概述 Impala 直接針對存儲在 HDFS、HBase或 Amazon Simple Storage Service (S3)中的 Apache Hadoop 數據提供快速的互動式 SQL 查詢。Impala是一個基於Hive、分散式、大規模並行處理(MPP:Massively Paralle ...
  • 軟硬體環境 軟體 版本 操作系統 Redhat 7 及以上版本,麒麟V10 DM 資料庫 DM 8.0 及以上版本 CPU 架構 x86、ARM、龍芯、飛騰等國內外主流 CPU DM-Oracle 環境準備 在DM伺服器中完成Dblink環境準備工作。 配置Oracle oci客戶端 在Oracle ...
一周排行
    -Advertisement-
    Play Games
  • 使用原因: 在我們服務端調用第三方介面時,如:支付寶,微信支付,我們服務端需要模擬http請求並加上一些自己的邏輯響應給前端最終達到我們想要的效果 1.使用WebClient 引用命名空間 using System.Net; using System.Collections.Specialized; ...
  • WPF 實現帶蒙版的 MessageBox 消息提示框 WPF 實現帶蒙版的 MessageBox 消息提示框 作者:WPFDevelopersOrg 原文鏈接: https://github.com/WPFDevelopersOrg/WPFDevelopers.Minimal 框架使用大於等於.N ...
  • 一、JSON(JavaScript Object Notation)的簡介: ① JSON和XML類似,主要用於存儲和傳輸文本信息,但是和XML相比,JSON更小、更快、更易解析、更易編寫與閱讀。 ② C、Python、C++、Java、PHP、Go等編程語言都支持JSON。 二、JSON語法規則: ...
  • 1.避免Scoped模式註冊的服務變成Singleton模式 當提供一個生命周期模式為Singleton的服務實例時,如果發現該服務中還依賴生命周期模式為Scoped的服務實例(Scoped服務實例將被一個Singleton服務實例所引用),那麼這個被依賴的Scoped服務實例最終會成為一個Sing ...
  • 索引時資料庫提高數據查詢處理性能的一個非常關鍵的技術,索引的使用可以對性能產生上百倍甚至上千倍的影響。接下來,會介紹索引的基本原理、概念,並深入學習資料庫中所使用的索引結構和存儲方式,以及如何管理、維護索引等。 1.索引的基本概念 索引時用來快速查詢表記錄的一種存儲結構,一般使用索引有一下兩個方面: ...
  • django2 路由控制器 Route路由,是一種映射關係。路由是把客戶端請求的url路徑和用戶請求的應用程式,這裡意指django裡面的視圖進行綁定映射的一種關係。 請求路徑和視圖函數不是一一對應的關係 在django中所有的路由最終都被保存到一個叫urlpatterns的文件里,並且該文件必須在 ...
  • 1、我們的目標是獲取微博某博主的全部圖片、視頻 2、拿到網址後 我們先觀察 打開F12 隨著下滑我們發現載入出來了一個叫mymblog的東西,展開響應發現需要的東西就在裡面 3、重點來了!!! 通過觀察發現第二頁比第一頁多了參數since_id 而第二頁的since_id參數剛好在上一頁中能獲取到, ...
  • 一、實現原理 在Servlet3協議規範中,包含在JAR文件/META-INFO/resources/路徑下的資源可以直接訪問。 二、舉例說明 如下圖所示,是我新建的一個Spring Boot Starter項目:zimug-minitor-threadpool,用於實現可配置、可觀測的線程池。其中 ...
  • 精華筆記: static final常量:應用率高 必須聲明同時初始化 由類名打點來訪問,不能被改變 建議:常量所有字母都大寫,多個單詞用_分隔 編譯器在編譯時會將常量直接替換為具體的數,效率高 何時用:數據永遠不變,並且經常使用 抽象方法: 由abstract修飾 只有方法的定義,沒有具體的實現( ...
  • Python有一個for...else語法,它的寫法如下 for i in range(0,100): if i == 3: break else: print("Not found") 該語句表示:若for迴圈遍歷完畢,則執行else部分的語句。也就是說上述代碼不會有任何輸出,而下述代碼會輸出“N ...