Opzioni per leggere file di grandi dimensioni (puro testo, xml, json, csv) da hdfs in RStudio con SparkR 1.5

Question

Sono nuovo a Spark e vorrei sapere se ci sono altre opzioni oltre a quelle sotto per leggere i dati memorizzati in un hdf da RStudio usando SparkR o se li uso correttamente. I dati potrebbero essere di qualsiasi tipo (testo puro, csv, json, xml o qualsiasi database contenente tabelle relazionali) e di qualsiasi dimensione (1kb - diversi gb).

So che textFile (sc, path) non dovrebbe essere più utilizzato, ma ci sono altre possibilità di leggere questo tipo di dati oltre alla funzione read.df?

Il codice seguente utilizza il read.df e jsonFile ma jsonFile produce un errore:

Sys.setenv(SPARK_HOME = "C:\\Users\\--\\Downloads\\spark-1.5.0-bin-hadoop2.6") 
.libPaths(c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib"), .libPaths())) 
#load the Sparkr library 
library(SparkR) 

# Create a spark context and a SQL context 
sc <- sparkR.init(master="local", sparkPackages="com.databricks:spark-csv_2.11:1.0.3") 
sqlContext <- sparkRSQL.init(sc) 

#create a sparkR DataFrame 
df <- read.df(sqlContext, "hdfs://0.0.0.0:19000/people.json", source = "json") 
df <- jsonFile(sqlContext, "hdfs://0.0.0.0:19000/people.json") 

read.df funziona per JSON, ma come faccio a leggere il testo, come messaggi di log che sono separati solo da una nuova linea? Per esempio.

> df <- read.df(sqlContext, "hdfs://0.0.0.0:19000/README.txt", "text") 
    Error in invokeJava(isStatic = TRUE, className, methodName, ...) : 
    java.lang.ClassNotFoundException: Failed to load class for data source: text. 
    at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.lookupDataSource(ResolvedDataSource.scala:67) 
    at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.apply(ResolvedDataSource.scala:87) 
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:114) 
    at org.apache.spark.sql.api.r.SQLUtils$.loadDF(SQLUtils.scala:156) 
    at org.apache.spark.sql.api.r.SQLUtils.loadDF(SQLUtils.scala) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:606) 
    at org.apache.spark.api.r.RBackendHandler.handleMethodCall(RBackendHandler.scala:132) 
    at org.apache.spark.api.r.RBackendHandler.channelRead0(RBackendHandler.scala:79) 
    at org.apache.spark.ap 

L'errore con jsonFile è:

> df <- jsonFile(sqlContext, "hdfs://0.0.0.0:19000/people.json") 
    Error in invokeJava(isStatic = FALSE, objId$id, methodName, ...) : 
    java.io.IOException: No input paths specified in job 
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:201) 
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313) 
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:207) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 
    at scala.Option.getOrElse(Option.scala:120) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 
    at scala.Option.getOrElse(Option.scala:120) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35) 
    at org.apache.spark.rdd.RDD$$anonfu 

non so il motivo per cui read.df getta un errore perché non ho restard SparkR o chiamo SparkR.stop()

Per lo stesso codice oltre a usare read.df uso la funzione SparkR ::: textFile e sc invece di sqlContext (seguendo l'introduzione obsoleta su amplab).

Il messaggio di errore è:

data <- SparkR:::textFile(sc, "hdfs://0.0.0.0:19000/people.json") 
Error in invokeJava(isStatic = FALSE, objId$id, methodName, ...) : 
    java.lang.IllegalArgumentException: java.net.URISyntaxException: Expected scheme-specific part at index 5: hdfs: 
    at org.apache.hadoop.fs.Path.initialize(Path.java:206) 
    at org.apache.hadoop.fs.Path.<init>(Path.java:172) 
    at org.apache.hadoop.fs.Path.<init>(Path.java:94) 
    at org.apache.hadoop.fs.Globber.glob(Globber.java:211) 
    at org.apache.hadoop.fs.FileSystem.globStatus(FileSystem.java:1644) 
    at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:257) 
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:228) 
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313) 
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:207) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 
    at scala.Option.getOrElse(Option.scala:120) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 
    at or 

Questo errore si presenta come il percorso non è corretto, ma non so perché.

Quello che ho attualmente in uso:

scintilla 1.5.0-bin-hadoop2.6 Hadoop-2.6.0 di Windows (8.1) R Versione 3.2.2 Rstudio versione 0.99.484

Spero che qualcuno possa darmi qualche suggerimento su questo argomento qui.

Answer 1

Prova

% hadoop fs -put people.json/
    % sparkR 
    > people <- read.df(sqlContext, "/people.json", "json") 
    > head(people) 

Answer 2

probabilmente avete bisogno di una libreria per il parsing altri file, come DataBricks CSV libreria:

https://github.com/databricks/spark-csv

allora si sarebbe iniziare R con il pacchetto caricato, ad esempio:

$ sparkR --packages com.databricks:spark-csv_2.10:1.0.3

e caricare il file come:

> df <- read.df(sqlContext, "cars.csv", source = "com.databricks.spark.csv", inferSchema = "true")

Questo presuppone che il file di test "cars.csv" nella directory HDFS casa.

hth