Spark.read.fornat("kafka").load(), и там параметры вроде есть сколько читать, но надо уточнить

Да вот не хотелось бы. Ранее пробовал с офсетами приближенными к последним, и подобное поведение устраивало. Но под батчи не хотелось бы переделывать

Можно убивать стрим по таймауту😆

Жёска)

Коллеги, всем привет!

Вопрос

Есть папка в HDFS, в ней лежат паркеты. Паркеты имеют одну из двух схем:

- либо col_1, col_2, col_3

- либо col_1, col_2, col_4, col_5

Все колонки в обоих схемах являются nullable

Если я читаю спарком всю папку целиком и не указываю mergeSchema - то получившийся датафрейм имеет первый вариант схемы:

col_1, col_2, col_3

Кто-нибудь может объяснить почему это работает именно так?
Спасибо заранее

вероятно потому, что в паркете есть схема

или вопрос, почему не вторая схема?

Он первый файл читает, берёт оттуда схему и ему норм

О, спасибо. Предполагал что так, но пруфов не нашел :/

а, тут без Хайва, тогда да

Да, вопрос именно в логике, почему именно этот набор колонок, а не второй, например

When we read multiple Parquet files using Apache Spark, we may end up with a problem caused by schema differences. When Spark gets a list of files to read, it picks the schema from either the Parquet summary file or a randomly chosen input file

Нашел, спасибо большое

Есть возможность записи файлов с приоритетной схемой (_metadata, _common_metadata). Но эти файлы deprecated, так что пользоваться ими не совсем правильно. А всё потому что дополнительные файлы предоставляют ещё один вариант выстрела в ногу, но не решают проблемы в реализации schema evolution

схема эволюшн.. полностью эту проблему в случае любых схем не решает, наверное, ничто

Почитаю, спасибо

Всем привет. Кто нибудь знает как заменить значения типа Long в колонке (без udf и конструирования огромного case when)? na.replace тип Long не поддерживает :(

заменить на что?

@pal548 , есть объект Map[Long, Long]

заменить key -> value, как в na.replace только для Long колонки