java - Sparklyr 失败并出现 java.lang.OutOfMemoryError : GC overhead limit exceeded

Question

我在使用 spark_apply 的 Spark 中遇到了 GC 开销限制超出错误。这是我的规范:

sparklyr v0.6.2 Spark v2.1.0 4个worker，8核，29G内存

闭包get_dates一次从Cassandra中提取一行数据。总共大约有 200k 行。该过程运行了大约一个半小时，然后给了我这个内存错误。

我已经尝试过 spark.driver.memory ，它应该会增加堆大小，但它不起作用。

有什么想法吗？用法如下

> config <- spark_config()
> config$spark.executor.cores = 1 # this ensures a max of 32 separate executors
> config$spark.cores.max = 26 # this ensures that cassandra gets some resources too, not all to spark
> config$spark.driver.memory = "4G"
> config$spark.driver.memoryOverhead = "10g" 
> config$spark.executor.memory = "4G"
> config$spark.executor.memoryOverhead = "1g"
> sc <- spark_connect(master = "spark://master",
+                     config = config)
> accounts <- sdf_copy_to(sc, insight %>%
+                           # slice(1:100) %>% 
+                           {.}, "accounts", overwrite=TRUE)
> accounts <- accounts %>% sdf_repartition(78)
> dag <- spark_apply(accounts, get_dates, group_by = c("row"), 
+                    columns = list(row = "integer",
+                                   last_update_by = "character",
+                                   last_end_time = "character",
+                                   read_val = "numeric",
+                                   batch_id = "numeric",
+                                   fail_reason = "character",
+                                   end_time = "character",
+                                   meas_type = "character",
+                                   svcpt_id = "numeric",
+                                   org_id = "character",
+                                   last_update_date = "character",
+                                   validation_status = "character"
+                                   ))
> peak_usage <- dag %>% collect  
Error: java.lang.OutOfMemoryError: GC overhead limit exceeded
    at org.apache.spark.sql.execution.SparkPlan$$anon$1.next(SparkPlan.scala:260)
    at org.apache.spark.sql.execution.SparkPlan$$anon$1.next(SparkPlan.scala:254)
    at scala.collection.Iterator$class.foreach(Iterator.scala:743)
    at org.apache.spark.sql.execution.SparkPlan$$anon$1.foreach(SparkPlan.scala:254)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeCollect$1.apply(SparkPlan.scala:276)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeCollect$1.apply(SparkPlan.scala:275)
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
    at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:275)
    at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$execute$1$1.apply(Dataset.scala:2371)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57)
    at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2765)
    at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$execute$1(Dataset.scala:2370)
    at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$collect$1.apply(Dataset.scala:2375)
    at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$collect$1.apply(Dataset.scala:2375)
    at org.apache.spark.sql.Dataset.withCallback(Dataset.scala:2778)
    at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collect(Dataset.scala:2375)
    at org.apache.spark.sql.Dataset.collect(Dataset.scala:2351)
    at sparklyr.Utils$.collect(utils.scala:196)
    at sparklyr.Utils.collect(utils.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at sparklyr.Invoke$.invoke(invoke.scala:102)
    at sparklyr.StreamHandler$.handleMethodCall(stream.scala:97)
    at sparklyr.StreamHandler$.read(stream.scala:62)
    at sparklyr.BackendHandler.channelRead0(handler.scala:52)
    at sparklyr.BackendHandler.channelRead0(handler.scala:14)
    at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:367)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:353)

Answer 1

也许我误读了你的例子，但内存问题似乎是在你收集时发生的，而不是在你使用spark_apply时发生的。尝试一下

config$spark.driver.maxResultSize <- XXX 

其中 XXX 是您期望需要的(对于类似的工作，我已将其设置为 4G)。请参阅https://spark.apache.org/docs/latest/configuration.html了解更多详情。