導讀：本文主要介紹騰訊是如何基于Apache Iceberg進行數據的入湖、治理以及后面的一些優化。將從數據入湖、數據治理服務、數據查詢優化以及未來展望四個方面展開介紹。

01數據入湖

本部分主要介紹Apache Iceberg基本概念以及結合Flink構建實時數據入湖鏈路。

1. Apache Iceberg是什么？

iceberg其實就是在存儲和計算層之間的一個表格式，表格式的作用主要是對計算引擎提供一個訪問存儲層的接口，能夠提供一些ACID語義和MVCC的能力，以及像歷史回溯之類的功能。像傳統的Hive表都會帶一些partition或者是數據的格式、壓縮格式、目錄信息等，但這些信息都存儲在Hive Metastore里，這里也可以將Metastore理解為一種文件的組織格式。從圖中可以看到，下層存儲層這塊是一個比較開放的存儲層，可以支持傳統的HDFS、對象存儲，文件格式也是支持行和列。

2.?Apache?Iceberg的特性

基于快照的讀寫分離和回溯

流批統一的寫入和讀取

不強綁定計算存儲引擎

ACID語義以及數據多版本

表，模式及分區的變更

Iceberg很重要的一個特性是支持快照的讀寫分離回溯以及不綁定任何計算存儲引擎，可以方便用戶快速接入自己的存儲引擎，比如Spark、Flink或者Presto，在Iceberg上用的時候可以基于Iceberg的API做一些connector。還有一個很重要的功能就是支持ACID語義及數據多版本控制，并且也支持表、模式及分區的一個變更。這些功能對于后面我們來構建準實時的數據入湖是一些非常重要的特性。

3.?Apache?Iceberg文件組織

圖中上層可以看到Commit的一個Timeline，Iceberg每次寫Commit操作都會生成一個新的Snapshot，比如snapshot-1其實是包含snapshot-0的所有數據，這里可以想象我們平時用git的時候，git的每一次新的提交都包含之前提交的所有信息，通過git log就可以看到，這里我們也可以這么理解snapshot的Timeline。下層是Iceberg基本的文件組織格式，每一個manifest管理n個DataFiles，DataFiles在manifest記錄的是DataFiles的路徑信息。通過這樣的一個文件組織格式，在讀取的時候就可以很方便地做Commit TimeLine，比如說現在是11點，Commit的是Snapshot-1，如果想讀snapshot-0的話其實只需要指定Snapshot-id，就可以很方便的實現數據的回溯。

4.?Apache?Iceberg讀寫流程

接下來這個是在寫Iceberg的時候的一個簡單讀寫過程。從這個鏈路上我們可以看到每次Iceberg的寫操作，比如說正在寫第一個S1的時候是沒有辦法讀的，只有發生了Commit C之后S1才是可讀的，這個時候如果有n個線程同時在讀但有一些線程在寫的時候，可以做到只有commit完整的數據之后，對用戶的讀操作才能被用戶的讀線程所看到。這也是Iceberg里面非常重要的特性，即讀寫分離。在對S4進行寫操作的時候，S3、S2、S1的讀操作是不受影響的，同時這個時候S4是沒有辦法讀得到的，只有Commit之后S4才能夠讀得到。Current Snapshot這時候就會指向S4，默認Iceberg讀的時候都是從最新的Current Snapshot開始讀Iceberg的數據。那么如果要讀前面的數據其實就可以指定Snapshot的id去進行數據回溯的讀。

5.?增量讀取

我們前面知道Iceberg每個不同的Snapshot都包含了之前的所有數據，比如說像圖中S2是包含了S1的數據，在每次讀取的時候就可以指定S1之間的增量即紫色這部分的數據，不需要每次重復地讀全量的數據。增量數據在后面構建準實時的入湖鏈路是非常重要的，因為從構建一個Flink的job比如從Kafka寫入數據到Iceberg，寫入之后下游的任務可以繼續讀Iceberg，讀的時候就可以選擇增量的讀取，在整個鏈路上就可以實現實時的入湖鏈路。

6.?Apache Iceberg Flink Sink

