SeaTunnel 實踐 | 如何升級 OLAP 平臺？SeaTunnel 在唯品會的實踐！

導(dǎo)讀：本文主要帶來 SeaTunnel 在唯品會的多維度實踐。從選型組件方案，到 OLAP 方案的使用，到與唯品會數(shù)據(jù)平臺的集成方案，到人群方案中的使用。也同時會分享唯品會在使用 SeaTunnel 的一些集成和使用改造。

本文將圍繞下面幾點展開：

• ClickHouse數(shù)據(jù)導(dǎo)入的需求和痛點

• ClickHouse出倉入倉工具選型

• Hive to ClickHouse

• ClickHouse to Hive

• SeaTunnel與唯品會數(shù)據(jù)平臺的集成

• 未來展望

1.?唯品會數(shù)據(jù)OLAP架構(gòu)

圖中是唯品會OLAP架構(gòu)，我們負(fù)責(zé)的模塊是圖中的數(shù)據(jù)服務(wù)和計算引擎兩大部分。底層依賴的數(shù)據(jù)倉庫分為離線數(shù)倉、實時數(shù)倉和湖倉。計算引擎方面，我們使用Presto、Kylin和Clickhouse。雖然Clickhouse是一個存儲一體的OLAP數(shù)據(jù)庫，我們?yōu)榱死肅lickhouse的優(yōu)秀計算性能而將它歸入了計算引擎部分?；贠LAP組件之上，我們提供了SQL類數(shù)據(jù)服務(wù)和非SQL的唯品會自主分析，為不同智能服務(wù)。例如非SQL服務(wù)是為BI和商務(wù)提供更貼近業(yè)務(wù)的數(shù)據(jù)分析的服務(wù)。在數(shù)據(jù)服務(wù)至上抽象了多個數(shù)據(jù)應(yīng)用。

2.?需求

我們通過Presto Connector和Spark組件，把底層的Hive、Kudu、Alluxio組件打通。大數(shù)據(jù)組件之間可以互相導(dǎo)入導(dǎo)出數(shù)據(jù)，可以根據(jù)數(shù)據(jù)分析的需求和場景任意利用合適的組件分析數(shù)據(jù)。但我們引入Clickhouse時，它是一個數(shù)據(jù)孤島，數(shù)據(jù)的導(dǎo)入和導(dǎo)出比較困難。Hive和Clickhouse之間需要做很多工作才能實現(xiàn)導(dǎo)入導(dǎo)出。我們的第一個數(shù)據(jù)導(dǎo)入導(dǎo)出需求就是提升導(dǎo)入導(dǎo)出效率，把Clickhouse納入大數(shù)據(jù)體系中。

第二個需求是Presto跑SQL比較慢，圖中是一個慢SQL的例子。圖中的SQL where條件設(shè)置了日期、時間范圍和具體過濾條件，這類SQL使用由于Presto使用分區(qū)粒度下推，運行比較慢。即使用Hive的Bucket表和分桶等其他方式優(yōu)化后也是幾秒的返回時間、不能滿足業(yè)務(wù)要求。這種情況下，我們需要利用Clickhouse做離線的OLAP計算加速。

我們的實時數(shù)據(jù)是通過Kafka、Flink SQL方式寫入到Clickhouse中。但分析時只用實時數(shù)據(jù)是不夠的，需要用Hive維度表和已經(jīng)ETL計算號的T+1實時表一起在Clickhouse中做加速運輸。這需要把Hive的數(shù)據(jù)導(dǎo)入到Clickhouse中，這就是我們的第三個需求。

3.?痛點

首先，我們引入一項數(shù)據(jù)組件時要考慮其性能。Hive表粒度是五分鐘，是否有組件可以支撐五分鐘內(nèi)完成一個短小ETL流程并把ETL結(jié)果導(dǎo)入到Clickhouse中？第二，我們需要保證數(shù)據(jù)質(zhì)量，數(shù)據(jù)的準(zhǔn)確性需要有保障。Hive和Clickhouse的數(shù)據(jù)條數(shù)需要保障一致性，如果數(shù)據(jù)質(zhì)量出問題能否通過重跑等機制修復(fù)數(shù)據(jù)？第三，數(shù)據(jù)導(dǎo)入需要支持的數(shù)據(jù)類型是否完備？不同數(shù)據(jù)庫之間的數(shù)據(jù)類型和一些機制不同，我們有HiperLogLog和BitMap這類在某一存儲引擎中利用得比較多得數(shù)據(jù)類型，是否可以正確傳輸和識別，且可以較好地使用。

