Bloom Filter в Spark SQL: probabilistic dedup и ускорение join на больших таблицах

FPR ≈ (1 - e^(-kn/m))^k

k_opt = (m/n) * ln(2) ≈ 0.693 * (m/n)

m = -n * ln(FPR) / (ln(2))^2 ≈ -1.44 * n * log2(FPR)

# Включение Runtime Bloom Filter в Spark (по умолчанию включён с Spark 3.3)
spark.conf.set("spark.sql.runtime.filter.enabled", "true")

# Максимальный размер фильтра для Broadcast (по умолчанию 4 МБ)
spark.conf.set("spark.sql.runtime.filter.bloomFilter.maxNumBits", "33554432")  # 4 МБ в битах

# Максимальное число строк для сбора ключей (для построения фильтра)
spark.conf.set("spark.sql.runtime.filter.bloomFilter.maxNumItems", "4000000")

# Порог для Broadcast (если probe side > порога, BF не строится)
spark.conf.set("spark.sql.runtime.filter.semiJoinReductionThreshold", "0.5")

# Adaptive Query Execution должен быть включён для Runtime Filtering
spark.conf.set("spark.sql.adaptive.enabled", "true")

df_large = spark.table("silver.clickstream")   # 2 млрд строк
df_small = spark.table("silver.valid_sessions") # 10 млн строк

result = df_large.join(df_small, on="session_id", how="inner")
result.explain(mode="formatted")

== Physical Plan ==
BroadcastHashJoin [session_id], [session_id], Inner, BuildRight, false
:- Filter isnotnull(session_id)
:  +- BloomFilterMightContain(bloomFilter=<BF>, value=session_id)   ← применение
:     +- FileScan parquet silver.clickstream [session_id, ...]
+- BroadcastExchange HashedRelationBroadcastMode(...)
   +- BloomFilterAggregate(estimatedNumItems=10000000, numBits=94906265)  ← построение
      +- FileScan parquet silver.valid_sessions [session_id, ...]

Без BF: Инкремент (50 млн) Anti-Join History (2 млрд) = дорого
С BF:   Инкремент (50 млн) → BF pre-filter → 5 млн → Anti-Join History (2 млрд)
        При этом History тоже меньше нужно читать (за счёт file skipping)

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("bloom_filter_demo").getOrCreate()

# Читаем историческую таблицу (данные за 90 дней)
df_history = spark.table("silver.transactions").filter(
    F.col("transaction_date") >= F.date_sub(F.current_date(), 90)
)

# Создаём Bloom Filter по ключевой колонке
# Параметры:
#   colName          - колонка, по которой строится фильтр
#   expectedNumItems - ожидаемое число уникальных значений в колонке
#   fpp              - допустимый False Positive Rate (0.0 до 1.0)
bloom_filter = df_history.stat.bloomFilter(
    colName="transaction_id",
    expectedNumItems=500_000_000,  # 500M уникальных transaction_id
    fpp=0.01                        # FPR = 1%
)

print(f"Bloom Filter object: {bloom_filter}")
print(f"Expected bits: ~{500_000_000 * 9.6 / 8 / 1024 / 1024:.0f} МБ")

from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType

# Делаем Broadcast: Spark рассылает объект фильтра на все экзекьюторы
# Это критично - без broadcast каждый экзекьютор создаст свою копию (дорого)
bloom_filter_broadcast = spark.sparkContext.broadcast(bloom_filter)

@udf(returnType=BooleanType())
def bloom_filter_udf(value):
    """
    UDF: проверяет, может ли значение находиться в историческом наборе.

    Возвращает:
        True  - значение ВОЗМОЖНО есть в истории (нужна точная проверка)
        False - значение ТОЧНО отсутствует в истории (можно вставлять)

    ВАЖНО: False Negative невозможен - мы никогда не потеряем новую строку.
    """
    if value is None:
        # NULL-ключи нельзя хэшировать корректно - пропускаем их дальше
        return True
    return bloom_filter_broadcast.value.mightContain(value)


# Читаем инкрементальный батч
df_increment = spark.table("bronze.transactions_incoming")

# ШАГ 1: Bloom Filter pre-filter - быстро, вероятностно
# Оставляем только строки, которые МОГУТ быть новыми (+ FPR мусор)
df_maybe_new = df_increment.filter(~bloom_filter_udf(F.col("transaction_id")))
# Знак ~ (NOT): нас интересуют строки, которых НЕТ в истории
# bloom_filter_udf возвращает True = "есть в истории" → мы их отбрасываем
# bloom_filter_udf возвращает False = "точно новые" → мы их оставляем

print(f"Инкремент до BF: {df_increment.count():,}")
print(f"После BF pre-filter: {df_maybe_new.count():,}")

