Apache Spark'deki uzak makinedeki dosyayı okuyun ftp kullanarak

Question

ftp kullanarak Apache Spark (Scala sürümü) içindeki bir uzak makinedeki bir dosyayı okumayı deniyorum. Şu anda GitHub'da Databricks'in Öğrenme Spark repo'unda bir örnek takip ettim. Curl kullanarak, dosyayı indirebiliyorum, bu yüzden kullandığım yol var. Aşağıda

Ben çalıştırmak için deneyin kod snippet'idir:

var file = sc.textFile("ftp://user:pwd/192.168.1.5/brecht-d-m/map/input.nt") 
var fileDF = file.toDF() 
fileDF.write.parquet("out") 

dataframe üzerinde bir sayım yürütmeye çalışırken sonra ben şu olsun StackTrace (http://pastebin.com/YEq8c2Hf):

org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree: 
TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#1L]) 
+- TungstenExchange SinglePartition, None 
    +- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#4L]) 
     +- Project 
     +- Scan ExistingRDD[_1#0] 

... 

Caused by: org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: ftp://user:pwd@192.168.1.5/brecht-d-m/map/input.nt 

isterim Dosyaya ulaşılamayacağını farz edelim, ancak bu dosya ile çakışıyor: Curl:

curl ftp://user:pwd@192.168.1.5/brecht-d-m/map/input.nt 

curl ftp://user:pwd@192.168.1.5/brecht-d-m/map/input.nt 

Bu, terminalimdeki belirli bir dosyayı yazdıracaktır. Scala kodunda yanlış yaptığımı görmüyorum. Yukarıda verdiğim kod snippet'inde bir hata var mı, yoksa bu kod tamamen yanlış mı? peşin

sayesinde Brecht

Not: beklendiği gibi

• beri (tam yol (/home/brecht-dm/map/input.nt) da çalışmıyor belirtme Bu ayrıca "curl" de çalışmaz; "sunucu, belirtilen dizine değiştirmenizi reddetti"). Bunu Spark uygulamasında denemek, aramanın desteklenmediği IOException değerini verir (http://pastebin.com/b9EB9ru2). Yerel olarak çalışan (ör. Sc.textFile ("/ home/brecht-d-m/map/input.nt"))), mükemmel çalışır.

• Belirli bir dosya için dosya izinleri tüm kullanıcılar için R + W olarak ayarlanmıştır.

• Dosya boyutu (15MB) bir sorun olmamalı ve çok daha büyük dosyaları işleyebilmelidir.

• Yazılım sürümleri: Scala 2.11.7 Apache Spark 1.6.0 Java 1.8.0_74, Ubuntu 14.04.4

Answer 1

Ben geçici bir çözüm bulmak mümkün oldu. Aşağıdaki codesnippet Via:

import org.apache.spark.SparkFiles 

val dataSource = "ftp://user:pwd/192.168.1.5/brecht-d-m/map/input.nt" 
sc.addFile(dataSource) 
var fileName = SparkFiles.get(dataSource.split("/").last) 
var file = sc.textFile(fileName) 

ben (ilk kod pasajımızdaki aynı URL ile) FTP üzerinden bir dosya indirmek mümkün. Bu geçici çözüm ilk önce dosyayı (addFile aracılığıyla) indirecektir. Ardından, dosyanın indirildiği yolun yolunu buluyorum. Son olarak, bu dosyayı bir RDD'ye yüklemek için bu yolu kullanıyorum.