файлы мы из hdfs берем, кстати

файлы ещё можно читать из хдфса спарком одновременно просто как набор)

ну да, вообще возьми имена распаралель и считай их на разных машинах и потом запроцессь

так ? val process: String => Unit // через цепочку String => RDD[data] => Unit sc.parallelize(files).map(process)

а где файлы-то лежат?

в одной папке?

нет, много папок и подпапок

так ? val process: String => Unit // через цепочку String => RDD[data] => Unit sc.parallelize(files).map(process)

у тебя имена которые грузишь; sc.parellilize(names).repartition(кол-во партиций).foreach / foreachPartition (process)

кол-во патриций зависит от ковло файлов и колва ехекуторов

нет, много папок и подпапок

их можно задать регулярным выражением?

да

они регуляркой и задаются

ну тогда проще некуда

ну и все)

spark.read.csv("регулярка")

сейчас так sc.textFile(day)

хочешь сделать по красоте - фигачь на датафрейм апи)

spark.read.csv("регулярка")

а он не в 200 патриций загонит все?

а партицировать нужно чтобы лучше делилось между воркерами ? кусочки поменьшее были ?

не, если дохрена данных, то будет размер данных / 128мб

а партицировать нужно чтобы лучше делилось между воркерами ? кусочки поменьшее были ?

да

хорошо я зашел спросить, спасибо сообществу, так бы дураком и умер

не, если дохрена данных, то будет размер данных / 128мб

так себе

если терабайт, то соответственно инпут сплитов будет 80к

вполне хватает)

можно и подтюнить, если так хочется

можно даже сменить 200 на что-нибудь другое, например на число ядер, отданных джобе)

ну, датафреймы будут быстрее и по красоте, если ему не надо будет выводить 100% считанных данных

Так то да; наверное через csv + регулярку все заработает

быстрее писать код на 500 тыщ процентов

У меня ощущение что прям для такого случая и придумали эту функцию

можно даже сменить 200 на что-нибудь другое, например на число ядер, отданных джобе)

а их и будет 200

там небось в map написано line.split(",")?

быстрее писать код на 500 тыщ процентов

пока не придётся гадать почему схема партицирования такая странная

а их и будет 200

ну дык потому что у всех на кластерах по 200 коров, а если больше, значит народу дохрена)

так-с

ну дык потому что у всех на кластерах по 200 коров, а если больше, значит народу дохрена)

не думаю, что оригинальная задумка была в коровах))

где почитать код с нормальной обработкой csv с hdfs чтобы вас вопросами не мучать ?

я бы посмотрел бы код и нормально сделал

Csv, нормально и хдфс в одном предложении - это перебор ):

Нутпопробуй csv функцию юзать как подсказал Евгений

я потому на паркет и перепиливаю, но старые поставщики пока на cvs

https://docs.databricks.com/spark/latest/data-sources/read-csv.html

думаю оно само заработает лучше чем сейчас у тебя

спасибо, буду завтра палкой тыкать

и вообще про spark sql почитай, забудешь про rdd и про ручной парсинг, как про страшный сон

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

и вообще про spark sql почитай, забудешь про rdd и про ручной парсинг, как про страшный сон

К сожалению, нельзя этого делать

есть ощушение что таска запилить паркет превратиться в переписать все к чертовой матери

есть ощушение что таска запилить паркет превратиться в переписать все к чертовой матери

Ну да у тебя очевидно что датасеты юзать и скуль апи спарка над

ну rdd не работают с паркетами

ну rdd не работают с паркетами

все работает

ага, rdd.toDF().write.format("parquet")?

и наоборот)

)) можно не использовать спарк апи же

ну rdd не работают с паркетами

на это у меня уже тестик написан, можно при желании перегнать =)

все работает

у спарк-контекста нет метода чтения паркета, он есть только у спарк-сессии

у спарк-контекста нет метода чтения паркета, он есть только у спарк-сессии

А ты читай ниже; уже выше

ну, всё равно придётся считать в датафрейм, а потом преобразовывать его в rdd

нет

у спарк-контекста нет метода чтения паркета, он есть только у спарк-сессии

Их можно конвертнуть

нет

пандасом что ли?)

Их можно конвертнуть

проще всего воспользоваться датафрейм апи. После чтения файлов на любом этапе можно сделать df.rdd, и продолжить использовать старый код

пандасом что ли?)

паркет на прямую юзаешь, яж написал выше

единственная проблема, что это будет RDD[Row], что конечно же весьма неудобно, и нужно его будет конвертнуть, вытазить оттуда типы

у меня был случай когда дешевле было и эффективнее без спарк скуля все делать

Косяк что напрямую паркет апи и свои инпут форматы считай делать надо

проще всего воспользоваться датафрейм апи. После чтения файлов на любом этапе можно сделать df.rdd, и продолжить использовать старый код

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

единственная проблема, что это будет RDD[Row], что конечно же весьма неудобно, и нужно его будет конвертнуть, вытазить оттуда типы

Это тоже не так страшно

если там такой код, то проще переписать)

На фоне всего остального

не поделитесь инфой, где такое наследие обнаружили?)

Нет, извини.

Не офф место работы даже

без проблем)

ну хоть надежда есть, что что-нибудь поменяется)

Ты же знаешь Scala, глянь одним глазом...говорили они, я видел все возможные варианты плохого кода думал я)))

Из плюсов - кода мало и есть приятный чат

Сами глину намесят, а потом ругаются что на спарках плохая скала :/

Не ругаются) бизнесс устраивает все

На с++ было медленно, теперь все быстро

Поставщик поменял формат и появилась необходимость потрогать

ну тогда проще некуда

возвращаясб к хорошему доброму вечному =) паркет же я тоже могу регуляркой всосать ?

ss.read.parquet("hdfs://parquet/*")

Да, конечно

А можно и без звёздочки, просто папку

Или там внутри тоже папки?

ss.read.parquet("hdfs://parquet/*")

давайте все вместе для единообразия договоримся называть спарк-сессию так, как в документации, а именно просто spark)

давайте все вместе для единообразия договоримся называть спарк-сессию так, как в документации, а именно просто spark)

Без проблем, исправлюсь ) там папка с папкой и папкой, потому маску

ну в общем звёздочка должна обозначать набор папок последнего уровня

дайте угадаю, папки типа год/месяц/день?)

Ога