Можно ли как-то джоинить стриминговые данные?

Джоины стриминговых данных

Да, абсолютно возможно джоинить стриминговые данные, и это один из ключевых сценариев использования потоковых платформ обработки. Однако это гораздо более сложно, чем традиционные джоины в батч-системах, потому что данные прибывают в реальном времени и не имеют фиксированной границы.

Типы джойнов в стриминге

1. Stream-Stream Join (поток-поток)

Два потока данных джойняются в реальном времени. Самый сложный случай, так как нужно хранить состояние обоих потоков. Обычно используется временное окно (window):

# PySpark Structured Streaming
from pyspark.sql.functions import window

stream1 = spark.readStream.kafka(...).select("key", "timestamp1", "value1")
stream2 = spark.readStream.kafka(...).select("key", "timestamp2", "value2")

# Join на основе ключа и временного окна
result = stream1.join(
    stream2,
    (
        (stream1.key == stream2.key) &
        (stream1.timestamp1 >= stream2.timestamp2 - 300) &
        (stream1.timestamp1 <= stream2.timestamp2 + 300)
    )
)

2. Stream-Batch Join (поток-батч)

Поток джойнится с измерением (dimension table) из батч-хранилища. Проще, чем stream-stream, потому что батч-таблица статична:

# Загружаем измерение один раз
dimensions = spark.read.parquet("/path/to/dimensions")

# Джойним входящий поток с измерением
result = stream.join(
    dimensions,
    stream.customer_id == dimensions.id,
    "left"
)

Временные окна (Windows)

Ключевая концепция для stream-stream джойнов — это временное окно, которое определяет, как долго система будет ждать соответствующего события из второго потока:

• Tumbling window (30 секунд): данные группируются в непересекающиеся окна
• Sliding window (60 секунд с шагом 10 секунд): окна перекрываются
• Session window: окно закрывается при отсутствии событий в течение N времени

-- Kafka с временным окном в SQL
SELECT 
    a.key,
    a.value AS value_a,
    b.value AS value_b,
    WINDOW(a.event_time, INTERVAL 5 MINUTE) as time_window
FROM stream_a a
JOIN stream_b b
    ON a.key = b.key
    AND b.event_time BETWEEN a.event_time - INTERVAL 2 MINUTE 
        AND a.event_time + INTERVAL 2 MINUTE

Инструменты для стриминговых джойнов

Apache Flink — лучший выбор для сложных джойнов:

• Полноценная поддержка stream-stream, stream-batch джойнов
• Управление состоянием (state backend)
• Гарантии обработки (exactly-once semantics)

# Flink DataStream API
stream_a = env.add_source(...)
stream_b = env.add_source(...)

result = stream_a \
    .join(stream_b) \
    .where(lambda x: x[0]) \
    .equal_to(lambda x: x[0]) \
    .window(TumblingEventTimeWindow.of(Time.seconds(30))) \
    .apply(lambda x, y: (x[0], x[1] + y[1]))

Spark Structured Streaming — хороший выбор для микробатч обработки:

• Проще в освоении
• Интеграция с SQL
• Хорошая производительность для большых объёмов

Kafka Streams — для простых джойнов на Python/Java:

• KStream-KStream, KStream-KTable, KStream-GlobalKTable
• Низкая задержка
• Масштабируемость через партиции

Вызовы и решения

1. Управление состоянием

Для джоина нужно хранить события из обоих потоков в памяти до момента их сопоставления. Это требует:

• Достаточно памяти
• Правильного размера временного окна (не слишком большое, не слишком маленькое)
• Очистки старых данных (watermark/timeout)

2. Несинхронизированные часы

Источники могут отправлять события с разными задержками. Используют event time (время события) вместо processing time (время обработки):

from pyspark.sql.functions import col

# Правильно: используем колонку события в датасете
stream.withWatermark("event_timestamp", "10 minutes")

3. Out-of-order события

События могут прибывать в неправильном порядке. Watermark позволяет системе ждать запаздывающие события:

stream.withWatermark("timestamp", "5 minutes") \
    .groupBy(window("timestamp", "10 minutes")) \
    .count()

Best practices

1. Используйте event time, не processing time
2. Установите разумное временное окно — слишком большое = много памяти, слишком маленькое = потеря данных
3. Мониторьте lag джойна (задержку между приходом событий)
4. Тестируйте поведение при запаздывающих и дублирующихся событиях
5. Выбирайте инструмент на основе требований к задержке и сложности логики

Для production систем рекомендую Apache Flink из-за его надёжности и полноты функционала, а для простых случаев подойдёт Spark Structured Streaming или Kafka Streams.