Вы вроде из оракла читаете . Вы посмотрите как он отдает. Может у вас там блобики

Я остаюсь при своём мнении, что спарк как некий аналог CDC - не лучший выбор. Слишком много усилий требуется, чтобы вывести на промышленный уровень: сбалансировать производительность, нагрузку на источник и целостность данных.
Если нужно одну таблицу или её часть затянуть на кластер и сразу же работать с этим датасетом, то супер, спарк как раз для этого! Но в остальных случая придётся  кастомизировать очень много.

Возвращаясь к вопросу, забирать по партициям - отличная идея, но описанные проблемы наводят на мысль, что партиции надо организовать вовсе не по полю "дата по". С ходу советую делать партиции по времени или дате изменения цены. Возможно, для этого придётся добавить поле со значением sysdate. Если поле добавить нельзя (и это плохо), то выходить на scn/ČSN или что там есть ещё! В этом случае гораздо проще настроить фильтр для дельты, нужно просто брать все, что больше максимального значения этого поля, полученного в ходе предыдущей загрузки. И при ежедневной загрузке нужно будет забирать одну партицию. Или пару-тройку партиций, если был перерыв в загрузке.

Забираемые данные уже можно побить для равномерно загрузки экзекуторов через numPartitions или предикаты. Предикаты гораздо более гибкие, можно указывать почти любые условия, главное - не допустить потери или дублирования записей. Перегруженный метод sparkSession.read.jdbc вам в помощь)
Через опции в этот метод можно передать sessionInitStatement, пригодится, если нужна какая-то подготовка, установка параметров сессии, например, для ретроспективного запроса. И fetchsize там же, в опциях увеличить.

Очень крутой коммент. Спасибо.

К сожалению пока мы ждем "промышленного решения" ETL. Нужно что-то делать на коленке "около_промышленное". Бизнес в основном говорит  "сделайте как в BW". И самое простое это скачать BW) Сейчас есть прямой коннект к HANA. И тупой способ в лоб "брать данные по партициям".  (так более равномерно распределяется нагрузка на бд)
К сожалению, партиция по "дате изменения" не подошла. Слишком большой разброс изменений цен. Из-за различных джобов "закрывающих" старые цены, получается в день по 700-1300 измененных партиций (за 3+ года) . Пока лучший вариант это "дата по". В день около 30-40 измененных партиций, включая ту самую на 400+млн строк.

Если scn/ČSN это что-то вроlе журнала изменений, то да, к этому стремимся и это next level. Это другой уровень забора данных и их обработки.

про oracle scn надо только помнить, что ставится он на уровне блока, туда могут попасть неизмененные данные + если данные удаляются, то такие строки надо будет искать вручную

Это не совсем так. Он не на уровне блока т.е он не один на блок , а в каждом блоке есть ITL  (interested transaction list) , а у строки в блоке есть ссылка на itl. Это очень упрощённо по scn.

И у каждой записи в itl свой scn.

itl это другое - это список сейчас активных транзакций, scn - это глобальный счетчик транзакций, который записывается в блок

погодь, или на каждый итл если их несколько, свой сцн?

есть ссылка на доку?

Нет. Это не так. Посмотрите на столицы в itl.

не пойму где про итл написано, в доке есть про блок (поумолчанию) и про строку, если включить опцию https://docs.oracle.com/database/121/SQLRF/pseudocolumns007.htm#SQLRF50953

Например тут можно посмотреть . Это нагуглил первую попавшуюся https://richardfoote.wordpress.com/2010/07/28/index-block-dump-block-header-part-ii-and-read-consistency-i-cant-read/

Почитать можно у льюиса про все это. Сделать самому дампы разных блоков. Это просто.

Но даже если оно на уровне itl и строки ссылаются на них - это же не гарантирует , что все строки будут иметь свой scn?

у группы строк, ссылающихся на 1 itl будет один сцн, даже если поменялась 1?

Там все сложнее. Есть system change number. Есть system commit number а записывается в одно место. Причем иногда в itl он проставляется вообще в момент select из другой сессии. Но если очень упрощать ( и принять факт , что описанное не будет соответствовать действительности ) , то если строка менялась в транзакции , то если строка менялась транзакцией , то у нее будет ссылка на itl, а если она менялась другой транзакцией , то будет ссылка на другую строку в itl. Т.е если есть ссылка на itl, то строка ей и менялась

если каждая транзакция создавала бы новую строку itl в заголовке, то заголовок бы сильно разросся, но их число обычно небольшое.
Чтот мне кажется, что при обращении к блоку по с фильтром по scn номер изменения в строках не смотрится, по этому могут попасть лишние не изменившиеся строки

Число itl задаётся при создании таблицы. Если свободного слота нет , то транзакция будет висеть.это известная проблема будет ожидание enq: TX -allocate ITL entry.

это если несколько конкурентных транзакции на 1 блоке . Пример: транзакция 1 поменяла 100 строк, об это записался scn в блок / itl, через час пришла транзакция 2 и поменяла другие 10 строк. Итл останется 1 и у него будет последний scn?  Значит наш забор по scn возьмет возьмет все строки блока даже не измененные

Как отработает ora_rowscn - это вообще загадка. На уровне блока тоже есть что-то типа last modified scn. А itl будет переиспользоватся.