Anti-Join dedup (left_anti): загрузка инкремента без дублей - паттерн для batch ETL

Батч 1: события 10:00 - 12:00
Батч 2: события 11:00 - 13:00   ← час пересечения 11:00-12:00

-- LEFT ANTI JOIN (PySpark/Spark SQL)
SELECT inc.*
FROM incoming_events inc
LEFT ANTI JOIN target_events tgt
    ON inc.event_id = tgt.event_id;

-- Эквивалент через NOT EXISTS
SELECT inc.*
FROM incoming_events inc
WHERE NOT EXISTS (
    SELECT 1 FROM target_events tgt
    WHERE tgt.event_id = inc.event_id
);

-- Эквивалент через LEFT JOIN + IS NULL
SELECT inc.*
FROM incoming_events inc
LEFT JOIN target_events tgt
    ON inc.event_id = tgt.event_id
WHERE tgt.event_id IS NULL;

-- ПЛОХО: NOT IN с подзапросом
SELECT inc.*
FROM incoming_events inc
WHERE inc.event_id NOT IN (SELECT event_id FROM target_events);
-- Проблема: если в target есть NULL event_id - результат пустой!
-- SQL: "x NOT IN (NULL, ...)" всегда FALSE при наличии NULL в списке

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder \
    .appName("AntiJoin-Deep-Dive") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "hadoop") \
    .config("spark.sql.catalog.lakehouse.warehouse", "/tmp/warehouse") \
    .getOrCreate()

# Тестовые данные
incoming = spark.createDataFrame([
    ("evt-001", "user_1", "click"),
    ("evt-002", "user_2", "buy"),
    ("evt-003", "user_1", "view"),
], ["event_id", "user_id", "event_type"])

target = spark.createDataFrame([
    ("evt-001", "user_1", "click"),
    ("evt-003", "user_1", "view"),
], ["event_id", "user_id", "event_type"])

# Способ 1: left_anti (рекомендуемый)
anti_join = incoming.join(target, on="event_id", how="left_anti")
print("=== left_anti EXPLAIN ===")
anti_join.explain(mode="formatted")

# Способ 2: NOT EXISTS (эквивалент)
incoming.createOrReplaceTempView("incoming")
target.createOrReplaceTempView("target")
not_exists = spark.sql("""
    SELECT * FROM incoming i
    WHERE NOT EXISTS (SELECT 1 FROM target t WHERE t.event_id = i.event_id)
""")
print("=== NOT EXISTS EXPLAIN ===")
not_exists.explain(mode="formatted")

== Physical Plan ==
* BroadcastHashJoin [event_id], [event_id], LeftAnti   ← Anti семантика на уровне join
  :- * Scan: incoming
  +- BroadcastExchange                                  ← target broadcast в память
     +- * Scan: target

# Имитируем входящий батч с дублями внутри
raw_increment = spark.createDataFrame([
    ("evt-010", "order_2001", "created",  1500.0, "2026-05-23 10:00:00"),
    ("evt-010", "order_2001", "created",  1500.0, "2026-05-23 10:00:00"),  # дубль!
    ("evt-011", "order_2002", "pending",   750.0, "2026-05-23 10:05:00"),
    ("evt-012", "order_2003", "confirmed", 200.0, "2026-05-23 10:10:00"),
    # evt-001 уже есть в target - anti-join должен его отсечь
    ("evt-001", "order_1001", "confirmed",1500.0, "2026-05-23 09:00:00"),
], ["event_id", "order_id", "status", "amount", "event_ts"])

# ШАГ 1: Дедупликация самого инкремента
# Без этого evt-010 дважды попадёт в anti-join и оба пройдут фильтр
clean_increment = raw_increment.dropDuplicates(["event_id"])

print(f"Инкремент до dedup: {raw_increment.count()} строк")
print(f"Инкремент после dedup: {clean_increment.count()} строк")
# До: 5, после: 4 (убрали дубль evt-010)