基于數(shù)據(jù)業(yè)務(wù)上的痛點，我們對數(shù)據(jù)出倉入倉工具進(jìn)行了對比和選擇。我們主要在開源工具中進(jìn)行選擇，沒有考慮商業(yè)出入倉工具，主要對比DataX、SeaTunnel和編寫Spark程序并用jdbc插入ClickHouse這三個方案中取舍。

SeaTunnel和Spark依賴唯品會自己的Yarn集群，可以直接實現(xiàn)分布式讀取和寫入。DataX是非分布式的，且Reader、Writer之間的啟動過程耗時時間長，性能普通，SeaTunnel和Spark處理數(shù)據(jù)的性能可以達(dá)到DataX的數(shù)倍。

十億以上的數(shù)據(jù)可以平穩(wěn)地在SeaTunnel和Spark中運行，DataX在數(shù)據(jù)量大以后性能壓力大，處理十億以上數(shù)據(jù)吃力。

在讀寫插件擴展性方面，SeaTunnel支持了多種數(shù)據(jù)源，支持用戶開發(fā)插件。SeaTunnel支持了數(shù)據(jù)導(dǎo)入Redis。

穩(wěn)定性上，SeaTunnel和DataX由于是自成體系的工具，穩(wěn)定性會更好。Spark的穩(wěn)定性方面需要關(guān)注代碼質(zhì)量。

我們的曝光表數(shù)據(jù)量每天在幾十億級，我們有5min內(nèi)完成數(shù)據(jù)處理的性能要求，我們存在數(shù)據(jù)導(dǎo)入導(dǎo)出到Redis的需求，我們需要導(dǎo)入導(dǎo)出工具可以接入到數(shù)據(jù)平臺上進(jìn)行任務(wù)調(diào)度。出于數(shù)據(jù)量級、性能、可擴展性、平臺兼容性幾方面的考慮，我們選擇了SeaTunnel作為我們的數(shù)倉導(dǎo)入導(dǎo)出工具。

下面將介紹我們對SeaTunnel的使用。

圖中是一張Hive表，它是我們?nèi)壍纳唐肪S度表，包含品類商品、維度品類和用戶人群信息。表的主鍵是一個三級品類ct_third_id，下面的value是兩個uid的位圖，是用戶id的bitmap類型，我們要把這個Hive表導(dǎo)入到Clickhouse。

SeaTunnel安裝簡單，官網(wǎng)文檔有介紹如何安裝。下圖中是SeaTunnel的配置，配置中env、source和sink是必不可少的。env部分，圖中的例子是Spark配置，配置了包括并發(fā)度等，可以調(diào)整這些參數(shù)。source部分是數(shù)據(jù)來源，這里配置了Hive數(shù)據(jù)源，包括一條Hive Select語句，Spark運行source配置中的SQL把數(shù)據(jù)讀出，此處支持UDF進(jìn)行簡單ETL；sink部分配置了Clickhouse，可以看到output_type=rowbinary，rowbinary是唯品會自研加速方案；pre_sql和check_sql是自研的用于數(shù)據(jù)校驗的功能，后面也會詳細(xì)介紹；clickhouse.socket_timeout和bulk_size都是可以根據(jù)實際情況進(jìn)行調(diào)整的。

運行SeaTunnel，執(zhí)行sh腳本文件、配置conf文件地址和yarn信息，后即可。

運行過程中會產(chǎn)生Spark日志，運行成功和運行中錯誤都可以在日志中查看。

為了更貼合業(yè)務(wù)，唯品會對SeaTunnel做了一些改進(jìn)。我們的ETL任務(wù)都是需要重跑的，我們支持了pre_sql和check_sql實現(xiàn)數(shù)據(jù)的重跑和對數(shù)。主要流程是在數(shù)據(jù)準(zhǔn)備好后，執(zhí)行pre_sql進(jìn)行預(yù)處理，在Clickhouse中執(zhí)行刪除舊分區(qū)數(shù)據(jù)、存放到某一目錄下在失敗時恢復(fù)該分區(qū)、rename這類操作。check_sql會檢驗，校驗通過后整個流程結(jié)束；如果檢驗不通過，根據(jù)配置進(jìn)行重跑，重跑不通過則報警到對應(yīng)負(fù)責(zé)人。