在Iceberg實時入湖的鏈路上我們用的是現在比較流行的實時計算Flink，我們知道上游InputStream是會源源不斷地往下游去寫的，如果在寫的時候不做多個并發寫的話對整個性能會有非常大的影響，因此把Iceberg Flink Sink拆成Writer和Commiter兩個部分。那么為什么只有一個Commiter呢？我們知道Iceberg的commit操作其實要到Hive MetaStore去獲得一個鎖，如果進行多個commit的話，每個commit都會到MetaStore獲取那個鎖，對MetaStore來說不管有多少commit操作都會進行排隊。所以這里只有一個并發commit是為了讓Iceberg前面n個Writer所寫的數據一次性從不可見的狀態變成可見的狀態。其實到commit狀態的數據已經都到了存儲上了，只是現在的狀態是不可見，這對準實時的數據接入有非常大的幫助，比如說HDFS寫數據的時候需要在一個temp里面把整個temp目錄move到可見的目錄上，這里其實數據已經全部都寫到存儲上了，所做的操作僅僅是把它的狀態從原來的不可見狀態變成了可見的狀態，也就是前面我們所說的每個snapshot commit操作。

7.?近實時數據入湖

上圖是我們內部大量采用的Iceberg實時入湖的一個簡單鏈路。上游消費Kafka或者Binlog的數據，中間采用Flink將數據寫入到Iceberg，下游可以基于Iceberg基礎上繼續再接Flink去做一些ETL或者其他的操作，也可以直接在Iceberg的基礎上跑Spark或者Presto的一些任務。

8.?實時入湖平臺化建設

這塊是我們內部在整個實時入湖鏈路的基礎上為了方便用戶而構建的一個任務管理平臺，可以非常方便地幫助用戶去新建一個端到端的入湖任務，也可以看到一些任務運行的狀態。

02數據治理服務

我們知道Flink任務跟傳統的任務不同的是，Flink任務是一個實時任務，實時任務的特點是常駐性，起了一個Flink任務就長時間運行，理想狀態下是不會中斷的。這種情況下，上游數據是源源不斷地進來的，Flink任務會源源不斷地進行Commit操作，如果對數據的實效性要求比較高的話，比如說Flink任務運行的時間是一分鐘、五分鐘或者是十分鐘級別。當運行了幾天或者是一兩個星期之后，在磁盤上發生Commit的次數就會非常地多，如果根據partition進行分區的話，磁盤上的文件數量會膨脹的非常大。如果是傳統的批任務的話，跑完一批之后在后面再跑一次compaction任務進行compact。實時任務因為是不中斷的，所以就會遇到小文件數量膨脹、元數據膨脹等的一些問題。

1.?實時數據入湖遇到的問題

我們知道實時任務為了保證實時性進行高頻的commit操作引起的小文件數量以及元數據數目膨脹引起查詢性能的降低，還有數據本身缺乏生命周期管理。有時候寫很多的數據到HDFS上，一段時間就要根據實際業務場景的需求對過去的數據進行清理，比如清理掉兩個星期前的數據，這時候就需要額外的服務化平臺去幫助用戶去做這個事情。還有一點就是數據的實時寫入并不能夠根據用戶真實的查詢條件進行分布，因為寫入只能根據寫入的條件去寫入，但是查詢條件比如說where或者過濾的條件可能是不一樣的，這個時候如果某些查詢經常頻繁發生的話，就會導致訪問這n個節點的查詢性能不太高，后面的服務也需要對數據做一個合理的重分布。

2.?不合適的小文件合并方案

我們在FlinkIcebergSink這邊嘗試了很多小文件合并的方案。我們知道Flink Sink上游每次都會做commit操作告訴當前commit操作是commit到了哪個Snapshot，snapshot里面增加了哪些文件，這些文件其實都是當前Snapshot里面的。比如說真實的數據文件在下游再接上一個operator的話，就可以對每次commit操作的文件進行compaction的操作。這里的rewrite也是同樣的道理，比如上游commit了90個文件，假設rewrite分到了30個文件，會對這30個文件進行rewrite操作把這30個文件rewrite成一個文件，把文件rewrite成一個文件的時候就會把rewrite的文件數量告訴下面的replace，replace知道當前的事物比如說新增30個rewrite，就會對30個rewrite文件再次進行commit操作，也就是說replace和sink其實做的都是commit操作，只是replace commit的是rewrite的結果，而sink commit的是上游寫下來的數據。replace之后生成新的snapshot的文件數量就是3而不是之前的90。這個合并方案我們之前也做過非常多的嘗試以及生產和測試環境的大量測試，實際證明這個其實是不合適的。因為下游所有的邏輯都是跟著Flink的任務走的，下游的不管是replace、rewrite或者是ScanTaskGen都要占用Flink TaskManager的計算資源。在計算資源有限的情況下，在后面再接上任務的compaction寫一些任務的話都會大量的占用整個集群的計算資源。如果是同步的任務，下游的rewrite都會阻塞掉上游數據的輸入，假設把它改造成異步的去跑后臺的線程，后臺的線程也要占用task的計算資源，所以整體在生產上面通過觀察發現，每次如果rewrite操作的時候整個集群主鏈路上的數據處理都會受到大大的影響，我們為了保證用戶對小文件合并的透明，就想到了要提出一個完整的數據治理的服務。

