Как сохранить файл в кластере

Я подключен к кластеру с помощью ssh, и я отправляю программу в кластер, используя

spark-submit --master yarn myProgram.py

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

counts.write.json("hdfs://home/myDir/text_file.txt")
counts.write.csv("hdfs://home/myDir/text_file.csv")

Однако ни одна из них не работает. Программа заканчивается, и я не могу найти текстовый файл в myDir. Вы знаете, как я могу это сделать?

Кроме того, есть ли способ записи непосредственно на локальную машину?

EDIT: я узнал, что каталог home не существует, поэтому теперь я сохраняю результат как: counts.write.json("hdfs:///user/username/text_file.txt") Но это создает каталог с именем text_file.txt, и внутри у меня есть много файлов с частичными результатами внутри. Но я хочу получить один файл с конечным результатом внутри. Любые идеи, как я могу это сделать?

Ответ 1

Spark сохранит результаты в нескольких файлах, так как распределение вычисляется. Поэтому запись:

counts.write.csv("hdfs://home/myDir/text_file.csv")

означает сохранение данных на каждом разделе в виде отдельного файла в папке text_file.csv. Если вы хотите сохранить данные в виде одного файла, сначала используйте coalesce(1):

counts.coalesce(1).write.csv("hdfs://home/myDir/text_file.csv")

Это поместит все данные в один раздел, и количество сохраненных файлов будет таким образом 1. Однако это может быть плохой идеей, если у вас много данных. Если данные очень малы, то использование collect() является альтернативой. Это поместит все данные на машину драйвера в виде массива, который затем может быть сохранен как один файл.

Ответ 2

Вы можете объединить результаты в один файл из командной строки:

hadoop fs -cat hdfs:///user/username/text_file.txt/* > path/to/local/file.txt

Это должно быть быстрее, чем использование coalesce - по моему опыту все операции типа collect() выполняются медленно, потому что все данные передаются через мастер node. Кроме того, вы можете столкнуться с проблемами с collect(), если ваши данные превышают память на вашем хозяине node.

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

Чтобы удалить:

hadoop fs -rm -r hdfs:///user/username/text_file.txt/*

Ответ 3

Есть ли у вас какие-либо ошибки? Возможно, вы можете проверить, есть ли у вас правильные права на запись/чтение из этой папки.

Также подумайте, что Spark по умолчанию создаст папку text_file.txt с некоторыми файлами внутри, в зависимости от количества разделов, которые у вас есть.

Если вы хотите записать на свой локальный компьютер, вы можете указать путь с помощью file:///home/myDir/text_file.txt. Если вы используете путь типа /user/hdfs/... по умолчанию, он записывается в формате HDFS

Ответ 4

Чтобы иметь уникальный файл (не названный так, как вы хотите), вам нужно .repartition(1), посмотреть здесь, подключенный к вашему RDD. Я полагаю, что ваш путь hdfs неверен. В Spark HDFS для текстового файла используется значение по умолчанию, а в Hadoop (по умолчанию) в корневом каталоге нет родительского каталога, если только вы его не создали. Если вы хотите, чтобы файл csv/txt (с этим расширением) был единственным способом его записи, он не имеет функций RDD или DF, но, используя обычные библиотеки python csv и io, после того, как вы собрали, с .collect(), ваш RDD в мартиксе (набор данных не огромен).

Если вы хотите напрямую писать в своей файловой системе (а не на HDFS), используйте

counts.write.csv("file:///home/myDir/text_file.csv")

Но это не будет писать один файл с расширением csv. Он создаст папку с part-m-0000n из n разделов вашего набора данных.