# Создаём target таблицу (имитация существующей Silver)
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.silver.order_events (
        event_id   STRING,
        order_id   STRING,
        status     STRING,
        amount     DECIMAL(10,2),
        event_ts   TIMESTAMP,
        event_date DATE
    )
    USING iceberg
    PARTITIONED BY (event_date)
""")

# Имитируем существующие данные в Silver
existing_data = spark.createDataFrame([
    ("evt-001", "order_1001", "confirmed", 1500.0, "2026-05-23 09:00:00", "2026-05-23"),
    ("evt-002", "order_1001", "created",   1500.0, "2026-05-22 08:00:00", "2026-05-22"),
], ["event_id", "order_id", "status", "amount", "event_ts", "event_date"])

existing_data \
    .withColumn("event_ts", F.to_timestamp("event_ts")) \
    .withColumn("event_date", F.to_date("event_date")) \
    .withColumn("amount", F.col("amount").cast("decimal(10,2)")) \
    .writeTo("lakehouse.silver.order_events") \
    .append()

# ОПТИМИЗАЦИЯ: читаем target только за последние 7 дней
# Spark применяет Partition Pruning - читает только нужные партиции
LOOKBACK_DAYS = 7

target_window = (
    spark.table("lakehouse.silver.order_events")
    .filter(
        F.col("event_date") >= F.date_sub(F.current_date(), LOOKBACK_DAYS)
    )
    .select("event_id")  # берём только ключ - минимальный объём данных
)

print(f"Строк в target window ({LOOKBACK_DAYS} дней): {target_window.count()}")

# ШАГ 3: Anti-join - оставляем только действительно новые события
new_events = (
    clean_increment                        # левая: дедуплицированный инкремент
    .join(
        target_window,                     # правая: только ключи из target-окна
        on="event_id",
        how="left_anti"                    # строки без совпадения в правой
    )
)

print(f"\nАнти-джойн результат:")
new_events.show()
# event_id | order_id   | status    | amount | event_ts
# evt-010  | order_2001 | created   | 1500.0 | ...
# evt-011  | order_2002 | pending   | 750.0  | ...
# evt-012  | order_2003 | confirmed | 200.0  | ...
# evt-001 отсечён: он уже есть в target

print(f"Новых событий для записи: {new_events.count()}")
# 3 из 4 (evt-001 отсечён anti-join)

# ШАГ 4: Безопасный APPEND только новых строк
(
    new_events
    .withColumn("event_ts",   F.to_timestamp("event_ts"))
    .withColumn("amount",     F.col("amount").cast("decimal(10,2)"))
    .withColumn("event_date", F.to_date("event_ts"))
    .writeTo("lakehouse.silver.order_events")
    .append()
)

print("Данные записаны в Silver через безопасный APPEND")
print(f"Итого в Silver: {spark.table('lakehouse.silver.order_events').count()} строк")
# 5 строк: 2 существующих + 3 новых

-- Эквивалентный запрос на Spark SQL (более читаемый для SQL-разработчиков)
INSERT INTO lakehouse.silver.order_events
SELECT
    inc.event_id,
    inc.order_id,
    inc.status,
    CAST(inc.amount AS DECIMAL(10,2)),
    CAST(inc.event_ts AS TIMESTAMP),
    TO_DATE(inc.event_ts)  AS event_date
FROM clean_increment_view AS inc
WHERE NOT EXISTS (
    SELECT 1
    FROM lakehouse.silver.order_events AS tgt
    WHERE tgt.event_id   = inc.event_id
      AND tgt.event_date >= CURRENT_DATE() - INTERVAL 7 DAYS
);

# Случай 2: Broadcast target-ключи (маленькая правая сторона)
# Spark автоматически broadcast-ит таблицы < spark.sql.autoBroadcastJoinThreshold (10MB)
# Но если target_window.count() < 100k строк - лучше указать явно:

new_events_broadcast = (
    clean_increment
    .join(
        F.broadcast(target_window),   # явный broadcast hint
        on="event_id",
        how="left_anti"
    )
)

new_events_broadcast.explain(mode="formatted")
# Physical Plan:
# * BroadcastHashJoin [event_id], [event_id], LeftAnti
#   :- * Scan: clean_increment  (без Shuffle!)
#   +- BroadcastExchange       (target распределяется в память каждого executor)
#      +- * Scan: target_window

# Случай 1: Broadcast инкремент (маленькая левая сторона)
# Не всегда возможен для left_anti: broadcast в left_anti оптимально
# применяется к ПРАВОЙ таблице
# Если инкремент мал - рассмотрите перестановку: target ANTI JOIN increment
small_increment_anti = (
    target_window
    .join(
        F.broadcast(clean_increment.select("event_id")),
        on="event_id",
        how="left_anti"           # НЕ то же самое! Это строки target без совпадений
    )
    # Это anti-join другой направленности - осторожно!
)

# Неоптимально: Spark читает ВСЕ партиции target
bad_anti = (
    clean_increment
    .join(
        spark.table("lakehouse.silver.order_events"),  # вся таблица!
        on="event_id",
        how="left_anti"
    )
)

# Оптимально: Spark читает только партиции за последние 7 дней
# Static Partition Pruning: фильтр применяется ДО join-а
good_anti = (
    clean_increment
    .join(
        spark.table("lakehouse.silver.order_events")
             .filter(F.col("event_date") >= F.date_sub(F.current_date(), 7)),
        on="event_id",
        how="left_anti"
    )
)

# Spark SQL вариант с явным фильтром в подзапросе
spark.sql("""
    SELECT inc.*
    FROM clean_increment inc
    LEFT ANTI JOIN (
        SELECT event_id
        FROM lakehouse.silver.order_events
        WHERE event_date >= CURRENT_DATE() - INTERVAL 7 DAYS  -- Partition Pruning!
    ) tgt
    ON inc.event_id = tgt.event_id