# ШАГ 2: Точный Anti-Join для устранения False Positives
# На этом этапе df_maybe_new уже в 10–20 раз меньше исходного инкремента
# → Shuffle в Anti-Join в 10–20 раз дешевле

# Берём только ключи из истории для Anti-Join (минимизируем читаемые данные)
history_keys = (
    spark.table("silver.transactions")
    .filter(F.col("transaction_date") >= F.date_sub(F.current_date(), 90))
    .select("transaction_id")
    .distinct()
)

# Финальный Anti-Join: гарантированно новые строки
df_truly_new = df_maybe_new.join(
    history_keys,
    on="transaction_id",
    how="left_anti"
)

print(f"Гарантированно новые строки: {df_truly_new.count():,}")

# Шаг 1: Сколько строк прошло BF pre-filter?
count_after_bf = df_maybe_new.count()

# Шаг 2: Сколько из них - настоящие новые (после точного Anti-Join)?
count_truly_new = df_truly_new.count()

# Шаг 3: Разница - это False Positives (строки уже есть в истории, но BF пропустил)
false_positives = count_after_bf - count_truly_new
fpr_actual = false_positives / count_after_bf if count_after_bf > 0 else 0

print(f"Строк после BF:         {count_after_bf:,}")
print(f"Истинно новых строк:    {count_truly_new:,}")
print(f"False positives:         {false_positives:,}")
print(f"Фактический FPR:         {fpr_actual:.3%}")

# Строки инкремента, отсечённые BF (True Negatives - гарантированно верно)
count_original = df_increment.count()
true_negatives_by_bf = count_original - count_after_bf
print(f"Отсечено фильтром (TN): {true_negatives_by_bf:,}")
print(f"Эффективность pre-filter: {true_negatives_by_bf / count_original:.1%}")

import io
from pyspark.ml.feature import BucketedRandomProjectionLSH  # не нужен, просто пример импорта

# Сериализация фильтра в байты
def serialize_bloom_filter(bf) -> bytes:
    """Сохраняет Bloom Filter в байтовый массив для хранения в HDFS/S3."""
    buffer = io.BytesIO()
    # Spark BloomFilter имеет метод writeTo(OutputStream)
    # Используем Java-обёртку через py4j
    bf._jvm_bf.writeTo(buffer)  # псевдокод - в реальности нужен Java IO bridge
    return buffer.getvalue()


def deserialize_bloom_filter(spark: SparkSession, data: bytes):
    """Загружает Bloom Filter из байтового массива."""
    buffer = io.BytesIO(data)
    return spark._jvm.org.apache.spark.util.sketch.BloomFilter.readFrom(buffer)


# Практичный паттерн: кэшируем фильтр в памяти между запусками
import pickle

# ВАРИАНТ A: Pickle через pyspark util (работает с Python Bloom Filter)
# ВАРИАНТ B: Хранить в Delta/Iceberg-таблице как BINARY колонку
# ВАРИАНТ C: Пересоздавать фильтр каждый запуск (чаще всего - самый простой)

# Пересоздание фильтра - наиболее надёжный подход в production
def build_bloom_filter(spark, table: str, key_col: str, days: int = 90, fpp: float = 0.01):
    """
    Строит свежий Bloom Filter из ключей таблицы за последние N дней.

    Рекомендуется запускать при старте пайплайна или раз в час
    для поддержания актуальности фильтра.
    """
    df = (
        spark.table(table)
        .filter(F.col("transaction_date") >= F.date_sub(F.current_date(), days))
        .select(key_col)
        .dropna()
        .distinct()
    )

    # Считаем количество уникальных ключей для правильного sizing
    n_unique = df.count()
    expected_items = int(n_unique * 1.3)  # Запас 30% для новых данных

    bf = df.stat.bloomFilter(key_col, expectedNumItems=expected_items, fpp=fpp)
    print(f"BF built: {n_unique:,} unique keys, expected={expected_items:,}, fpp={fpp}")
    return bf

from pyspark.sql import SparkSession, functions as F, DataFrame
from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType
from typing import Optional
import logging

logger = logging.getLogger(__name__)


