Structured Streaming: бесконечная таблица и watermarks

Диспетчеру «Гонца» нужно видеть, сколько заказов курьеры забирают в каждой зоне за каждые десять минут, и видеть это сейчас, а не утром. В блоке про Kafka мы разбирали, чем поток отличается от батча, и считали окна в Kafka Streams. Spark умеет то же самое, и почти тем же кодом, что для обычных таблиц.

Бесконечная таблица

Главная идея Structured Streaming: поток это таблица, в которую всё время дописываются строки. Вы пишете обычный запрос к этой таблице: фильтр, группировка, окна. А Spark сам решает, как пересчитывать результат, когда приходят новые строки.

Под капотом Spark работает микробатчами: забирает всё, что пришло с прошлого раза, и обрабатывает как маленькое задание. Задержка получается от сотен миллисекунд до секунд. Этого хватает почти всем задачам «Гонца».

Отличия от обычного DataFrame:

• источник читают через spark.readStream вместо spark.read, и схему задают явно;
• результат пишут через writeStream в приёмник: файлы, Kafka, консоль, таблицу в памяти;
• start() запускает бесконечный запрос, а show() и count() на потоке недоступны.

Окна по времени события

Считать «за последние десять минут» можно по двум часам: по времени, когда событие пришло в Spark, или по времени, когда оно случилось. Правильно второе: время события. Курьер забрал заказ в 12:04, а телефон без связи прислал событие в 12:15: оно всё равно должно лечь в окно 12:00-12:10.

Окно задаёт функция F.window("event_time", "10 minutes"). Она превращает время в структуру с началом и концом окна, и по ней группируют, как по обычной колонке.

Водяной знак

Но сколько ждать опоздавших? Если хранить каждое окно вечно, память кончится. Водяной знак отвечает на этот вопрос: withWatermark("event_time", "10 minutes") говорит Spark, что события опаздывают не больше чем на десять минут. Spark следит за самым поздним увиденным временем события и вычитает из него десять минут. Всё, что старше этой границы, отбрасывается, а окна до неё закрываются окончательно.

Водяной знак сдвигается после каждого микробатча, а действует в следующем.