""")

# Для больших объёмов: настраиваем shuffle
spark.conf.set("spark.sql.shuffle.partitions", "400")  # больше партиций = меньше размер каждой
spark.conf.set("spark.sql.adaptive.enabled", "true")   # AQE автоматически подбирает оптимум

# AQE (Adaptive Query Execution) автоматически:
# 1. Уменьшает shuffle.partitions если данных меньше ожидаемого
# 2. Переключает SortMergeJoin → BroadcastHashJoin если одна сторона оказалась малой
# 3. Обрабатывает Data Skew через split-операции

large_increment = spark.table("lakehouse.bronze.orders_large")  # 100M строк
large_target    = spark.table("lakehouse.silver.orders_large")   # 1B строк

# AQE определит правильную стратегию автоматически
new_only = (
    large_increment
    .join(
        large_target
        .filter(F.col("event_date") >= F.date_sub(F.current_date(), 30))
        .select("event_id"),
        on="event_id",
        how="left_anti"
    )
)

# Включаем Bloom Filter для колонки event_id при записи
spark.sql("""
    ALTER TABLE lakehouse.silver.order_events
    SET TBLPROPERTIES (
        'write.metadata.metrics.column.event_id' = 'full',
        'write.parquet.bloom-filter-enabled.column.event_id' = 'true',
        'write.parquet.bloom-filter-false-positive-probability.column.event_id' = '0.01'
    )