唯品會基于1.0版本SeaTunnel增加了RowBinary做加速，也讓HuperLogLog和BinaryBitmap的二進(jìn)制文件能更容易地從Hive導(dǎo)入到Clickhouse。我們在ClickHouse-jdbc、bulk_size、Hive-source幾處進(jìn)行了修改。使用CK-jdbc的extended api，以rowbinary方式將數(shù)據(jù)寫入CK，bulk_size引入了以rowbinary方式寫入CK的控制邏輯，Hive-source

RDD以HashPartitioner進(jìn)行分區(qū)將數(shù)據(jù)打散，防止數(shù)據(jù)傾斜。

我們還讓SeaTunnel支持了多類型，為了圈人群的功能，需要在Clickhouse、Preso、Spark中實現(xiàn)對應(yīng)的方法。我們在Clickhouse-jdbc中增加支持Batch特性的Callback、HttpEntity、RowBinaryStream，在Clickhouse-jdbc和Clickhouse-sink代碼中增加了bitmap類型映射，在Presto和Spark中實現(xiàn)了Clickhouse的Hyperloglog和Bitmap的function的UDF。

前面的配置中，Clickhouse-sink部分可以指定表名，這里有寫入本地表和分布式表的差異。寫入分布式表的性能比寫入本地表差對Clickhouse集群的壓力會更大，但在計算曝光表、流量表，ABTest等場景中需要兩表Join，兩張表量級均在幾十億。這時我們希望Join key落在本機，Join成本更小。我們建表時在Clickhouse的分布式表分布規(guī)則中配置murmurHash64規(guī)則，然后在Seatunnel的sink里直接配置分布式表，把寫入規(guī)則交給Clickhouse，利用了分布式表的特性進(jìn)行寫入。寫入本地表對Clickhouse的壓力會更小，寫入的性能也會更好。我們在Seatunnel里，根據(jù)sink的本地表，去Clickhouse的System.cluster表里獲取表的分布信息和機器分布host。然后根據(jù)均分規(guī)則寫入這些host。把數(shù)據(jù)分布式寫入的事情放到Seatunnel里來做。

針對本地表和分布式表的寫入，我們未來的改造方向是在Seatunnel實現(xiàn)一致性哈希，直接按照一定規(guī)則寫如Clickhouse、不依賴Clickhouse自身做數(shù)據(jù)分發(fā)，改善Clickhouse高CPU負(fù)載問題。

我們有圈人群的需求，每天唯品會為供應(yīng)商圈20萬個人群，比如80后、高富帥、白富美的人群集合。這些在Clickhouse中的Bitmap人群信息需要導(dǎo)出到Hive表，在Hive中與其他ETL任務(wù)進(jìn)行配合，最后推到PIKA交給外部媒體使用。我們使SeaTunnel將Clickhouse Bitmap人群數(shù)據(jù)反推到Hive。

圖中是SeaTunnel配置，我們把source配置為Clickhouse、sink配置為Hive，數(shù)據(jù)校驗也配置在Hive內(nèi)。

由于我們接入SeaTunnel較早，我們對一些模塊間進(jìn)行了加工，包括新增plugin-spark-sink-hive模塊、plugin-spark-source-ClickHouse模塊，重寫Spark Row相關(guān)方法，使其能封裝經(jīng)過Schem映射后的Clickhouse數(shù)據(jù)，重新構(gòu)造StructField并生成最終需要落地Hive的DataFrame。最新版本已經(jīng)有了很多source和sink組件，在SeaTunnel使用上更方便?，F(xiàn)在也可以在SeaTunnel中直接集成Flink connector。

各個公司都有自己的調(diào)度系統(tǒng)，例如白鯨、宙斯。唯品會的調(diào)度工具是數(shù)坊，調(diào)度工具中集成了數(shù)據(jù)傳輸工具。下面是調(diào)度系統(tǒng)架構(gòu)圖，其中包含各類數(shù)據(jù)的出入倉。