3.?架構總覽

上圖是我們數據治理的整體架構，中間藍色的四個框是我們主要要做的業務邏輯，我們把數據治理服務主要分為四大塊，第一塊是Compaction Service，Compaction Service是為了解決小文件過多的問題，我們知道Iceberg是讀寫分離的，我們可以對Iceberg實時鏈路上寫到磁盤上的一些小文件進行異步的compaction，這個compaction需要獨立的一部分計算資源。這樣的話，計算資源能夠幫助解決一些小文件的問題，又不會影響到主鏈路上的數據。Compaction Service服務會定期的根據Iceberg上游的表決定進行多大力度的合并，比方說文件合并的target是128M，每個snapshot文件數量是n，那就會根據這樣的數值去判斷當前的Iceberg寫入snapshot里面這些值的狀態是多少，以此決定要不要去觸發異步文件compaction的操作。Expiration Service是為了定期清理snapshot，比如說現在的snapshot里面的文件是1、2、3，合并之后的文件假設是1+2+3=6，我們知道6其實是包含1、2、3的一些數據，那么現在6的snapshot的數據就跟1、2、3的snapshot數據是重復的，在磁盤上是存在Double的數據，這個時候就需要定期的跑snapshot把合并之前的那些snapshot數據進行定期的清理操作，刪除一些冗余的數據可以大大的減少存儲壓力。第三個就是Clustering Service，對某些查詢比較頻繁的操作可以通過Clustering ?Service進行數據的重分布，比如說根據某些查詢的列進行數據的一些聚合，這樣的話在某些查詢經常發生時盡可能避免掃描過多的文件，對查詢的性能會有極大的提升。最后是Cleaning Service，針對某些用戶會有一些對過去數據的清理操作，后臺會有一個Service根據用戶會去配置表里，比如說配置表里面的TTL是3天或者30天，Cleaning Service就會定期地根據用戶配置的去清理這些過期的數據，這點比較類似kafka，kafka也有一些類似數據超時的機制。

4.?總體流程

上圖是整體數據的流程，可以看到Compaction整個服務中的數據流。我們先從Compaction服務來簡單的介紹，比如說用戶對Iceberg進行操作，我們在Iceberg接口這邊已經為Iceberg實現了Iceberg的Metrics匯報到外接系統的功能，首先Metrics的Reporter會將Iceberg的一些建表、刪除、更新或者任何Commit操作所產生的snapshot創建snapshot的summery匯報到iceberg的Metrics Event Handler那邊，Metrics Event Handler接收到不同的事件之后會根據不同事件的類型將這些事件存儲到MySQL。這里我們做了一個改造采用一個消息隊列來保證事件的時效性，并且對消息隊列里面的數據定期的保存在CheckPoint中。我們知道表其實是有兩種狀態，DDL狀態或者DML狀態，表的一些基礎的記錄信息比如表在compact之前/后的文件數量、以及表的文件數、操作的類型比如新建、commit、delete這些表所能提供的一些metrics信息，當數據通過消息隊列發送到中間階段的時候，中間階段內部有個規則管理器會去配置大量的規則，比如一些用戶希望表在每產生100個10M文件的時候就進行一次合并。這些規則的接口其實是開放給用戶去配置的，這些接口配置之后會將配置傳給下游的任務調度器，任務讀取器會讀取上游發送過來的一些規則，以決定現在要根據這些規則去起一個什么樣的任務。圖中我們可以看到下游會有很多不同的任務，比如JOB1、JOB2、JOB3這些任務目前是采用離線的Spark任務去跑上游發送過來的信息。執行的頻率有5分鐘、10分鐘和60分鐘，主要就是根據用戶所配置的表。用戶的表里面如果要保留過去100個文件，這個時候在監控里面看到用戶會一直頻繁地在提交，那么在單位時間內所產生的文件數量會非常非常的多，這個時候就需要更低的頻率比如5分鐘去對用戶的表執行一次compact操作。有些用戶比是10分鐘或者20分鐘才commit一次，這個時候可能只需要跑小時級別或者跑5個小時調度一次去做這個文件的處理，這邊是針對不同用戶的表的一些metrics的情況來決定應該將用戶的表放給哪一個粒度的調度任務去執行。那么我們知道每個job可以看到很多用戶的表，一個job可能會處理三四個表，每次處理完之后會將這三四個表的處理邏輯通過Metrics System消息隊列再反饋給剛剛我們記錄的MySQL，然后再通過Grafana或者一些監控的工具就可以看到整個任務的compaction的運行情況，包括compact之后表是什么狀態這里都可以看得到。還有一個記錄重要的點是每次compact的表任務執行的過程中花了多少時間，這樣就可以通過Job Handler動態地調整每個Job所負責的表的數量，比方說一個job執行的表是1、2、3三個表，發現1表跟2表執行compact任務花了10秒鐘就執行完了，3表執行了5分鐘，因為整個任務是并發提交的，所以需要等到第三個表執行完之后這個任務才能夠繼續調度下一次。這個時候就可以在下次調度的時候把3表調度到其他的任務區，1表就可以在一分鐘之內進行不斷地做文件數量的處理。

