а как over без окна инициализировать, в pyspark - это обязательный аргумент. Я сделал просто через фильтр, и первое значение mean_rank (ибо по нему отсортировано) для нуля коллектю, и еще один фильтр по этому значению.

Over - это и есть показатель оконной функции, но есть функции где order by не обязательный параметр.

Over в pyspark это метод класса column, который на вход принимает окно.

Ну попробуй вызвать просто min(...) Over() в пайспарке. В скале работает и передается дефолтовое окно(в плане видно)

Так я же пробовал))

TypeError: over() missing 1 required positional argument: 'window'

да и в доках ясно видно что

def over(self, window):

Судя по докам скалы он окно на все данные применяет - Defines an empty analytic clause. In this case the analytic function is applied and presented for all rows in the result set. Тоесть тоже что и Иван выше написал.

Вечер добрый. Подскажите по checkpoint-ам для stateful обработки (классический Spark Streaming DStreams)
В кафку сыпятся события, объединенные id - все события с одинаковым id принадлежат одной сессии. Spark по каждому событию создает/продляет сессию, суммирует разные данные и т.п., для связки м/у батчами используется mapWithState. В примерах в качестве checkpoint-а речь идет об HDFS, S3 и подобных ФС.
А можно ли использовать в качестве checkpoint-а кафку? 😁
Похоже, что она может быть использована только для Metadata checkpoint, но в данном случае без Data checkpointing видимо не обойтись

Идея простая - не плодить сущностей, поэтому хочется обойтись без HDFS

У тебя стрим должен быть идемпотентным, данные фактически не так важны; у тебя же будут данные с предыдущего шага, какая разница сколько раз перезапустишь?

делая чекпоинт, он тоже будет перезапускать весь последний шаг

Пожтому часто последний шаг хранят в Кафке / в базе какой-то, чтоб на креше ресторнуться на нужный офсет кафки

вручную*

у чекпойнта много ограничений на код чейнж (раньше это было большой проблемой, я думаю ничего особо не поменялось)

Непонятно вот что - а как быть с потерянной сессией, которая уже собралась на предыдущих обработанных событиях. Так то она уже есть в чекпоинте - или я не прав и вообще неверно понял чекпоинт? После креша взял ее из чекпоинта и пошел продлять дальше

Или она уже есть в БД, и в случае креша ее брать оттуда?

что у тебя есть сессия?

какая разница как конектится

У тебя есть оффсет

и ручной коммит он саксес