class BloomFilterDeduplicator:
    """
    Трёхуровневый дедупликатор для инкрементальных ETL-пайплайнов.

    Уровни:
    1. Intra-batch dedup (dropDuplicates)
    2. Bloom Filter pre-filter (вероятностный)
    3. Точный Anti-Join (детерминированный)

    Гарантии:
    - False Negative невозможен: ни одна реально новая строка не будет потеряна
    - False Positive возможен: небольшой % уже существующих строк пройдёт в Anti-Join
    """

    def __init__(
        self,
        spark: SparkSession,
        history_table: str,
        key_col: str,
        lookback_days: int = 90,
        bloom_fpp: float = 0.01,
        bloom_expected_items: Optional[int] = None,
    ):
        self.spark = spark
        self.history_table = history_table
        self.key_col = key_col
        self.lookback_days = lookback_days
        self.bloom_fpp = bloom_fpp
        self._bloom_filter = None
        self._bloom_broadcast = None
        self._bloom_expected_items = bloom_expected_items

    def build_bloom_filter(self) -> None:
        """
        Строит Bloom Filter из исторических данных.
        Вызывается один раз при инициализации пайплайна.

        Читает только ключевую колонку за последние lookback_days —
        минимальный IO при построении.
        """
        logger.info(f"Building Bloom Filter from {self.history_table} "
                    f"(last {self.lookback_days} days)...")

        df_keys = (
            self.spark.table(self.history_table)
            .filter(
                F.col("event_date") >= F.date_sub(F.current_date(), self.lookback_days)
            )
            .select(self.key_col)
            .dropna(subset=[self.key_col])
            .distinct()
        )

        # Если expectedNumItems не задан - считаем точно (extra action)
        if self._bloom_expected_items is None:
            n = df_keys.count()
            expected = int(n * 1.3)  # +30% запас для роста данных
            logger.info(f"Counted {n:,} unique keys, setting expectedNumItems={expected:,}")
        else:
            expected = self._bloom_expected_items

        self._bloom_filter = df_keys.stat.bloomFilter(
            colName=self.key_col,
            expectedNumItems=expected,
            fpp=self.bloom_fpp,
        )

        # Broadcast: рассылаем фильтр на все экзекьюторы один раз
        self._bloom_broadcast = self.spark.sparkContext.broadcast(self._bloom_filter)
        logger.info(f"Bloom Filter built and broadcast. FPP={self.bloom_fpp}")

    def _make_udf(self):
        """Создаёт UDF, замыкающий broadcast-переменную."""
        bf_bc = self._bloom_broadcast

        @udf(returnType=BooleanType())
        def might_exist_in_history(value) -> bool:
            """True = строка ВОЗМОЖНО уже есть в истории."""
            if value is None:
                return False  # NULL-ключи всегда считаем новыми (вставляем)
            return bf_bc.value.mightContain(value)

        return might_exist_in_history

    def deduplicate(self, df_increment: DataFrame) -> DataFrame:
        """
        Выполняет трёхуровневую дедупликацию инкремента.

        Returns:
            DataFrame с гарантированно новыми строками
            (небольшой % False Positives устранён точным Anti-Join)
        """
        if self._bloom_filter is None:
            raise RuntimeError("Call build_bloom_filter() first!")

        key = self.key_col

        # ── Уровень 1: Intra-batch dedup ─────────────────────────────────────
        # Убираем дубли внутри самого инкремента.
        # При дублях берём строку с максимальным event_timestamp
        # (или любым другим tiebreaker-критерием).
        df_intra_deduped = df_increment.dropDuplicates([key])
        logger.info("Level 1: intra-batch dedup done")

        # ── Уровень 2: Bloom Filter pre-filter ───────────────────────────────
        # Быстро отсекаем строки, которых точно нет в истории.
        # Остаются: (реально новые) + (FPR% строк, уже есть в истории).
        might_exist_udf = self._make_udf()

        # might_exist_udf(key) = True  → возможно уже есть → ОТБРОСИТЬ (дубль?)
        # might_exist_udf(key) = False → точно новое    → ОСТАВИТЬ
        df_bf_filtered = df_intra_deduped.filter(
            ~might_exist_udf(F.col(key))
        )
        logger.info("Level 2: Bloom Filter pre-filter applied")

        # ── Уровень 3: Точный Anti-Join ──────────────────────────────────────
        # Устраняем False Positives. На этом этапе df_bf_filtered уже
        # в 10–20 раз меньше исходного инкремента → Shuffle минимален.
        history_keys = (
            self.spark.table(self.history_table)
            .filter(
                F.col("event_date") >= F.date_sub(F.current_date(), self.lookback_days)
            )
            .select(key)
            .dropna()
        )

        df_truly_new = df_bf_filtered.join(
            history_keys,
            on=key,
            how="left_anti",
        )
        logger.info("Level 3: exact anti-join done")

        return df_truly_new


# ── Использование ──────────────────────────────────────────────────────────