5.?實踐效果

對于用戶來說要使用compaction服務其實是非常簡單的，只需要創建一個表然后在表里面配置文件處理的參數，圖中表示的是snapshot保留過去10分鐘的snapshot，或者是過去10個snapshot的數量，metadata的文件保存過去10個metadata文件，每新產生5個snapshot就觸發一次rewrite操作。這個時候用戶只需要去配置后端的 metrics的匯報和文件的compaction以及文件的一些expiration，這些所有的動作在這個時候全部對用戶是透明的，用戶只需要去配置這個表，后面我們的服務都會自動地幫用戶去做好。

接下來可以看到整體的數據文件和meta文件數量，根據剛剛我們配置的值，如果在長時間運行compaction之后是能夠控制在一個比較合理的范圍。我們可以看到下面meta文件夾里面放的其實是iceberg的meta信息，包括像m0、avro這些都是snapshot的信息。上層的parquet是真實的數據文件，我們可以看到第二位有140、269、286這些的文件其實都是執行的compact之后的rewrite之后的文件。

03數據查詢優化

從剛剛這個合并里面我們可以知道，在做rewrite的時候只是把這些文件進行簡單的重寫，比方說將三個文件寫成一個文件，對整個的查詢性能其實是已經能夠得到一定的提升，因為相當于掃描的文件數量得到大大的降低，但是如果說真的要對某些頻繁發生的查詢性能進一步優化的話，這樣是遠遠不夠的。所以我們接下去會介紹我們在數據查詢優化方面所做的一些工作，首先介紹基于空間曲線算法優化iceberg的數據查詢效率。

1.?空間填充曲線簡介

首先介紹一下什么是空間查詢曲線？

在數學分析里面，空間填充曲線是一個參數化的組合函數，會將單位區間內的區間映射到單位的正方形或者立方體中。比如說在一個二維的空間里，可以通過一維的一條曲線穿過這個空間里面的每一個點，直到填充滿整個二維的空間平面，如果曲線所填充的粒度越來越密的話，其實整個二維平面會被填充滿，這個是數學的一個重要特性。查詢優化這塊其實就是基于這樣的特性，我們可以看到空間填充曲線的話，因為在二維平面里面它經過了二維平面的每一個點所以我們就可以將整個二維平面的空間降成一維，將多維的一個空間點轉化成一維對于后面的數據查詢優化算法是非常大的幫助，今天我們講的一個重點其實是利用了圖中第四個Z-Order的算法。

2.?GeoHash算法介紹

我們知道GeoHash算法就是基于Geo的特性來做的，我們可以看到圖中GeoHash算法其實是一個地理位置編碼將空間分成一個網格，在網格中可以定位某一些點以及哪些點離這個點最近。這個算法常用的一個場景是點評、外賣查看附近有多少外賣商家。從圖中我們可以看到，對于生成一個z order地址來說，比如說黑色的虛線所畫的這塊，四個地址我們就可以認為是靠的比較近的。一個點附近hash字符串如果前綴是一樣的我們就認為它的點是靠的比較近的，我們可以看到每個虛線框里面的前綴都是100，通過這樣的一個z地址可以把二維平面里面的數據進行降維，可以讓降維之后的距離變得比較近，通過這種算法我們就可以很方便的進行多維數據的聚合操作。