""")
# Теперь при anti-join Iceberg будет пропускать data files,
# в которых точно нет искомых event_id - без чтения самих файлов

# ЛОВУШКА: NULL-значения в join-ключе
events_with_null = spark.createDataFrame([
    ("evt-001", "order_1001"),
    (None,      "order_1002"),  # NULL event_id!
    ("evt-003", "order_1003"),
], ["event_id", "order_id"])

target_keys = spark.createDataFrame([
    ("evt-001",),
    ("evt-003",),
], ["event_id"])

# LEFT ANTI JOIN с NULL ключом в левой таблице
result = events_with_null.join(target_keys, on="event_id", how="left_anti")
result.show()
# event_id | order_id
# null     | order_1002   ← NULL-строка ПРОШЛА anti-join!
# (evt-001 и evt-003 отсечены корректно)

# В SQL: NULL != NULL → совпадение не найдено → строка попадает в ANTI результат
# Это может быть нежелательным поведением!

# РЕШЕНИЕ: фильтруем NULL-ключи перед anti-join
clean_for_anti = events_with_null.filter(F.col("event_id").isNotNull())
result_clean = clean_for_anti.join(target_keys, on="event_id", how="left_anti")
result_clean.show()
# Пустой результат - оба не-NULL ключа найдены в target

# Некоторые системы не имеют глобального UUID - ключ составной
# Пример: события уникальны по (order_id, event_type, event_ts)

increment_comp = spark.createDataFrame([
    ("order_1001", "created",  "2026-05-23 10:00:00", 1500.0),
    ("order_1001", "paid",     "2026-05-23 10:05:00", 1500.0),
    ("order_1002", "created",  "2026-05-23 10:10:00",  750.0),
], ["order_id", "event_type", "event_ts", "amount"])

target_comp = spark.table("lakehouse.silver.order_events") \
    .filter(F.col("event_date") >= F.date_sub(F.current_date(), 7)) \
    .select("order_id", "event_type", "event_ts")

# Anti-join по составному ключу
new_comp = (
    increment_comp
    .join(
        target_comp,
        on=["order_id", "event_type", "event_ts"],  # составной ключ
        how="left_anti"
    )
)

# Spark SQL эквивалент:
spark.sql("""
    SELECT inc.*
    FROM increment_comp inc
    LEFT ANTI JOIN target_comp tgt
    ON  inc.order_id   = tgt.order_id
    AND inc.event_type = tgt.event_type
    AND inc.event_ts   = tgt.event_ts
""")

# Пример: LOOKBACK_DAYS = 7
# evt-OLD было загружено 10 дней назад (за пределами lookback window)
# Теперь источник снова прислал evt-OLD (например, при переотправке истории)

LOOKBACK_DAYS = 7
target_window = (
    spark.table("lakehouse.silver.order_events")
    .filter(F.col("event_date") >= F.date_sub(F.current_date(), LOOKBACK_DAYS))
    .select("event_id")
)

# evt-OLD (10 дней назад) НЕ попадёт в target_window
# Anti-join не найдёт его в правой таблице
# evt-OLD ПРОЙДЁТ как «новое» и будет записан повторно!

# Решения:
# 1. Увеличить LOOKBACK_DAYS (но растёт стоимость join-а)
# 2. Хранить отдельную "seen events" таблицу с только ключами (экономично)
# 3. Использовать MERGE INTO вместо anti-join для таких случаев

# Решение 2: отдельная компактная таблица ключей
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.silver.order_events_keys (
        event_id STRING,
        loaded_date DATE
    )
    USING iceberg
    PARTITIONED BY (loaded_date)
    TBLPROPERTIES ('write.metadata.metrics.column.event_id' = 'full')
""")

# Ключевая таблица хранит ТОЛЬКО event_id - минимальный размер
# Lookback по loaded_date, не по event_ts - всегда актуально

# АНТИПАТТЕРН: anti-join без предварительной дедупликации инкремента
raw_with_dups = spark.createDataFrame([
    ("evt-NEW", "order_3001", "created"),
    ("evt-NEW", "order_3001", "created"),  # дубль в инкременте!
], ["event_id", "order_id", "status"])

# Anti-join не сможет помочь: оба evt-NEW отсутствуют в target
# оба пройдут фильтр и оба будут записаны!
bad_result = raw_with_dups.join(target_keys, on="event_id", how="left_anti")
print(f"Строк пройдёт: {bad_result.count()}")  # 2 - оба дубля!

# ПРАВИЛЬНО: сначала дедуплицировать инкремент
clean = raw_with_dups.dropDuplicates(["event_id"])
good_result = clean.join(target_keys, on="event_id", how="left_anti")
print(f"Строк пройдёт: {good_result.count()}")  # 1

# АНТИПАТТЕРН: использование anti-join для обновляемых сущностей

# Заказ order_1001 уже в Silver со статусом "created"
# Новый батч содержит тот же order_id со статусом "confirmed"

increment_update = spark.createDataFrame([
    ("evt-001-v2", "order_1001", "confirmed", 1500.0),  # обновлённая версия
], ["event_id", "order_id", "status", "amount"])

