spark-csv 与 SparkR 和 RStudio 分崩离析答案

【问题标题】：spark-csv falls apart with SparkR & RStudiospark-csv 与 SparkR 和 RStudio 分崩离析
【发布时间】：2016-06-09 23:12:44
【问题描述】：

我已经尝试了 How to load csv file into SparkR on RStudio? 中建议的几种排列方式，但我只能让 Spark 解决方案的内存工作：

Sys.setenv(SPARK_HOME='C:/Users/myuser/apache/spark-1.6.1-bin-hadoop2.6')
.libPaths(c(file.path(Sys.getenv("SPARK_HOME"),"R","lib"),.libPaths()))

library(SparkR)
sparkR.stop()
sc=sparkR.init(master="local")
sqlContext=sparkRSQL.init(sc)

df=read.csv(file="C:/.../file.csv",     
            header=T,sep=",",na.strings = c('NULL',''),fileEncoding = "UTF-8-BOM",stringsAsFactors = F)

df<- createDataFrame(sqlContext, df)
df=dropna(df)
names(df)
summary(df)

上面的问题是，如果 file.csv 太大而无法放入内存，则会导致问题。（hack 是加载一系列 csv 文件并在 sparkR 中 rbind。）首选通过 read.df 读取 CSV 文件。

如果我将 init 更改为：

sc <- sparkR.init(master='local', sparkPackages="com.databricks:spark-csv_2.11:1.2.0")

按照使用 read.df 的建议，无论我做什么，sparkR 现在都被冲洗掉了。

df <- read.df(sqlContext, "C:/file.csv",          source="com.databricks.spark.csv", header="true", inferSchema="true")

甚至

df<- createDataFrame(sqlContext, df)

呕吐：

Error in invokeJava(isStatic = FALSE, objId$id, methodName, ...) : 
  org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 1, localhost): java.lang.NullPointerException
    at java.lang.ProcessBuilder.start(Unknown Source)
    at org.apache.hadoop.util.Shell.runCommand(Shell.java:482)
    at org.apache.hadoop.util.Shell.run(Shell.java:455)
    at org.apache.hadoop.util.Shell$ShellCommandExecutor.execute(Shell.java:715)
    at org.apache.hadoop.fs.FileUtil.chmod(FileUtil.java:873)
    at org.apache.hadoop.fs.FileUtil.chmod(FileUtil.java:853)
    at org.apache.spark.util.Utils$.fetchFile(Utils.scala:406)
    at org.apache.spark.executor.Executor$$anonfun$org$apache$spark$executor$Executor$$updateDependencies$5.apply(Executor.scala:405)
    at org.apache.spark.executor.Executor$$anonfun$org$apache$spark$executor$Executor$$updateDependencies$5.apply(Executor.scala:397)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:7

SparkR 缺少什么精灵粉？

是否有更简单的方法来指定或确认正确的数据块设置2.11:1.2.0？

有没有办法加载制表符分隔的文件或其他不需要数据块的格式？

附注我注意到 H2O 与 R 集成起来更加愉快，并且不需要神秘的咒语。 sparkR 的人真的需要让 sparkR 成为 1 班轮恕我直言......

【问题讨论】：

标签： r apache-spark sparkr spark-csv

【解决方案1】：

以下内容对我来说完美无缺：

Sys.setenv(SPARKR_SUBMIT_ARGS='"--packages" "com.databricks:spark-csv_2.11:1.4.0" "sparkr-shell"')
Sys.setenv(SPARK_HOME='/path/to/spark')
.libPaths(c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib"), .libPaths()))

library(SparkR)

sparkR.stop()

sc <- sparkR.init(master="local")
sqlContext <- sparkRSQL.init(sc)

df <- read.df(sqlContext, 
              "/path/to/mtcars.csv", 
              source="com.databricks.spark.csv", 
              inferSchema="true")

我将spark-csv_2.11-1.4.0.jar（最新的jar）放入spark/jars目录，适当地修改了env var，然后做剩下的事情。 collect(df) 表明它有效。

【讨论】：

当您说“适当地修改了 env var”时，您是指通过上面的 R 代码，还是必须设置一个 Windows 环境变量？
上面的代码。唯一的“外部”模块是确保 spark 二进制目录（bin 和 sbin）的路径在我的 PATH 中。不过，这不应该影响上述内容。

【解决方案2】：

预构建的 Spark 1.x 发行版是使用 Scala 2.10 而不是 2.11 构建的。因此，如果您使用这样的发行版（您似乎这样做了），您还需要一个适用于 Scala 2.10 的 spark-csv 构建，而不是适用于 Scala 2.11（就像您在代码中使用的那样）。将spark-csv_2.11 更改为spark-csv_2.10，它应该可以正常工作（另请参阅接受的SO 答案here 和here）。

【讨论】：