3. 為什么需要多維數據聚合

我們知道在N列數排序的時候比如order by FirstName，第一列的效果往往是比較好的，越往后可能效果會越差，到n列之后整個數據可能就是離散的。如果查詢條件比較多的情況下，文件過濾效果是比較差的，因為可能需要掃描表的所有數據才能去讀。數據如果呈現自然聚集的話會有幾個特點，比如單調遞增的id或者其他根據數據寫入的時間或是寫入前對數據進行的排序，這在這個例子中可以看到越往后幾列數據越亂。

4.?Iceberg表多維聚合

同理，我們在iceberg表中間做多維聚合時，首先將不同的snapshot的文件進行合并寫入小文件，然后進行optimize優化數據的分布，其實就是剛剛我們說的基于z order算法。我們將原先分散在集群中不同地方的文件進行重分布，這樣在查詢的時候只需要根據查詢optimize之后的結果文件就可以了。在這個例子中綠色的點可以理解為是符合過濾條件的，紅色是不符合過濾條件的，過濾條件指的是在做where的時候的過濾條件。可以看到在Snapshot N的時候數據是處于上游寫入的狀態，在第二個階段的時候進行optimize的時候可能是第一次進行optimize操作它的strategy是all需要掃描所有的文件，這個時候不符合過濾條件的都被聚集到m1和m4，m1、m4里面都是紅點不符合聚合條件。當然它也不能保證所有的紅點都scan到某些文件，因為數據要保證相同的過濾條件盡可能的聚集在一個文件里。在第二個階段比如說這個時候會在第一次進行optimize之后還會進行一次寫，因為上游只要是實時的就會不斷的往Snapshot里面寫入數據，比如f2001到f3000這一段寫了n個數據文件，在Snapshot N+3階段執行的是incremental的optimize就只去優化新寫入的這些文件，經過這樣的操作之后，文件的數量會大大的減少，在查詢的時候就可以避免非常多的沒有用的文件掃描操作。

我們在圖中可以看到，在Spark sql這邊其實是支持這樣的一個語法，比如說在optimize table employee zorder by first_name, last_name，我們假設first_name和last_name是二維的，因為是兩個字段其實就是二維的。首先根據first_name和last_name會先計算它的分區id，計算的規則目前我們實現的是一個固定的partition值，然后將這個分區id轉換成一個二進制，然后基于GEOHash算法。這里可以看到是交錯位去生成z地址的，Thomas 0放奇數位，More 0放偶數位，以奇偶交錯的方式生成z地址。這樣生成z地址可以看到Thomas More和Thomas Alva Edison以及Melisa Kort在第二列的排序都不是有序的。經過ZOrder之后前綴都是16個0，這時候就可以將這幾個聚合在一個文件里。當根據FirstName或者LastName進行查找的時候就可以很方便的根據ZOrder的地址進行查詢操作避免其他文件的掃描。可以看到它其實是根據多維Column生成Z地址，從f2到f1000的數據先根據里面的數據進行線性掃描，掃描生成一次地址，生成地址之后，接下來要做的一個很重要的事情就是Repartition。

根據Z地址進行Range重分區，因為只有根據Z地址進行Range重分區之后我們才能夠將原先分布在不同的點的數據的文件聚合到同一個點上。比如根據Z地址重分區之后可能生成了兩個partition，在查詢的時候就必須對數據進行寫回存儲。repartition之后要進行一個重寫操作，重寫之后生成一個新的snapshot N+1，這個過程也就是剛剛S N到S N+1的一個中間發生的詳細的過程。

經過事物回寫存儲之后，在查詢的時候就根據where條件智慧掃描m1和m3的數據，因為m2里面都是紅點不符合查詢的條件。

5.?查詢性能優化評測

經過我們的優化之后，可以看到圖底部有條select語句計算一個簡單的count，根據first_name和last_name進行過濾。上部分是在沒有優化之前在HDFS上和優化之后的性能對比，可以看到性能差距是非常大的，性能優化的一個主要的點就是把大量的小文件掃描的時間優化了。

04

未來展望

Iceberg內核及數據湖平臺化的工作規劃

