Bloom-фильтры для data skipping

Ограничение min/max statistics

В предыдущем уроке мы узнали, что min/max statistics позволяют пропускать row groups. Но для высококардинальных столбцов (user_id, session_id, transaction_id) они бесполезны:

Row Group 1: user_id min=1, max=999999
Row Group 2: user_id min=2, max=999998
Row Group 3: user_id min=5, max=999997

Запрос: WHERE user_id = 42
→ Все row groups содержат диапазон [1..999999]
→ Все row groups проходят min/max проверку
→ Data skipping: 0% (читаем ВСЁ)

Min/max statistics не помогают, когда значения равномерно распределены по всем row groups. Нужна другая структура данных.

Что такое bloom filter

Bloom filter — это вероятностная структура данных, которая отвечает на вопрос “содержит ли множество элемент X?” с двумя возможными ответами:

• “Точно НЕТ” — элемента гарантированно нет (false negatives невозможны)
• “Возможно ДА” — элемент, вероятно, есть (false positives возможны)

Bloom filter для Row Group 1: {user_id: 42, 100, 205, 500, ...}

Запрос: WHERE user_id = 42
  Row Group 1: bloom.contains(42) = true  → МОЖЕТ содержать → READ
  Row Group 2: bloom.contains(42) = false → ТОЧНО нет       → SKIP!
  Row Group 3: bloom.contains(42) = false → ТОЧНО нет       → SKIP!

Результат: читаем 1 row group вместо 3

Как работает внутри

Bloom filter — это битовый массив фиксированного размера с K хеш-функциями:

Добавление элемента "42":
  hash1(42) = 3  → bits[3] = 1
  hash2(42) = 7  → bits[7] = 1
  hash3(42) = 12 → bits[12] = 1

Битовый массив: [0,0,0,1,0,0,0,1,0,0,0,0,1,0,0,0]
                       ↑           ↑           ↑

Проверка "42":
  bits[3]=1, bits[7]=1, bits[12]=1 → "Возможно ДА"

Проверка "99":
  hash1(99)=5 → bits[5]=0 → "Точно НЕТ" (достаточно одного 0)

False Positive Rate (FPP)

Bloom filter может ошибочно сказать “возможно ДА” для элемента, которого нет (false positive). Вероятность ложноположительного результата (FPP) зависит от:

• Размера битового массива (больше → меньше FPP)
• Количества хеш-функций (больше → сначала лучше, потом хуже)
• Количества элементов (больше → выше FPP)

Компромисс: меньше FPP = больше размер bloom filter = больше storage overhead.

Bloom filters в Spark / Parquet

Parquet поддерживает bloom filters начиная с версии 1.12:

# Включение bloom filter при записи Parquet
df.write \
  .option("parquet.bloom.filter.enabled#user_id", "true") \
  .option("parquet.bloom.filter.expected.ndv#user_id", "1000000") \
  .option("parquet.bloom.filter.fpp#user_id", "0.05") \
  .parquet("/data/events/")

# Глобальное включение bloom filter
spark.conf.set("spark.sql.parquet.bloom.filter.enabled", "true")

# Настройка per-column
spark.conf.set("parquet.bloom.filter.enabled#session_id", "true")
spark.conf.set("parquet.bloom.filter.expected.ndv#session_id", "5000000")

Параметры конфигурации

Bloom filters в Delta Lake

Delta Lake предоставляет более удобный API:

-- Создание bloom filter index
CREATE BLOOMFILTER INDEX ON TABLE events
FOR COLUMNS(user_id OPTIONS (fpp = 0.05, numItems = 1000000))

# Через DeltaTable API
delta_table = DeltaTable.forPath(spark, "/data/events/")
# Bloom filters настраиваются в table properties
spark.sql("""
  ALTER TABLE delta.`/data/events/`
  SET TBLPROPERTIES (
    'delta.bloomFilter.columns' = 'user_id,session_id',
    'delta.bloomFilter.fpp' = '0.05',
    'delta.bloomFilter.numItems' = '1000000'
  )
""")

Анти-паттерн: bloom filters на низкокардинальных столбцах

# ПЛОХО: bloom filter на столбце с 5 уникальными значениями
df.write \
  .option("parquet.bloom.filter.enabled#status", "true") \
  .option("parquet.bloom.filter.expected.ndv#status", "5") \
  .parquet("/data/orders/")

# status IN ('new', 'processing', 'shipped', 'delivered', 'cancelled')
# Min/max statistics: min='cancelled', max='shipped'
# Min/max УЖЕ достаточно для data skipping!
# Bloom filter добавляет overhead без пользы

Bloom filters оправданы для столбцов с тысячами и более уникальных значений, где min/max statistics бесполезны.

Практический сценарий

Задача: таблица событий (1TB, 5 миллиардов строк) с поиском по user_id (10M уникальных) и event_date.

# Оптимальная стратегия:
# 1. Partition by event_date (грубый фильтр по дате)
# 2. Z-ordering by user_id (кластеризация внутри partition)
# 3. Bloom filter on user_id (точечный поиск)

df.write \
  .partitionBy("event_date") \
  .option("parquet.bloom.filter.enabled#user_id", "true") \
  .option("parquet.bloom.filter.expected.ndv#user_id", "10000000") \
  .option("parquet.bloom.filter.fpp#user_id", "0.01") \
  .parquet("/data/events/")

# Для Delta Lake:
# OPTIMIZE events
# ZORDER BY (user_id)

Запрос WHERE user_id = 12345 AND event_date = '2024-02-01':

1. Partition pruning: читает только директорию event_date=2024-02-01 (~3GB из 1TB)
2. Bloom filter: пропускает row groups, не содержащие user_id=12345 (~50MB из 3GB)
3. Итого: читает 0.005% данных

Что дальше?

В следующем уроке мы разберём compaction и vacuum — как поддерживать оптимальный размер файлов и очищать устаревшие версии данных.