# Anti-join по event_id: evt-001-v2 НЕТ в target → пройдёт как «новое»
# В Silver появится ДВЕ строки для order_1001: created и confirmed
# Это ПРАВИЛЬНО если Silver хранит историю событий (event sourcing)
# Но НЕВЕРНО если Silver хранит текущее состояние заказа!

# Для mutable сущностей (current state) → MERGE INTO, не anti-join

# АНТИПАТТЕРН: использование event_ts как ключа anti-join
# Разные источники могут присылать одно событие с разным timestamp!
source_a = ("order_1001", "created", "2026-05-23 10:00:00.000")
source_b = ("order_1001", "created", "2026-05-23 10:00:00.001")  # 1 мс разницы!

# Anti-join по (order_id, event_ts): не найдёт совпадение из-за timestamp разницы
# Оба события вставятся как «новые» - дубль по бизнес-смыслу

# ПРАВИЛЬНО: использовать стабильный бизнес-ключ (order_id + event_type)
# или UUID, если источник его предоставляет

import datetime
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder \
    .appName("AntiJoin-ETL-E2E") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "hadoop") \
    .config("spark.sql.catalog.lakehouse.warehouse", "/tmp/warehouse") \
    .getOrCreate()

spark.conf.set("spark.sql.adaptive.enabled", "true")

# ─── ПАРАМЕТРЫ БАТЧА ─────────────────────────────────────────────────────────
BATCH_TS       = "2026-05-23 11:00:00"  # эталонное время запуска (не current_timestamp!)
LOOKBACK_DAYS  = 7                       # окно поиска в target
TARGET_TABLE   = "lakehouse.silver.order_events"

# ─── ВХОДЯЩИЙ БАТЧ ИЗ API ────────────────────────────────────────────────────
# API вернул данные за 10:00–11:00. Часть из них уже загружена предыдущим батчем!
raw_api_response = spark.createDataFrame([
    # Технический дубль: evt-001 загружен прошлым батчем
    ("evt-001", "order_1001", "created",    1500.0, "2026-05-23 10:00:00"),
    # Технический дубль внутри батча
    ("evt-010", "order_2001", "created",     500.0, "2026-05-23 10:15:00"),
    ("evt-010", "order_2001", "created",     500.0, "2026-05-23 10:15:00"),
    # Действительно новые события
    ("evt-011", "order_2002", "confirmed",  2000.0, "2026-05-23 10:30:00"),
    ("evt-012", "order_2003", "pending",     750.0, "2026-05-23 10:45:00"),
    ("evt-013", "order_2004", "created",    1200.0, "2026-05-23 10:55:00"),
], ["event_id", "order_id", "status", "amount", "event_ts"])

print(f"Событий из API: {raw_api_response.count()}")  # 6

# ─── ШАГ 1: ИНИЦИАЛИЗАЦИЯ TARGET ─────────────────────────────────────────────
spark.sql(f"""
    CREATE TABLE IF NOT EXISTS {TARGET_TABLE} (
        event_id   STRING,
        order_id   STRING,
        status     STRING,
        amount     DECIMAL(10,2),
        event_ts   TIMESTAMP,
        event_date DATE
    )
    USING iceberg
    PARTITIONED BY (event_date)
""")

# Имитируем существующие данные (загружены прошлым батчем)
existing = spark.createDataFrame([
    ("evt-001", "order_1001", "created", 1500.0, "2026-05-23 10:00:00", "2026-05-23"),
], ["event_id", "order_id", "status", "amount", "event_ts", "event_date"])

existing \
    .withColumn("event_ts", F.to_timestamp("event_ts")) \
    .withColumn("amount", F.col("amount").cast("decimal(10,2)")) \
    .withColumn("event_date", F.to_date("event_date")) \
    .writeTo(TARGET_TABLE).append()

print(f"Существующих строк в Silver: {spark.table(TARGET_TABLE).count()}")  # 1

