```I start spark with:
config[["spark.dynamicAllocation.enabled"]] <- FALSE
config[["spark.executor.extraJavaOptions"]] <- paste0("-XX:+UseG1GC')
config[["spark.driver.extraJavaOptions"]] <- paste0("-XX:+UseG1GC')
config[["spark.r.command"]] <- paste0(R.home("bin"), "/Rscript")
config[["spark.sql.crossJoin.enabled"]] <- TRUE
config[["spark.driver.maxResultSize"]] <- "10G"
config[["spark.executor.maxResultSize"]] <- "10G"
config[["spark.executor.cores"]] <- 5
config[["spark.driver.memory"]] <- "15G"
config[["spark.executor.memory"]] <- "15G"
config[["spark.network.timeout"]] <- "1024s"
...
sparkConfigIn[["spark.locality.wait"]] <- 3000
sparkConfigIn[["spark.task.maxFailures"]] <- 1
sparkConfigIn[["spark.scheduler.minRegisteredResourcesRatio"]] <- 1
sparkConfigIn[["spark.executor.heartbeatInterval"]] <- "10s"
sparkConfigIn[["spark.driver.extraJavaOptions"]] <- "-XX:MaxPermSize=384m"
sparkConfigIn[["spark.executor.extraJavaOptions"]] <- "-XX:MaxPermSize=384m"
sparkConfigIn[["spark.ext.h2o.client.extra"]] <- "max_mem_size=\"10G\""
...
spark <- spark_connect(master=sparkMasterURL,
spark_home = sparkHome, config=config)
h2o <- rsparkling::h2o_context(spark)```
Here is an example of setting I using with spark, H2O and It is working fine
Example, I ‘m using worker (master) nodes with 48 CPUs and 384GB of memory
#!/usr/bin/env bash NB_CPU_PER_NODE=48 NODE_MEMORY=384 YARN_NODEMANAGER_RESSOURCE_MEMORY=378 # NODE_MEMORY - 6 GB or plus for the system os ... NB_CPU_PER_EXECUTOR=5 #=> Number of CPU per Executor NB_EXECUTOR_PER_NODE=$(( (NB_CPU_PER_NODE - 1) / NB_CPU_PER_EXECUTOR)) NB_EXECUTORS_MAX=$((NB_EXECUTOR_PER_NODE * NB_WORKERS - 1)) # -1 for the application master on one node EXECUTOR_MEMORY_PLUS_OVERHEAD_MEMORY=$((YARN_NODEMANAGER_RESSOURCE_MEMORY / NB_EXECUTOR_PER_NODE)) EXECUTOR_MEM_OVERHEAD=$(( EXECUTOR_MEMORY_PLUS_OVERHEAD_MEMORY * 15/100 )) EXECUTOR_MEMORY=$((EXECUTOR_MEMORY_PLUS_OVERHEAD_MEMORY - EXECUTOR_MEM_OVERHEAD)) DRIVER_MEMORY=${EXECUTOR_MEMORY}g DRIVER_MEM_OVERHEAD=${EXECUTOR_MEM_OVERHEAD} DRIVER_CORE=$NB_CPU_PER_EXECUTOR printf "\n---------------------------------\n" echo "Settings" echo "---------------------------------" echo "NB_EXECUTORS_MAX=$NB_EXECUTORS_MAX" echo "NB_CPU_PER_EXECUTOR=$NB_CPU_PER_EXECUTOR" echo "NB_EXECUTOR_PER_NODE=$NB_EXECUTOR_PER_NODE" echo "EXECUTOR_MEMORY_PLUS_OVERHEAD_MEMORY=${EXECUTOR_MEMORY_PLUS_OVERHEAD_MEMORY}g" echo "EXECUTOR_MEMORY=${EXECUTOR_MEMORY}g" echo "EXECUTOR_MEM_OVERHEAD=${EXECUTOR_MEM_OVERHEAD}g" echo "DRIVER_MEMORY=${DRIVER_MEMORY}" echo "DRIVER_MEM_OVERHEAD=${DRIVER_MEM_OVERHEAD}g" echo "DRIVER_CORE=$DRIVER_CORE"
Here is an example on how to get the right number of executor based on the executor memory and executor cpu
below is my sample code. library(survival)
library(MASS)
library(h2o)
library(rsparkling)
library(sparklyr)
library(dplyr)
Sys.setenv(SPARK_HOME="/usr/lib/spark")
conf <- spark_config()
conf$spark.executor.instances <- 171
spark.yarn.executor.memoryOverhead<- 2048
conf$spark.executor.memory <- "18g"
conf$spark.executor.cores <- 5
spark.yarn.driver.memoryOverhead<- 39936
conf$spark.driver.memory<-"57.6g"
conf$spark.driver.cores<- 5
conf$'sparklyr.shell.executor-memory' <- "32g"
conf$'sparklyr.shell.driver-memory' <- "32g"
conf$spark.yarn.am.memory <- "32g"
conf$spark.dynamicAllocation.enabled <- "false"
sc <- spark_connect(master = "yarn-client", version = "2.4.3",config = conf)
data=spark_read_parquet(sc,path ="s3aparquetfile")
TRanform data
mydata2<-mydata
data_modelt_Full_Sample<-mydata%>%mutate(arr = explode(array_repeat(column_1,column1)))
H2o Setup
library(h2o)
library(rsparkling)
h2o_context(sc)
h2o setup end
Full_Data<-as_h2o_frame(sc,data_modelt_Full_Sample)
H2o tranform factors
Full_Data$column_2 <- as.factor(Full_Data$ column_2)
Full_Data$column_3 <- as.factor(Full_Data$ column_3)
.
.
.
.
.upto 270 columns
…
End
Full_Data$strata<-as.factor(Full_Data$ strata)
Coxph h2o model
predictorsSt <- c( column_1,column_2,column_3,….. up to 270 columns
)
h2o_modelt_D1_p <- h2o.coxph(
x = predictorsSt,
event_column = " event_column ",
start_column = " start_column ",
stop_column = " stop_column ",
offset_column = " offset_column ",
ties = c("breslow"),
stratify_by = ("strata"),
interaction_pairs=list(
c("X", "X1"),
c("X", "X2"),
c("X3","X"),
c("X3","X2")
), training_frame = Full_Data)
coefficients_Stage_1<-h2o_modelt_D1_p@model$coefficients_table$coefficients
config[["spark.driver.maxResultSize"]] <- "10G"
config[["spark.executor.maxResultSize"]] <- "10G"
config[["spark.executor.cores"]] <- 5
config[["spark.executor.instances"]] <- 5
config[["spark.driver.cores"]] <- 5
config[["spark.driver.memory"]] <- "54G"
config[["spark.executor.memory"]] <- "54G"
config[["spark.network.timeout"]] <- "1024s"
Hi @DivyaMereddy007 , I looked at to your code but It’s like your are using only R instead of RSparklink Water and the second thing I noticed is that you have a lot of dimensions (SUM(Number of levels of all categorical variables) + Number of numerical variables
I didn’t see where you used H2O or RSparkling or Sparklyr ….
(Doc: http://docs.h2o.ai/sparkling-water/2.2/latest-stable/doc/rsparkling.html)
I think your problem is that you are using R (not very scalable ) which always try to load all the data into memory plus the fact that you have a lot of categorical features increasing the dimentionality of your problem