Что такое LEFT ANTI JOIN в PySpark?

LEFT ANTI JOIN в PySpark возвращает только те строки из левой таблицы (DataFrame), для которых нет совпадений в правой таблице по условию соединения. Это эффективный способ фильтрации данных, аналогичный операции "NOT IN" или "NOT EXISTS" в SQL.

Синтаксис и пример:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Создание сессии Spark
spark = SparkSession.builder 
    .appName("LeftAntiJoinExample") 
    .getOrCreate()

# Создание тестовых DataFrames
orders = spark.createDataFrame([
    (1, "2024-01-01", 100.0),
    (2, "2024-01-02", 200.0),
    (3, "2024-01-03", 150.0),
    (4, "2024-01-04", 300.0)
], ["order_id", "order_date", "amount"])

cancelled_orders = spark.createDataFrame([
    (2, "client_request"),
    (4, "out_of_stock")
], ["order_id", "cancel_reason"])

# LEFT ANTI JOIN: найти НЕ отмененные заказы
active_orders = orders.join(
    cancelled_orders,
    orders.order_id == cancelled_orders.order_id,
    "left_anti"
)

active_orders.show()

Результат:

+--------+----------+------+
|order_id|order_date|amount|
+--------+----------+------+
|       1|2024-01-01| 100.0|
|       3|2024-01-03| 150.0|
+--------+----------+------+

Ключевые особенности LEFT ANTI JOIN:

1. Возвращает только левую таблицу — столбцы правой таблицы не включаются в результат
2. Условие соединения — обязательно указать условие (ON), иначе будет CROSS JOIN
3. Производительность — оптимизированная реализация в Spark, часто быстрее чем filter(NOT EXISTS...)

Эквивалентные операции:

# Способ 1: LEFT ANTI JOIN (рекомендуется)
df1.join(df2, "key", "left_anti")

# Способ 2: Filter с подзапросом (менее эффективно)
df1.filter(~col("key").isin(df2.select("key").rdd.flatMap(lambda x: x).collect()))

# Способ 3: Использование except (требует одинаковой схемы)
df1.select("key").exceptAll(df2.select("key"))

Практические сценарии использования:

1. Поиск отсутствующих записей: клиенты без заказов, продукты без продаж
2. Очистка данных: удаление тестовых или некорректных записей
3. Валидация: проверка целостности данных между системами
4. A/B тестирование: выделение контрольной группы

Пример с несколькими условиями:

# ANTI JOIN по нескольким полям
result = df1.join(
    df2,
    (df1.id == df2.id) & (df1.date == df2.date),
    "left_anti"
)

Важное замечание: LEFT ANTI JOIN в Spark игнорирует дубликаты в правой таблице — достаточно хотя бы одного совпадения, чтобы исключить строку из левой таблицы.