def run_incremental_ingestion(spark: SparkSession) -> None:
    """Основной ETL-пайплайн с Bloom Filter дедупликацией."""

    dedup = BloomFilterDeduplicator(
        spark=spark,
        history_table="silver.clickstream",
        key_col="event_id",
        lookback_days=90,
        bloom_fpp=0.01,
        bloom_expected_items=2_000_000_000,  # 2B строк в истории за 90 дней
    )

    # Строим фильтр один раз при старте (займёт 2–5 минут для 2B строк)
    dedup.build_bloom_filter()

    # Обрабатываем инкремент
    df_incoming = spark.table("bronze.clickstream_incoming")
    df_new = dedup.deduplicate(df_incoming)

    # Записываем в Silver
    df_new.writeTo("silver.clickstream").append()
    logger.info(f"Written {df_new.count():,} new rows to silver.clickstream")

# Проверка: насколько велик реальный FPR?
def measure_dedup_accuracy(
    df_increment: DataFrame,
    df_result: DataFrame,
    df_bf_only: DataFrame,
    key_col: str,
) -> dict:
    """
    Измеряет точность Bloom Filter фильтрации.

    Returns:
        dict с метриками: фактический FPR, эффективность pre-filter, и т.д.
    """
    n_original = df_increment.count()
    n_after_bf = df_bf_only.count()
    n_final = df_result.count()

    n_true_new = n_final           # Гарантированно новые (после точного AJ)
    n_false_positives = n_after_bf - n_final  # Строки, прошедшие BF, но отсечённые AJ
    n_filtered_by_bf = n_original - n_after_bf  # True Negatives от BF

    actual_fpr = n_false_positives / n_after_bf if n_after_bf > 0 else 0
    bf_efficiency = n_filtered_by_bf / n_original if n_original > 0 else 0

    return {
        "original_count": n_original,
        "after_bf_count": n_after_bf,
        "final_new_count": n_final,
        "false_positives": n_false_positives,
        "true_negatives_by_bf": n_filtered_by_bf,
        "actual_fpr": actual_fpr,
        "bf_pre_filter_efficiency": bf_efficiency,
    }

# Заполнение фильтра сверх нормы резко увеличивает FPR
# Формула: FPR = (1 - e^(-kn/m))^k

import math

def calculate_fpr(m_bits: int, n_elements: int, k_functions: int) -> float:
    """Вычисляет теоретический FPR Bloom Filter."""
    exponent = -k_functions * n_elements / m_bits
    fpr = (1 - math.exp(exponent)) ** k_functions
    return fpr


# Пример: фильтр рассчитан на 500M элементов с FPR=1%
m = int(500_000_000 * 9.6)  # ~4.8 миллиарда бит (~600 МБ)
k = 7  # оптимальное число хэш-функций

print("Реальный FPR при разном заполнении:")
for n_actual in [250_000_000, 500_000_000, 750_000_000, 1_000_000_000]:
    fpr = calculate_fpr(m, n_actual, k)
    overfill = n_actual / 500_000_000
    print(f"  n={n_actual/1e6:.0f}M ({overfill:.1f}x расчётного): FPR = {fpr:.3%}")

Реальный FPR при разном заполнении:
  n=250M (0.5x расчётного): FPR = 0.001% (лучше расчётного)
  n=500M (1.0x расчётного): FPR = 1.000% (соответствует)
  n=750M (1.5x расчётного): FPR = 3.681% (деградация)
  n=1000M (2.0x расчётного): FPR = 8.865% (серьёзная деградация)

# Количество False Positives при разных FPR и объёмах
# Допустим, инкремент = 10M строк, из них реально новых = 1M (10% новых)
# → В истории есть 9M строк из этих 10M

n_increment = 10_000_000
n_truly_new = 1_000_000      # 10% новых
n_existing = 9_000_000       # 90% уже есть в истории

for fpr in [0.001, 0.01, 0.05, 0.10]:
    # BF пропустит: все 1M новых (True Negatives правильно пропустит)
    # + FPR% от 9M существующих = False Positives
    false_positives = int(n_existing * fpr)
    total_after_bf = n_truly_new + false_positives

    print(f"FPR={fpr:.1%}: {false_positives:,} False Positives, "
          f"{total_after_bf:,} строк идут в Anti-Join "
          f"(из {n_increment:,} исходных)")

FPR=0.1%: 9,000 False Positives, 1,009,000 строк идут в Anti-Join (из 10,000,000)
FPR=1.0%: 90,000 False Positives, 1,090,000 строк идут в Anti-Join (из 10,000,000)
FPR=5.0%: 450,000 False Positives, 1,450,000 строк идут в Anti-Join (из 10,000,000)
FPR=10.0%: 900,000 False Positives, 1,900,000 строк идут в Anti-Join (из 10,000,000)

