Для текущего итога мало прочитать новую порцию событий. Объясняем состояние Structured Streaming, причины его роста и правила завершения расчётов без кода.
Почему новая порция не содержит весь ответ
Представим условное табло заказов, которое обновляется по мере поступления сведений. В предыдущей порции было семь покупок магазина, в новой пришли ещё две. Чтобы показать общий итог девять, обработке нужно помнить то, что уже было посчитано. Одного чтения двух новых событий недостаточно.
Spark Structured Streaming — механизм обработки непрерывно поступающих данных. Расчёты, которым нужен контекст прошлых порций, называют вычислениями с состоянием. Состояние здесь — промежуточные сведения, необходимые для продолжения работы: например, текущая сумма, данные незавершённой группы или записи, которые ещё ожидают пару из другого источника.
Что хранится между порциями
Смысл состояния задаёт сама задача. Для подсчёта заказов по магазину это может быть накопленное количество. Для среднего чека нужны сведения, позволяющие продолжить расчёт по новым покупкам. При поиске повторов важно помнить уже встреченные события, а при сопоставлении двух потоков — ещё не сопоставленные записи.
В руководстве Apache Spark такие вычисления отделены от простых преобразований отдельных записей. Если из каждой новой записи лишь выбирают нужные сведения, история других записей может не требоваться. Но наличие потокового источника само по себе не означает, что вся обработка обходится без памяти о прошлом.
Состояние и опубликованный итог — разные вещи
Значение на табло — результат для читателя. Состояние нужно обработке, чтобы корректно получить следующий результат. Оно может содержать больше сведений, чем видно в отчёте. Поэтому удаление старых строк из итоговой таблицы не обязательно уменьшает внутренний объём работы потокового расчёта.
Условный пример: для среднего значения по группе хранится сумма и число наблюдений, хотя на экране показано только среднее. Если известен лишь прошлый средний чек без количества заказов, новые покупки нельзя правильно учесть простым усреднением двух показателей. Обработка должна сохранить достаточный контекст для выбранного действия.
Почему объём постепенно растёт
Пока расчёт помнит всё больше разных групп или ожидающих событий, состоянию нужны дополнительные ресурсы. Например, появляются новые магазины, временные интервалы и ещё не завершённые сопоставления. Если у задачи нет правила завершения или удаления ненужного контекста, поток продолжает накапливать работу.
Apache Spark использует специальное хранилище состояния между порциями. В документации описаны способы его размещения в памяти и на диске. Выбор хранилища влияет на использование ресурсов, но не отвечает на главный смысловой вопрос: какие сведения ещё могут понадобиться расчёту, а какие уже допустимо убрать. Больший диск не заменяет такое решение.
Когда прошлый интервал можно завершить
Событие может произойти раньше, чем попасть в обработку. Поэтому при расчёте по времени события нужно учитывать опоздания. Watermark, или временная граница продвижения потока, помогает определить, для каких завершившихся интервалов можно очищать состояние при подходящих операциях. Граница связана с увиденным временем событий и допустимым запаздыванием, а не просто с часами на стене.
Согласно описанию обработки с состоянием Microsoft Fabric, очистка также зависит от условий конкретного расчёта. Само упоминание временной границы не гарантирует ограниченный размер любой группы или любого соединения. Особенно важно понять, что произойдёт, если один из источников перестанет продвигаться.
Как обсудить опоздания на простом примере
Допустим, команда считает покупки по завершённым временным интервалам и согласилась учитывать опоздавшие сведения в установленных пределах. Пока интервал ещё может дополняться, его промежуточный итог нужен обработке. После продвижения границы старые сведения могут перестать менять этот расчёт. Что делать с более поздними исправлениями, обсуждают отдельно.
Нельзя обещать, что событие за пределами границы обязательно будет учтено. Команде полезно заранее решить, нужен ли отдельный путь сверки и пересчёта. Хранение исходных данных в файлах и оперативный потоковый итог решают разные задачи; наличие файла само по себе ещё не описывает процедуру исправления отчёта.
Какие вопросы помогут понять расход ресурсов
Перед обсуждением настройки спросите: что именно помнит расчёт, сколько разных групп открыто, когда группа завершается и как обрабатываются опоздания? Это помогает отличить рост состояния от других причин расхода памяти. В частности, память управляющей части Spark — отдельная тема.
Разобраться в распределённых и потоковых вычислениях можно на курсе PySpark для инженеров и аналитиков данных. Полезный первый шаг — нарисовать словами, какие сведения переходят из одной порции в следующую, и объяснить, почему без них изменился бы ответ.
