Судя по всему UDF единственный разумный вариант :)

нихрена не понятно, если честно))

Хорошо, могу перефразировать, почему DataFrame.na.replace не поддерживает Long тип и как это обойти не прибегая к UDF и генерации огромного case when?

Тоже ничего не понял.
Напишите, пожалуйста, что у вас есть и что хотите иметь на выходе

всем привет. вопрос по structured streaming. кто-то сталкивался с ситуацией, когда нужно считать окна по какому-то кафка топику, но в нём только свежие данные и хочется дополнить их историей? как в таком случае инициализировать состояние?
в 3.2.0 появилась возможность задавать initial state для flatMapGroupsWithState, но хочется понять может это можно как-то сделать без arbitrary stateful processing

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

спасибо, но у меня вопрос скорее в том можно ли как-то предрассчитать и инициализировать начальное состояние, а не ждать пока оно накопится

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

Привет вопрос касательно оптимизации запроса.

- У меня есть некоторые большие и долгие вычисления с таблицей.
- Перед следующей трансформацией пускай у меня есть в датафреме df_init следующие колонки period, client_id, income.
- Следующей трансформацией для датафрейма является фильтрация по определенному period по некоторым признакам с income (например income > 200). Далее идет select client_id  и добавление некоторого флага is_fraud = 1, пускай это будет новый df_exclude
- И вот теперь я хочу df_exclude заджоинить с df_init, т.о. в df_init для всех client_id, которые есть в  df_exclude, будет is_fraud = 1 независимо от period
- В итоге делается некоторая action операция с df_init в конце, запись на диск

Как реализовать у меня вопросов нет. Проблема в следующем - я заметил, что спарк делает 2 параллельных поддерева вычислений для df_exclude и df_init, которые до момента их взамодействия одинаковые и соответственно все операции он также дублирует и запускате параллельно, которые довольно тяжелые.

Моя цель - сделать некоторое общее поддерева, которое оптимизитатор переиспользует, т.е. в буквальном смысле закешировал общий этап до их различных между собой преобразований. Самый тупой вариант я вижу запись на диск общего этапа и потом чтение. Возможно есть способы сделать это эффективнее ?

для контекста spark 2.4.4

> два поддерева, которые одинаковые

в физическом плане они тоже одинаковые? вплоть до чтения файлов?

+

у вас вроде фильтр в одном есть, а в другом - нет. или путаю?

да все так, фильтр только в df_exclude

а этот фильтр появляется в физическом плане в этапе чтения файлов? как predicate pushdown

нет

потому что этот фильтр (пункт 3) после предшествующих долгих вычислений и добавлений колонок, т.е. income появляется в середине графа где-то

возможно, это норм кейс для вызова df_init.cache() тогда
сам вызов пометит df_init как "кэшни мне это плз при вызове action"

попробуйте вызвать cache() после того, как "общий" df_init логически сформирован, а в конце принтануть план. поменялось ли что-нибудь?

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