# ─── ШАГ 2: ДЕДУПЛИКАЦИЯ ИНКРЕМЕНТА ──────────────────────────────────────────
clean_increment = (
    raw_api_response
    .filter(F.col("event_id").isNotNull())    # убираем NULL-ключи
    .dropDuplicates(["event_id"])             # технические дубли внутри батча
)

print(f"После дедупликации инкремента: {clean_increment.count()} строк")  # 5

# ─── ШАГ 3: LOADING TARGET WINDOW ────────────────────────────────────────────
target_keys = (
    spark.table(TARGET_TABLE)
    .filter(F.col("event_date") >= F.date_sub(F.current_date(), LOOKBACK_DAYS))
    .select("event_id")             # только ключ - минимальный объём данных
)

print(f"Ключей в target window ({LOOKBACK_DAYS} дней): {target_keys.count()}")  # 1

# ─── ШАГ 4: ANTI-JOIN ────────────────────────────────────────────────────────
new_events = (
    clean_increment
    .join(
        F.broadcast(target_keys),   # target_keys мал - broadcast
        on="event_id",
        how="left_anti"
    )
)

print(f"Новых событий для вставки: {new_events.count()}")  # 4
new_events.select("event_id", "order_id", "status").show()
# evt-010 | order_2001 | created     (прошёл: нет в target)
# evt-011 | order_2002 | confirmed   (прошёл: нет в target)
# evt-012 | order_2003 | pending     (прошёл: нет в target)
# evt-013 | order_2004 | created     (прошёл: нет в target)
# evt-001 ОТСЕЧЁН: уже есть в target

# ─── ШАГ 5: БЕЗОПАСНЫЙ APPEND ────────────────────────────────────────────────
(
    new_events
    .withColumn("event_ts",   F.to_timestamp("event_ts"))
    .withColumn("amount",     F.col("amount").cast("decimal(10,2)"))
    .withColumn("event_date", F.to_date("event_ts"))
    .writeTo(TARGET_TABLE)
    .append()
)

print(f"\nИтого в Silver: {spark.table(TARGET_TABLE).count()} строк")  # 5
spark.table(TARGET_TABLE).orderBy("event_ts").show()

# ─── ПРОВЕРКА ИДЕМПОТЕНТНОСТИ ─────────────────────────────────────────────────
print("\n=== СИМУЛЯЦИЯ AIRFLOW RETRY (те же входные данные) ===")

# Повторяем шаги 2-5 с теми же raw_api_response
clean_increment_retry = (
    raw_api_response
    .filter(F.col("event_id").isNotNull())
    .dropDuplicates(["event_id"])
)

target_keys_retry = (
    spark.table(TARGET_TABLE)
    .filter(F.col("event_date") >= F.date_sub(F.current_date(), LOOKBACK_DAYS))
    .select("event_id")
)

new_events_retry = (
    clean_increment_retry
    .join(F.broadcast(target_keys_retry), on="event_id", how="left_anti")
)

print(f"Новых событий при retry: {new_events_retry.count()}")  # 0 - ВСЕ УЖЕ ЕСТЬ!

# APPEND 0 строк - таблица не изменилась
if new_events_retry.count() > 0:
    new_events_retry \
        .withColumn("event_ts", F.to_timestamp("event_ts")) \
        .withColumn("amount", F.col("amount").cast("decimal(10,2)")) \
        .withColumn("event_date", F.to_date("event_ts")) \
        .writeTo(TARGET_TABLE).append()

print(f"Silver после retry: {spark.table(TARGET_TABLE).count()} строк")  # всё те же 5!
print("Идемпотентность подтверждена: retry не создал дублей ✓")

Событий из API: 6
Существующих строк в Silver: 1

После дедупликации инкремента: 5 строк
Ключей в target window (7 дней): 1

Новых событий для вставки: 4
event_id | order_id   | status
evt-010  | order_2001 | created
evt-011  | order_2002 | confirmed
evt-012  | order_2003 | pending
evt-013  | order_2004 | created

Итого в Silver: 5 строк

=== СИМУЛЯЦИЯ AIRFLOW RETRY ===
Новых событий при retry: 0
Silver после retry: 5 строк
Идемпотентность подтверждена: retry не создал дублей ✓