Optymalizacja Iceberg: sortowanie i deletion vectors

Optymalizacja tabel Iceberg: sortowanie, statystyki i wektoryzowane usuwanie

Sortowanie danych podczas zapisu do tabel Iceberg pozwala silnikom wykonawczym pomijać niepotrzebne pliki na podstawie statystyk min/max w plikach manifest. Bez sortowania zakresy wartości w plikach nakładają się, co prowadzi do odczytu wszystkich danych nawet przy ścisłej filtracji.

Przykład tworzenia tabeli i zapisu bez sortowania:

spark.sql("""
CREATE TABLE habr.paper.write_ordered (
  id INT,
  name STRING,
  event STRING
)
USING ICEBERG
""")

data = [(i, f"name_{i}", f"event_{i}") for i in range(1, 1000000)]
df = spark.createDataFrame(data, ["id", "name", "event"]).repartition(16)
df.writeTo("habr.paper.write_ordered").append()

Zapytanie do metadanych pokazuje chaotyczne zakresy:

| id_min | id_max | file_path |

|--------|--------|-----------|

| 2 | 999994 | ... |

| 43 | 999988 | ... |

Filtr id < 100000 odczyta wszystkie pliki.

Aktywacja sortowania:

spark.sql("""
ALTER TABLE habr.paper.write_ordered
WRITE ORDERED BY id
""")

Powtórny zapis daje sekwencyjne zakresy:

| id_min | id_max | file_path |

|--------|--------|-----------|

| 1 | 196479 | ... |

| 196480 | 395983 | ... |

Teraz filtr id < 196470 czyta tylko jeden plik.

Konfiguracja zbierania statystyk dla kolumn

Iceberg domyślnie zbiera pełne statystyki min/max/truncated dla wszystkich kolumn, co zwiększa czas zapisu bez korzyści dla zapytań. W typowych tabelach filtracja odbywa się po 1-2 kolumnach, pozostałym statystyki nie są potrzebne.

Przykład płaskiej tabeli z 10 kolumnami:

spark.sql("""
CREATE TABLE habr.paper.typical_flat_table (
  id INT,
  name STRING,
  col_1 STRING,
  col_2 STRING,
  col_3 STRING,
  col_4 STRING,
  col_5 STRING,
  col_6 STRING,
  col_7 STRING,
  col_8 STRING
)
USING ICEBERG
""")

Konfiguracja przez TBLPROPERTIES:

spark.sql("""
ALTER TABLE habr.paper.typical_flat_table SET TBLPROPERTIES (
    'write.metadata.metrics.default' = 'counts',
    'write.metadata.metrics.column.id' = 'full',
    'write.metadata.metrics.column.col_1' = 'none'
)
""")

• default: 'counts' — tylko liczniki wierszy dla wszystkich kolumn
• full: min/max + truncated dla krytycznych filtrów (id)
• none: wyłączenie dla nieużywanych kolumn

To zmniejsza narzut zapisu i objętość metadanych bez utraty data skipping po kluczowych polach.

Wektoryzowane usuwanie w MoR

W strategii Merge-on-Read wydajność odczytu zależy od mechanizmu stosowania usunięć. Klasyczne pliki delete wymagają join'u z danymi.

Typy plików delete:

• Equality delete: reguły usuwania (przestarzałe w v3)
• Position delete: pozycje wierszy w plikach

Z Iceberg v3 position delete ewoluowały w Deletion vectors oparte na formacie Puffin. Deletion vector to maska bitmapowa, stosowana w jednym takcie do pliku danych.

Przykład position delete:

| file_path | pos |

|-----------|-----|

| ...parquet | 873 |

| ...parquet | 230 |

Zamiast join'u: załadowanie bitmapy + operacja logiczna AND. To przyspiesza odczyt tabel MoR 5-10x przy wysokim % usunięć.

Deletion vectors są szczególnie efektywne dla:

• Batchowych ETL z częstymi aktualizacjami
• Tabel z slowly changing dimensions
• Streaming z MoR

Co jest ważne

• Sortowanie WRITE ORDERED BY po filtrowanych kolumnach redukuje czytane pliki w analityce
• Konfiguracja write.metadata.metrics.* minimalizuje narzut zapisu, zachowując data skipping
• Deletion vectors w Iceberg v3 zastępują position delete, przyspieszając odczyt MoR przez bitmapy
• Optymalizacje działają bez zmian zapytań i logiki ETL
• Efekt kumuluje się: mniej plików, metadanych, obciążenia dla object storage

— Editorial Team