r - 如何使用 SparkR 的 as.DataFrame() 将大型 R data.frames 加载到 Spark 中？

Question

我的目标是将大型 R data.frame 加载到 Spark 中。 data.frame 的大小是 500 万。各种类型的行和 7 列。一旦加载到 R 中，这个 data.frame 占用大约。 200MB 内存。但是，当我尝试使用 as.DataFrame() 函数将其加载到 Spark 中时，R session 永远被占用，它已经运行了 1 小时，我不得不取消该操作。

详情如下:

我正在创建以下数据集以在此示例中使用:

n=5e6 # set sample size

d <- data.frame(
    v1=base::sample(1:9,n,replace=TRUE), 
    v2=base::sample(1000:9000,n,replace=TRUE), 
    v3=seq(as.Date("2016-01-01"), as.Date("2016-12-31"), by = "day")[base::sample(1:365,n,replace=TRUE)],
    v4=LETTERS[base::sample(1:length(LETTERS),n,replace=TRUE)], 
    v5=base::sample(1000:9000,n,replace=TRUE)/1000, 
    v6=seq(ISOdate(2016,1,1), ISOdate(2018,1,1), "sec")[base::sample(1:63158401,n,replace=TRUE)],
    v7=c(TRUE,FALSE)[base::sample(1:2,n,replace=TRUE)]
)

上面创建了一个示例 data.frame

大小，大约 200mb:

paste0("size: ", round(as.numeric(object.size(d))/1000000,1)," mb")

接下来，我创建一个 Spark session :

Sys.setenv(SPARK_HOME='C:\\soft\\spark-2.0.0-bin-hadoop2.7',HADOOP_HOME='C:\\soft\\hadoop')
.libPaths(c(file.path(Sys.getenv('SPARK_HOME'), 'R', 'lib'),.libPaths()))
Sys.setenv('SPARKR_SUBMIT_ARGS'='"sparkr-shell"')

library(SparkR)
library(rJava)
sparkR.session(enableHiveSupport = FALSE,master = "local[*]", sparkConfig = list(spark.driver.memory = "1g",spark.sql.warehouse.dir="C:\\soft\\hadoop\\bin"))

现在，我正在尝试将上面创建的 data.frame 加载到 Spark 中:

d_sd <- as.DataFrame(d)

上面的命令需要永远运行。

我做错了什么吗？它可以与我原始 R data.frame 中列的 class() 相关吗？我应该采用替代方法将大型数据集从 R 加载到 Spark 中吗？如果是，请随时提出建议。

提前谢谢你。

附言:

我能够使用这种方法在 Spark 中快速转换和操作小型数据集。

以下是有关我的 R session 和我正在运行的操作系统的一些背景信息:

R 版本 3.2.5 (2016-04-14)平台:x86_64-w64-mingw32/x64(64 位)运行于:Windows 7 x64 (build 7601) Service Pack 1

我在 Windows 7 Professional(64 位)、8 GB RAM 下运行 Microsoft 版本的 R (Revolution)。处理器:i5-2520M @ 2.50GHz

---

编辑 2016-09-19:

谢谢 Zeydy Ortiz 和 Mohit Bansal。根据您的回答，我尝试了以下方法，但我仍然面临同样的问题:

Sys.setenv(SPARK_HOME='C:\\soft\\spark-2.0.0-bin-hadoop2.7',HADOOP_HOME='C:\\soft\\hadoop')
.libPaths(c(file.path(Sys.getenv('SPARK_HOME'), 'R', 'lib'),.libPaths()))
Sys.setenv('SPARKR_SUBMIT_ARGS'='"sparkr-shell"')

library(SparkR)
library(rJava)
sparkR.session(enableHiveSupport = FALSE,master = "local[*]", sparkConfig = list(spark.driver.memory = "1g",spark.sql.warehouse.dir="C:\\soft\\hadoop\\bin"))


n=5e6 # set sample size

d_sd <- createDataFrame(sqlContext,data=data.frame(
        v1=base::sample(1:9,n,replace=TRUE), 
        v2=base::sample(1000:9000,n,replace=TRUE), 
        v3=seq(as.Date("2016-01-01"), as.Date("2016-12-31"), by = "day")[base::sample(1:365,n,replace=TRUE)],
        v4=LETTERS[base::sample(1:length(LETTERS),n,replace=TRUE)], 
        v5=base::sample(1000:9000,n,replace=TRUE)/1000, 
        v6=seq(ISOdate(2016,1,1), ISOdate(2018,1,1), "sec")[base::sample(1:63158401,n,replace=TRUE)],
        v7=c(TRUE,FALSE)[base::sample(1:2,n,replace=TRUE)]
    ))

将 R df 转换为 Spark df 的命令运行了几个小时。不得不取消。请协助。

---

编辑 2016-12-14:

以上是使用 Spark 1.6.1 和 R 3.2.0 尝试的。我最近使用 Spark 2.0.2(最新)和 R 3.2.5 尝试过这个，我遇到了同样的问题。

如有任何帮助，我们将不胜感激。

Answer 1

这与内存限制有关，为什么必须先创建 Base Dataframe 并将其转换为 Spark DataFrame？

您可以将这两个步骤合并为一个并获得结果:

Sys.setenv(SPARK_HOME='C:\\soft\\spark-2.0.0-bin-hadoop2.7',HADOOP_HOME='C:\\soft\\hadoop')
.libPaths(c(file.path(Sys.getenv('SPARK_HOME'), 'R', 'lib'),.libPaths()))
Sys.setenv('SPARKR_SUBMIT_ARGS'='"sparkr-shell"')

library(SparkR)
library(rJava)
sparkR.session(enableHiveSupport = FALSE,master = "local[*]", sparkConfig = list(spark.driver.memory = "1g",spark.sql.warehouse.dir="C:\\soft\\hadoop\\bin"))

然后你可以加载你的自卫队:

n=5e6 # set sample size

d_sd <- as.DataFrame(data.frame(
    v1=base::sample(1:9,n,replace=TRUE), 
    v2=base::sample(1000:9000,n,replace=TRUE), 
    v3=seq(as.Date("2016-01-01"), as.Date("2016-12-31"), by = "day")[base::sample(1:365,n,replace=TRUE)],
    v4=LETTERS[base::sample(1:length(LETTERS),n,replace=TRUE)], 
    v5=base::sample(1000:9000,n,replace=TRUE)/1000, 
    v6=seq(ISOdate(2016,1,1), ISOdate(2018,1,1), "sec")[base::sample(1:63158401,n,replace=TRUE)],
    v7=c(TRUE,FALSE)[base::sample(1:2,n,replace=TRUE)]
))

你也可以引用类似的问题:How best to handle converting a large local data frame to a SparkR data frame?