# Создание таблицы с Bloom Filter Index в Delta Lake
spark.sql("""
    CREATE TABLE silver.clickstream (
        event_id      STRING     NOT NULL,
        session_id    STRING,
        user_id       BIGINT,
        event_type    STRING,
        event_date    DATE,
        event_ts      TIMESTAMP,
        page_url      STRING,
        properties    MAP<STRING, STRING>
    )
    USING delta
    PARTITIONED BY (event_date)
    TBLPROPERTIES (
        -- Включаем Bloom Filter для высококардинальных колонок
        'delta.bloomFilterIndex.enabled'                     = 'true',
        'delta.bloomFilterIndexOnColumn.event_id.enabled'    = 'true',
        'delta.bloomFilterIndexOnColumn.event_id.fpp'        = '0.01',
        'delta.bloomFilterIndexOnColumn.session_id.enabled'  = 'true',
        'delta.bloomFilterIndexOnColumn.session_id.fpp'      = '0.01'
    )
""")

# Добавление Bloom Filter Index к существующей таблице Delta
spark.sql("""
    ALTER TABLE silver.clickstream
    SET TBLPROPERTIES (
        'delta.bloomFilterIndex.enabled'                     = 'true',
        'delta.bloomFilterIndexOnColumn.event_id.enabled'    = 'true',
        'delta.bloomFilterIndexOnColumn.event_id.fpp'        = '0.01'
    )
""")

# ВАЖНО: Bloom Filter Index применяется только к НОВЫМ записываемым файлам.
# Для существующих файлов нужен OPTIMIZE/REWRITE:
spark.sql("OPTIMIZE silver.clickstream ZORDER BY (event_date, user_id)")
# После OPTIMIZE новые файлы будут содержать BF Index

# Создание Iceberg таблицы с Bloom Filter
spark.sql("""
    CREATE TABLE catalog.silver.clickstream (
        event_id      STRING     NOT NULL,
        session_id    STRING,
        user_id       BIGINT,
        event_type    STRING,
        event_date    DATE,
        event_ts      TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_ts))
    TBLPROPERTIES (
        -- Bloom Filter для event_id (высокая кардинальность)
        'write.parquet.bloom-filter-enabled.column.event_id'      = 'true',
        'write.parquet.bloom-filter-fpp.column.event_id'          = '0.01',
        -- Bloom Filter для session_id
        'write.parquet.bloom-filter-enabled.column.session_id'    = 'true',
        'write.parquet.bloom-filter-fpp.column.session_id'        = '0.01',
        -- Bloom Filter для user_id
        'write.parquet.bloom-filter-enabled.column.user_id'       = 'true',
        'write.parquet.bloom-filter-fpp.column.user_id'           = '0.01'
    )
""")

# Проверка работы Bloom Filter через Data Skipping статистику
spark.sql("""
    SELECT
        operation,
        numTargetFilesAdded,
        numTargetFilesRemoved,
        operationParameters
    FROM silver.clickstream.history
    ORDER BY version DESC
    LIMIT 10
""").show(truncate=False)

# Пример запроса для анализа
df_events = spark.table("silver.clickstream")      # 2 млрд строк
df_sessions = spark.table("silver.valid_sessions") # 10 млн строк

