Загружать локальный файл (а не HDFS) не удается в Spark

У меня есть один вопрос - как загрузить локальный файл (а не на HDFS, а не на S3) с помощью sc.textFile в PySpark. Я прочитал эту статью, затем скопировал sales.csv в master node локальный (а не HDFS), окончательно выполнив следующие

sc.textFile("file:///sales.csv").count()

но он возвращает следующую ошибку, говоря file:/click_data_sample.csv does not exist

Py4JJavaError: при вызове произошла ошибка г: org.apache.spark.api.python.PythonRDD.collectAndServe.: org.apache.spark.SparkException: Работа прерывается из-за срыва этапа: Задача 0 на этапе 3.0 не удалась 4 раза, последний сбой: потерянная задача 0.3 в стадии 3.0 (TID 10, ф-17х-хх-хх-xxx.ap-северо-восток-1.compute.internal): java.io.FileNotFoundException: Файл файла:/sales.csv не существует

Я пробовал file://sales.csv и file:/sales.csv, но оба они тоже потерпели неудачу.

Очень полезно дать мне добрый совет по загрузке локального файла.

Noted1:

My envrionment - Amazon emr-4.2.0 + Spark 1.5.2.
Все порты открыты.

Noted2:

Я подтвердил файл загрузки из HDFS или S3 работает.

Вот код загрузки из HDFS - загрузите csv, скопируйте в hdfs заранее, затем загрузите sc.textFile( "/path/at/hdfs" )

commands.getoutput('wget -q https://raw.githubusercontent.com/phatak-dev/blog/master/code/DataSourceExamples/src/main/resources/sales.csv')
commands.getoutput('hadoop fs -copyFromLocal -f ./sales.csv /user/hadoop/')
sc.textFile("/user/hadoop/sales.csv").count()  # returns "15" which is number of the line of csv file

Вот код загрузки из S3 - запустите файл csv на S3 заранее, затем загрузите с помощью sc.textFile( "s3n://path/at/hdfs" ) с помощью "s3n://".

sc.textFile("s3n://my-test-bucket/sales.csv").count() # also returns "15"

Ответ 1

Чтение файла происходит на исполнителе node. Чтобы ваш код работал, вы должны распространять файл по всем узлам.

Если программа драйвера Spark запускается на том же компьютере, на котором находится файл, то вы можете попробовать прочитать файл (например, с помощью f=open("file").read() для python), а затем вызвать sc.parallelize для преобразования содержимого файла к RDD.

Ответ 2

Если вы работаете в кластерном режиме, вам нужно скопировать файл на все узлы той же общей файловой системы. Затем искру читает этот файл, иначе вы должны использовать HDFS

Я скопировал txt файл в HDFS, и искра берет файл с HDFS.

Я скопировал txt файл в общую файловую систему всех узлов, а затем искривил этот файл.

Оба работали для меня

Ответ 3

У меня была аналогичная проблема, facha правильно, что данные, которые вы пытаетесь загрузить, должны быть доступны через ваш кластер (как для мастера, так и для исполнителей).

Я верю в ваше дело, что файл:/команда все еще пытается загрузить с вашего HDoFood, который не существует, вы можете проверить это, используя следующую команду

hadoop fs -cat yourfile.csv

Я решил эту проблему, загрузив файл из hdfs и прочитав из hdfs, вот код:

var conf = new org.apache.hadoop.conf.Configuration();     
var fs = org.apache.hadoop.fs.FileSystem.get(conf); 
var filenamePath = new org.apache.hadoop.fs.Path("myfile.json");  

   if (fs.exists(filenamePath))
   {
       fs.delete(filenamePath, true);
   }

   var fin = fs.create(filenamePath);
   fin.writeBytes(html);
   fin.close();

val metOffice = sql.read.json("myfile.json")