分享嘉賓：王玉 唯品會 OLAP團隊負責人

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

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

ClickHouse數據導入的需求和痛點

ClickHouse出倉入倉工具選型

Hive to ClickHouse

ClickHouse to Hive

SeaTunnel與唯品會數據平臺的集成

未來展望

1.?唯品會數據OLAP架構

圖中是唯品會OLAP架構，我們負責的模塊是圖中的數據服務和計算引擎兩大部分。底層依賴的數據倉庫分為離線數倉、實時數倉和湖倉。計算引擎方面，我們使用Presto、Kylin和Clickhouse。雖然Clickhouse是一個存儲一體的OLAP數據庫，我們為了利用Clickhouse的優秀計算性能而將它歸入了計算引擎部分。基于OLAP組件之上，我們提供了SQL類數據服務和非SQL的唯品會自主分析，為不同智能服務。例如非SQL服務是為BI和商務提供更貼近業務的數據分析的服務。在數據服務至上抽象了多個數據應用。

2.?需求

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

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

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

3.?痛點

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

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

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

十億以上的數據可以平穩地在SeaTunnel和Spark中運行，DataX在數據量大以后性能壓力大，處理十億以上數據吃力。

在讀寫插件擴展性方面，SeaTunnel支持了多種數據源，支持用戶開發插件。SeaTunnel支持了數據導入Redis。

穩定性上，SeaTunnel和DataX由于是自成體系的工具，穩定性會更好。Spark的穩定性方面需要關注代碼質量。

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

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

圖中是一張Hive表，它是我們三級的商品維度表，包含品類商品、維度品類和用戶人群信息。表的主鍵是一個三級品類ct_third_id，下面的value是兩個uid的位圖，是用戶id的bitmap類型，我們要把這個Hive表導入到Clickhouse。

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

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

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

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

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

RDD以HashPartitioner進行分區將數據打散，防止數據傾斜。

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

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

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

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

圖中是SeaTunnel配置，我們把source配置為Clickhouse、sink配置為Hive，數據校驗也配置在Hive內。

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

各個公司都有自己的調度系統，例如白鯨、宙斯。唯品會的調度工具是數坊，調度工具中集成了數據傳輸工具。下面是調度系統架構圖，其中包含各類數據的出入倉。

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

我們把SeaTunnel集成到了調度系統中，數坊調度Master會根據任務類型把任務分配到對應的Agent上，根據Agent負載情況分配到合適的機器上運行，管控器把前臺的任務調度配置和信息拉取到后生成SeaTunnel cluster，在類似于k8s pod、cgroup隔離的虛擬環境內進行執行。運行結果會由調度平臺的數據質量監控判斷任務是否完成、是否運行成功，失敗時進行重跑和告警。

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

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

流量進入Kafka，通過Flink入倉，再通過ETL形成用戶標簽表，用戶標簽表生成后，我們通過Presto實現了的BitMap方法，把數據打成Hive中的寬表。用戶通過在人群系統頁面中框選詞條創建任務，提交騰群，生成SQL查詢Clickhouse BitMap。Clickhouse的BitMap查詢速度非常快，由天生優勢，我們需要把Hive的BitMap表通過SeaTunnel導入到Clickhouse中。圈完人群后我們需要把表落地，形成Clickhouse的一個分區或一條記錄，再把生成的結果BitMap表通過SeaTunnel存儲到Hive中去。最后同步工具會將Hive的BitMap人群結果同步給外部媒體倉庫Pika。每天圈20w個人群左右。

整個過程中SeaTunnel負責把數據從Hive導出到Clickhouse，Clickhouse的ETL流程完成后SeaTunnel把數據從Clickhouse導出到Hive。

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

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