df_joined = df_events.join(df_sessions, on="session_id", how="inner")
df_joined.explain(mode="formatted")

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastHashJoin [session_id#12], [session_id#45], Inner, BuildRight, false
   :- Filter isnotnull(session_id#12)
   :  +- Filter might_contain(bloomFilter#23, session_id#12)     ← BF применяется
   :     +- FileScan parquet silver.clickstream [...]             ← до scan
   :        PushedFilters: [IsNotNull(session_id)]
   +- BroadcastExchange HashedRelationBroadcastMode(...)
      +- BloomFilterAggregate(
             estimatedNumItems=10000000,
             numBits=94906265)                                    ← BF строится
         +- FileScan parquet silver.valid_sessions [session_id]

Stage N: Exchange (Shuffle)
  Shuffle Write:        12.5 GB   ← данные, отправленные на shuffle
  Shuffle Read:         11.8 GB   ← данные, полученные после shuffle
  Records Written:      47,832,091

Stage N: Exchange (Shuffle) - без BF
  Shuffle Write:        187.3 GB  ← ~15x больше!
  Shuffle Read:         183.1 GB
  Records Written:      743,128,432

# Форсируем применение Bloom Filter (если AQE его не выбрал автоматически)
spark.conf.set("spark.sql.runtime.filter.enabled", "true")
spark.conf.set("spark.sql.runtime.filter.bloomFilter.enabled", "true")

# Увеличиваем лимиты (для больших probe side)
spark.conf.set(
    "spark.sql.runtime.filter.bloomFilter.maxNumBits",
    str(8 * 1024 * 1024 * 1024)  # 8 ГБ в битах = 8_589_934_592
)
spark.conf.set(
    "spark.sql.runtime.filter.bloomFilter.maxNumItems",
    "100000000"  # До 100M элементов в одном BF
)

# Отключение для диагностики (сравнение с и без BF)
spark.conf.set("spark.sql.runtime.filter.bloomFilter.enabled", "false")
df_no_bf = df_large.join(df_small, on="key", how="inner")
df_no_bf.explain(mode="formatted")  # Увидим обычный SortMergeJoin или BHJ

# Увеличить порог для broadcast BF объекта
spark.conf.set("spark.sql.runtime.filter.bloomFilter.maxNumBits", str(4 * 1024**3 * 8))  # 4 ГБ

# Или построить частичный BF (только по последним N дням)
# вместо всей истории
bf_partial = (
    spark.table("silver.events")
    .filter(F.col("event_date") >= F.date_sub(F.current_date(), 30))  # 30 дней вместо 90
    .stat.bloomFilter("event_id", expectedNumItems=500_000_000, fpp=0.01)
)

# Benchmark: время BF проверки vs время Anti-Join
import time

# BF pre-filter (только вычисление хэшей, без Shuffle)
start = time.time()
count_after_bf = df_increment.filter(~might_exist_udf(F.col("event_id"))).count()
bf_time = time.time() - start

# Полный Anti-Join (с Shuffle)
start = time.time()
count_exact = df_increment.join(history_keys, on="event_id", how="left_anti").count()
exact_time = time.time() - start

print(f"BF pre-filter time:    {bf_time:.1f}s (→ {count_after_bf:,} строк)")
print(f"Exact Anti-Join time:  {exact_time:.1f}s (→ {count_exact:,} строк)")
print(f"Speedup (BF vs exact): {exact_time / bf_time:.1f}x")

# ПЛОХО: BF на колонке с 50 уникальными значениями (низкая кардинальность)
# department_id: ["HR", "IT", "Finance", ..., "Legal"] - 50 значений
# BF из 50 значений: практически всегда говорит "есть"
# → FPR близок к 100% → BF ничего не отсекает
bad_bf = df_small.stat.bloomFilter("department_id", expectedNumItems=50, fpp=0.01)
# Фильтр размером ~1 КБ, но с крошечным эффектом

# ХОРОШО: BF на высококардинальной колонке
good_bf = df_small.stat.bloomFilter("user_id", expectedNumItems=50_000_000, fpp=0.01)
# Фильтр размером ~60 МБ, эффективно отсекает 95%+ строк

# Несвежий фильтр: пропускает больше False Positives
# Решение: пересоздавать фильтр при каждом запуске пайплайна
# или хранить TTL и проверять его

from datetime import datetime, timedelta

class CachedBloomFilter:
    """BF с временем жизни (TTL)."""

    def __init__(self, ttl_minutes: int = 60):
        self._filter = None
        self._built_at = None
        self._ttl = timedelta(minutes=ttl_minutes)

    def is_fresh(self) -> bool:
        if self._filter is None or self._built_at is None:
            return False
        return datetime.now() - self._built_at < self._ttl

    def get_or_build(self, spark, table: str, key_col: str) -> any:
        if not self.is_fresh():
            self._filter = _build_filter(spark, table, key_col)
            self._built_at = datetime.now()
        return self._filter

# ОПАСНО: указали 10M, реальных элементов 500M
# → FPR деградирует с 1% до ~30%+
bad_bf = df_large.stat.bloomFilter(
    "event_id",
    expectedNumItems=10_000_000,  # 10M
    fpp=0.01
)
# При реальных 500M элементов: FPR ≈ 30–40% (практически бесполезен)

# ПРАВИЛЬНО: с запасом 30%
actual_n = df_large.select("event_id").distinct().count()  # Считаем точно
good_bf = df_large.stat.bloomFilter(
    "event_id",
    expectedNumItems=int(actual_n * 1.3),
    fpp=0.01
)

# ХОРОШО: маленькая таблица помещается в broadcast
small_table_size_mb = 50  # МБ

if small_table_size_mb < 100:
    # Broadcast Hash Join - точный, без FPR, без overhead построения BF
    result = df_large.join(F.broadcast(df_small), on="key", how="inner")
else:
    # Bloom Filter оправдан
    bf = df_small.stat.bloomFilter("key", expectedNumItems=10_000_000, fpp=0.01)
    # ... применяем BF

"""
clickstream_dedup_pipeline.py
Production-кейс: Bloom Filter ускорение дедупликации clickstream
200M строк/час, 50B строк история
"""

from pyspark.sql import SparkSession, functions as F, DataFrame
from datetime import datetime
import logging
import time

logger = logging.getLogger(__name__)

# ── Конфигурация ──────────────────────────────────────────────────────────────

BRONZE_TABLE   = "catalog.bronze.clickstream_incoming"
SILVER_TABLE   = "catalog.silver.clickstream"
KEY_COL        = "event_id"
DATE_COL       = "event_date"
LOOKBACK_DAYS  = 90   # Насколько далеко смотреть в историю для Anti-Join
BLOOM_FPP      = 0.01  # 1% FPR для file-level BF Index


def configure_spark(spark: SparkSession) -> SparkSession:
    """
    Настраиваем Spark для максимальной эффективности с BF.

    Ключевые настройки:
    - AQE: адаптивное выполнение (перепланирование после каждого stage)
    - Runtime Filter: автоматическое применение BF в JOIN
    - Shuffle partitions: ~400 для 50 ГБ shuffle при 128 МБ/партицию
    """
    configs = {
        # AQE - критично для Runtime Bloom Filter
        "spark.sql.adaptive.enabled":                    "true",
        "spark.sql.adaptive.coalescePartitions.enabled": "true",
        "spark.sql.adaptive.skewJoin.enabled":           "true",

        # Runtime Bloom Filter
        "spark.sql.runtime.filter.enabled":                      "true",
        "spark.sql.runtime.filter.bloomFilter.enabled":          "true",
        "spark.sql.runtime.filter.bloomFilter.maxNumBits":       str(8 * 1024**3 * 8),
        "spark.sql.runtime.filter.bloomFilter.maxNumItems":      "100000000",

        # Shuffle
        "spark.sql.shuffle.partitions": "400",

        # Broadcast threshold
        "spark.sql.autoBroadcastJoinThreshold": "104857600",  # 100 МБ
    }

    for k, v in configs.items():
        spark.conf.set(k, v)

    return spark


# ── Stage 1: Нормализация и intra-batch dedup ─────────────────────────────────

def stage1_normalize_and_intra_dedup(spark: SparkSession) -> DataFrame:
    """
    Читает бронзовый батч, нормализует и дедуплицирует внутри батча.

    Intra-batch dedup выполняется через dropDuplicates по event_id.
    При дублях внутри батча (один источник прислал дважды) берём
    строку с максимальным event_ts.
    """
    t = time.time()

    df_raw = spark.table(BRONZE_TABLE)

    # Нормализация
    df_norm = (
        df_raw
        .withColumn(KEY_COL, F.trim(F.col(KEY_COL)))           # Убираем пробелы
        .withColumn("event_ts", F.to_utc_timestamp(            # Приводим к UTC
            F.col("event_ts"), "UTC"
        ))
        .withColumn(DATE_COL, F.to_date(F.col("event_ts")))    # Дата из timestamp
        .filter(F.col(KEY_COL).isNotNull())                    # Убираем NULL ключи
    )

    # Intra-batch dedup: берём запись с максимальным event_ts при дублях
    df_deduped = df_norm.dropDuplicates([KEY_COL])

    # Кэшируем - будем использовать несколько раз
    df_deduped.cache()
    count = df_deduped.count()

    logger.info(f"Stage 1: {count:,} unique rows after intra-batch dedup "
                f"({time.time() - t:.1f}s)")
    return df_deduped


# ── Stage 2 + 3: BF-assisted Anti-Join через Iceberg file-level index ─────────

def stage2_bloom_filter_anti_join(
    spark: SparkSession,
    df_batch: DataFrame,
) -> DataFrame:
    """
    Использует Iceberg Bloom Filter Index для эффективного Anti-Join.

    Вместо явного in-memory BF объекта полагаемся на file-level BF Index
    в Silver таблице (Iceberg). Spark автоматически применяет его при
    выполнении JOIN/Anti-Join, пропуская файлы, в которых точно нет ключей.

    Дополнительно: Runtime Bloom Filter от AQE строит in-memory BF
    по ключам батча и применяет к Silver при сканировании.
    """
    t = time.time()

    # Создаём временное представление батча
    df_batch.createOrReplaceTempView("batch_events")

    # Anti-Join: Catalyst + AQE + Runtime BF + Iceberg file-level BF
    # работают совместно:
    # 1. AQE строит Runtime BF по batch_events.event_id
    # 2. BF транслируется на экзекьюторы, читающие Silver
    # 3. Iceberg file-level BF дополнительно пропускает файлы
    # 4. Из оставшихся файлов читаем только нужные строки
    # 5. Anti-Join завершает точную фильтрацию

    silver_keys = (
        spark.table(SILVER_TABLE)
        .filter(
            # Ограничиваем lookback window - не смотрим в "вечность"
            F.col(DATE_COL) >= F.date_sub(F.current_date(), LOOKBACK_DAYS)
        )
        .select(KEY_COL)
    )

    df_new = df_batch.join(silver_keys, on=KEY_COL, how="left_anti")

    # Не вызываем count() здесь - это Action, откладываем до write
    logger.info(f"Stage 2+3: Anti-Join plan built ({time.time() - t:.1f}s)")
    return df_new


# ── Stage 4: Write ─────────────────────────────────────────────────────────────

def stage4_write_to_silver(df_new: DataFrame) -> None:
    """
    Записывает новые строки в Silver (Iceberg).

    При записи Iceberg автоматически строит BF Index для event_id
    (если TBLPROPERTIES настроены) - файлы сразу доступны для
    эффективного будущего Anti-Join.
    """
    t = time.time()

    # Записываем с партиционированием по дате
    df_new.writeTo(SILVER_TABLE).append()

    logger.info(f"Stage 4: Write complete ({time.time() - t:.1f}s)")


# ── Мониторинг ────────────────────────────────────────────────────────────────

def collect_metrics(spark: SparkSession, df_batch: DataFrame, df_new: DataFrame) -> dict:
    """
    Собирает метрики для мониторинга качества дедупликации.

    ВАЖНО: count() - это Action, выполняется отдельными задачами.
    В production используйте Spark Listener или metrics API
    для сбора без дополнительных Action.
    """
    batch_count = df_batch.count()
    new_count = df_new.count()
    existing_count = batch_count - new_count

    # Читаем статистику последнего commit из Iceberg metadata
    iceberg_stats = spark.sql(f"""
        SELECT
            added_data_files_count,
            added_records_count,
            total_data_files_count,
            total_records_count
        FROM {SILVER_TABLE}.snapshots
        ORDER BY committed_at DESC
        LIMIT 1
    """).collect()[0]

    return {
        "batch_size":          batch_count,
        "new_events":          new_count,
        "duplicate_events":    existing_count,
        "duplicate_rate":      f"{existing_count / batch_count:.1%}",
        "silver_total_files":  iceberg_stats["total_data_files_count"],
        "silver_total_rows":   iceberg_stats["total_records_count"],
    }


# ── Main Pipeline ──────────────────────────────────────────────────────────────

def main():
    spark = (
        SparkSession.builder
        .appName("clickstream_dedup_bloom_filter")
        .getOrCreate()
    )
    configure_spark(spark)

    pipeline_start = time.time()
    logger.info("=== Clickstream Dedup Pipeline START ===")

    # Stage 1: Normalize & intra-batch dedup
    df_batch = stage1_normalize_and_intra_dedup(spark)

    # Stage 2+3: BF-assisted Anti-Join
    df_new = stage2_bloom_filter_anti_join(spark, df_batch)

    # Stage 4: Write to Silver (triggers lazy evaluation)
    stage4_write_to_silver(df_new)

    # Collect & log metrics
    metrics = collect_metrics(spark, df_batch, df_new)
    for k, v in metrics.items():
        logger.info(f"  {k}: {v}")

    total_time = time.time() - pipeline_start
    logger.info(f"=== Pipeline DONE in {total_time/60:.1f} min ===")


if __name__ == "__main__":
    main()

# Создаём Silver таблицу с полными оптимизациями
spark.sql(f"""
    CREATE TABLE {SILVER_TABLE} (
        event_id     STRING        NOT NULL COMMENT 'Уникальный ID события',
        session_id   STRING        COMMENT 'ID сессии',
        user_id      BIGINT        COMMENT 'ID пользователя',
        event_type   STRING        COMMENT 'Тип события (click/view/purchase)',
        event_date   DATE          COMMENT 'Дата события (партиция)',
        event_ts     TIMESTAMP     COMMENT 'Время события в UTC',
        page_url     STRING        COMMENT 'URL страницы',
        device_type  STRING        COMMENT 'desktop/mobile/tablet',
        country_code STRING        COMMENT 'ISO код страны',
        _ingested_at TIMESTAMP     COMMENT 'Время ingestion',
        _batch_id    STRING        COMMENT 'ID батча для трейсабилити'
    )
    USING iceberg
    PARTITIONED BY (event_date)
    TBLPROPERTIES (
        -- Bloom Filter для высококардинальных ключей
        'write.parquet.bloom-filter-enabled.column.event_id'    = 'true',
        'write.parquet.bloom-filter-fpp.column.event_id'        = '0.01',
        'write.parquet.bloom-filter-enabled.column.session_id'  = 'true',
        'write.parquet.bloom-filter-fpp.column.session_id'      = '0.01',
        'write.parquet.bloom-filter-enabled.column.user_id'     = 'true',
        'write.parquet.bloom-filter-fpp.column.user_id'         = '0.01',

        -- Целевой размер файла: 512 МБ (баланс IO и кол-ва файлов)
        'write.target-file-size-bytes'  = '536870912',

        -- Автоматический compaction через OPTIMIZE
        'write.spark.fanout.enabled'    = 'true'
    )
""")