1.?Iceberg內核能力

進一步優化索引系統，提升查詢性能。前面說到我們對查詢性能進行了zorder索引系統的構建以提升查詢的性能。但是zorder是有一定的局限性的，它需要根據查詢條件去進行re-clusting，如果查詢條件發生變化的話需要重新計算。另一個是如果查詢條件特別多達到幾十個或者上百個的話，zorder會面臨維度膨脹的問題，計算出來的z地址會非常的長。這一點我們后面會根據不同的場景需求進行不同的索引來盡量避免過多的文件掃描以及使用zorder沒有解決的一些問題。

增量讀取能力的增強，MOR方式入湖。我們知道現在的增量讀取是每次讀取的是incremental的snapshot，但這個時候如果發生replace操作的話，產生的rewrite之后的snapshot在這里的增量讀取整個語義目前是沒有很好的定義的，我們是希望在引擎層通過skip以及通過記錄rewrite之前的一些meta信息來解決這個問題，這塊的話也是我們下一步的一個任務的規劃。

SQL能力的增強。用戶可能很希望只通過一些SQL就能夠執行一些任務，比如用戶通過我們的平臺化建表，我們可以將這個表很方便地納入平臺的管理上，用戶自己建的表如果我們沒有辦法check到的話，也是需要提供一些SQL增強的能力，方便用戶更好的去執行使用iceberg過程中遇到的一些數據管理問題。

2.?數據湖平臺化建設

持續迭代數據治理服務平臺。我們會持續迭代數據治理服務化平臺，包括如何更好地去執行小文件合并的策略，包括怎么樣盡量避免重復的小文件的rewrite操作，已經重寫了的小文件什么時候將它合并到我應該要被寫的文件列表里，這些都是需要在后面不斷地迭代中去不斷地優化。

統一的元數據管理，元數據發現。對于用戶來說，因為數據湖本身是寫后schema的模式，所以用戶其實并不希望數據只是上傳了CSV或者JSON這樣的原始數據對于里面的schema其實可能并不知道，希望平臺能夠幫助他發現這些schema，這一塊也是平臺化建設中后面不斷地去優化的內容。

與更多數據系統打通，構建入湖+分析平臺。比如說數據已經寫入iceberg里面去了，可能會在上面繼續構建一些分析型的任務，比如更好地去優化presto的查詢性能，或者去進行平臺化構建更多分析型的任務，比如spark或者flink的批任務，這塊會跟更多的計算引擎去打通，更方便用戶使用從端到端，從入湖到分析的整個的一個鏈路平臺。

05問答環節

Q：iceberg的zorder有計劃提交到社區嗎？

A：我們現在是把剩下的一些測試和優化進一步做完善之后，后續有計劃反饋給社區。

Q：zorder優化是針對大量小文件場景下的優化嗎？

A：zorder優化的場景是某些經常發生的查詢，查詢條件相對固定這種情況下為了提升查詢的性能會用zorder的算法根據查詢的條件將數據進行重新的分布。小文件就是剛剛講到的上面會有一個compaction先將文件進行一個合并，然后zorder對數據進行重分布，以此來提升整個的查詢性能。

Q：也就是說小文件不多的時候效果也是很好的嗎？

A：是的。

Q：hive全量歷史數據的遷移入湖有相關的支持方案嗎？

A：這塊我們內部構建的一個平臺上其實是已經有相應的任務，比如執行了一個hive的一個入庫，從hive表導入到iceberg表的鏈路上，內部其實是會有平臺化上面會取一些類似于像spark導數任務去做這樣的事情。

Q：需要把hive的全量導數這部分全量讀一遍，然后才能入湖嗎？

A：現在我們是這么做的，就是spark導數。

Q：Clustering service是通過數據的冗余存儲，把數據以其他的column做partition或bucket或sort來提高file proning的效果嗎？

A：clustering就剛剛我們說的，它其實根據的是Column，比如說查詢的話是根據Column就是我會將這個column的數據在某些經常發生的一些查詢的條件的column數據會把它聚合在同一個文件里面去，其實在iceberg上這個階段如果說針對clustering數據生成新的snapshot，針對去讀snapshot的時候就會讀到下面clustering之后的一些文件。如果說是擔心數據冗余的話，因為bean重新生成后的數據文件肯定是需要通過snapshot對外暴露的，可以去跑一些數據定期清理的一些動作去完成這個事情。