Спасибо огромное за обсуждение. Понятно что надо думать дальше и искать оптимальное решение.

честн говоря, с генераторами никогда не было необходимости заморачиваться. + аргументу про результат фильтров и жойнов. с детерминированными input output проще дебагать, сравнивая с ситуацией когда инпут при каждом запуске новый.

если уже есть статические тесты, и есть потребность их улучшать (слишком много багов не ловят) я бы, скорее, смотрел в сторону mutation testing

на моей практике (backend, data eng) статических тестов хватало в 99% случаях - включая интеграционные с контейнеризованными датабазами (если app для работы они нужны)

дальше уже начинает действовать правило парето - условно 70% покрытие в сотни раз лучше, чем 7%. но доводить до 100% может быть слишком дорого

а если на данный момент теста нет ни одного, то хотя бы начать с одного - двух, статических:

1. создаем temp folder
2. генерим статический датафрейм spark.createDataFrame
3. пишем в темп фолдер в ожидаемом формате - это будет наш инпут
4. запускаем основное приложение - spark = SparkSession.master(local)..
5. assert-ы

можно начать с прокидывания везде в приложении спарк сессии "насквозь". классический dependency injection

товарищи, а кто-нибудь в курсе, pyarrow нормально живет в ситуации, когда неймнод HDFS несколько и периодически активная меняется?

Надо в коде смотреть, естб ли там обработка отлупа, сомневаюсь, что где-то это нормально описано

Они ещё апи каждый раз меняют

Ща сек

ну, если вот так устроит...

host : NameNode. Set to "default" for fs.defaultFS from core-site.xml. 
port : NameNode's port. Set to 0 for default or logical (HA) nodes.

то есть вообще говоря конфиг подбирается прям из ходуп конфига, если мы подключаемся напрямую, а там учтено HA

и по идее жавка объект, через который работает с++, через который работает питон, поддерживает HA

это если через libhdfs работать, а pyarrow может не только через libhdfs работать

Ну так-то да, это я понял, спасибо. В итоге сделал через сессию спарка

подскажите где почитать как хранит данные RDD, допустим стринг колонка. это набор байтов или там ссылки на жава оъекты? если я возвращаю ArrayList из mapPartitions в виде итератора, что в RDD будет ? ссылка на ArrayList или он вынужден deep copy делать ? зы. off-heap сторидж вырублен

Хороший, интересный вопрос! Я не помню, чтобы это было где-то в одном месте, но можно обратиться к документации:
https://spark.apache.org/docs/latest/rdd-programming-guide.html

Или к сторонним источниками, например сюда (тут очень кратко):
https://www.waitingforcode.com/apache-spark/data-representation-in-spark-rdd/read

Как я себе сейчас это представляю, RDD всегда параметризован типом T: ClassTag, который может быть практически любым. Как правило, если работа идёт с датафреймами, можно прозрачно получить RDD[Row], но у него имеется внутреннее представление, RDD[InternalRow], которое быстрее, так как работает с сериализованными данными. InternalRow реализуется в разных классах:
https://www.waitingforcode.com/apache-spark-sql/generated-code-spark-sql/read
Из этой статьи видно, что String, например, может быть представлен и в виде raw memory (unsafe projection) и как ссылка на Object (safe projection). Оба эти представления располагаются в heap, если не включен off-heap.

Если возвращается (ArrayList<T>) list.iterator, по-моему, будет получен RDD<T>, но если возвращается сам list, то будет RDD<ArrayList<?>>, тип списка будет стёрт, но RDD сохранит TypeTag и всё равно будет знать, что там конкретный тип

Применяется ли copy, не могу сказать точно, но ни разу не видел в коде вызова этого метода для java Object. Может не туда смотрел. UnsafeRow.copy используется очень часто, и я применял этот метод много в каких случаях, иначе сохраняются ссылки на одну и ту же область памяти

огромное спасибо! особливо за "Оба эти представления располагаются в heap, если не включен off-heap."
пойду переваривать

привет, помогите, пожалуйста:

df.filter(
 col("searchTermGroup").isInCollection(searchTermGroupsExcluded.value(1)))
)

вместо 1 нужно подставить значение колонки col("userId)

Какая версия spark? Что такое searchTermGroupsExcluded? Какому датафрейму принадлежит userid?

Я бы сделал udf