SeaTunnel任務(wù)類型集成到平臺中，圖中是數(shù)坊的定時任務(wù)截圖，可以看到選中的部分，是一個配置好的SeaTunnel任務(wù)，負(fù)責(zé)人、最近一次耗時，前后依賴任務(wù)的血緣信息，消耗的資源信息。下面展示了歷史運行實例信息。

我們把SeaTunnel集成到了調(diào)度系統(tǒng)中，數(shù)坊調(diào)度Master會根據(jù)任務(wù)類型把任務(wù)分配到對應(yīng)的Agent上，根據(jù)Agent負(fù)載情況分配到合適的機器上運行，管控器把前臺的任務(wù)調(diào)度配置和信息拉取到后生成SeaTunnel cluster，在類似于k8s pod、cgroup隔離的虛擬環(huán)境內(nèi)進(jìn)行執(zhí)行。運行結(jié)果會由調(diào)度平臺的數(shù)據(jù)質(zhì)量監(jiān)控判斷任務(wù)是否完成、是否運行成功，失敗時進(jìn)行重跑和告警。

SeaTunnel本身是一個工具化的組件，是為了進(jìn)行數(shù)據(jù)血緣，數(shù)據(jù)質(zhì)量，歷史記錄，高警監(jiān)控，還包括資源分配這些信息的管控。我們把SeaTunnel集成到平臺中，可以利用平臺優(yōu)勢利用好SeaTunnel。

圈存人群中利用了SeaTunnel進(jìn)行處理。我們通過打點數(shù)據(jù)，把圈存人群按照路徑和使用情況分為不同的人，或稱千人千面，把用戶打上標(biāo)簽，圈出的某一類人群推送給用戶、分析師和供應(yīng)商。

流量進(jìn)入Kafka，通過Flink入倉，再通過ETL形成用戶標(biāo)簽表，用戶標(biāo)簽表生成后，我們通過Presto實現(xiàn)了的BitMap方法，把數(shù)據(jù)打成Hive中的寬表。用戶通過在人群系統(tǒng)頁面中框選詞條創(chuàng)建任務(wù)，提交騰群，生成SQL查詢Clickhouse BitMap。Clickhouse的BitMap查詢速度非?？?，由天生優(yōu)勢，我們需要把Hive的BitMap表通過SeaTunnel導(dǎo)入到Clickhouse中。圈完人群后我們需要把表落地，形成Clickhouse的一個分區(qū)或一條記錄，再把生成的結(jié)果BitMap表通過SeaTunnel存儲到Hive中去。最后同步工具會將Hive的BitMap人群結(jié)果同步給外部媒體倉庫Pika。每天圈20w個人群左右。

整個過程中SeaTunnel負(fù)責(zé)把數(shù)據(jù)從Hive導(dǎo)出到Clickhouse，Clickhouse的ETL流程完成后SeaTunnel把數(shù)據(jù)從Clickhouse導(dǎo)出到Hive。

為了完成這樣的需求，我們在Presto和Spark端現(xiàn)ClickHouse的Hyperloglog和BitMap的function的UDF；我們還開發(fā)Seatunnel接口，使得用戶在ClickHouse里使用Bitmap方法圈出來的人群，可以直接通過SeaTunnel寫入Hive表，無需中間落地步驟。用戶也可以在Hive里通過spark圈人群或者反解人群bitmap，調(diào)用SeaTunnel接口，使數(shù)據(jù)直接傳輸?shù)紺lickHouse的結(jié)果表，而無需中間落地。

后續(xù)我們會進(jìn)一步改善Clickhouse寫入數(shù)據(jù)時CPU負(fù)載高的問題，下一步會在SeaTunnel中實現(xiàn)Clickhouse數(shù)據(jù)源和讀取端的CK-local模式，讀寫分離，減輕Clickhouse壓力。未來我們也會增加更多sink支持，如數(shù)據(jù)推送到Pika和相應(yīng)的數(shù)據(jù)檢查。

今天的分享就到這里，謝謝大家。

在文末分享、點贊、在看，給個3連擊唄~

分享嘉賓：

??分享、點贊、在看，給個3連擊唄！??