Sparklyr

Sparkylr + sparkling water+ H2O

Markdown-Cheatsheet

Sparkling Water (H2O) Machine Learning

1 Code

1.1 Environmental variables

Sys.setenv(SPARK_HOME='/Users/olegbaydakov/Downloads/spark-2.2.0-bin-hadoop2.7/')
Sys.setenv(SPARK_LOCAL_IP='127.0.0.1')
Sys.setenv(HADOOP_CONF_DIR='/usr/local/Cellar/hadoop/2.8.1/libexec/etc/hadoop')
Sys.setenv(YARN_CONF_DIR = '/usr/local/Cellar/hadoop/2.8.1/libexec/etc/hadoop')
options(rsparkling.sparklingwater.version = "2.2.0") 
options(rsparkling.sparklingwater.location = "/Users/olegbaydakov/Documents/Books/R/sparkling-water-2.2.2/assembly/build/libs/sparkling-water-assembly_2.11-2.2.2-all.jar")

1.2 Libraries

library(rsparkling)
library(sparklyr)
library(h2o)

1.3 Spark configuration

config=spark_config()
config=c(config,list(
  "spark.executor.memory"="1G",
  "spark.driver.memory"="1G",
  "spark.executor.instances"="2",
  "spark.executor.cores" = "",
  "sparklyr.log.console" = TRUE,
  #     "spark.dynamicAllocation.enabled"=TRUE,
  #    "spark.shuffle.service.enabled"=TRUE,
  "spark.sql.tungsten.enabled"= TRUE
  #   "spark.cores.max" = "2"
  #      "spark.sql.warehouse.dir" = "/Users/olegbaydakov/Downloads/spark-2.2.0-bin-hadoop2.7/tmp/hive/warehouse"
))

1.4 Connect to Spark master (Standalone cluster)

sc <- spark_connect(master = "spark://127.0.0.1:7077", 
                    version = "2.2.0",
                    app_name = "sparklyr4",
                    spark_home = "/Users/olegbaydakov/Downloads/spark-2.2.0-bin-hadoop2.7/",
                    config = config)

1.5 Connect to YARN (version 1)

config=spark_config()
sc <- spark_connect(master = "yarn-client", 
                    version = "2.2.0",
                    app_name = "sparklyr4",
                    spark_home = "/Users/olegbaydakov/Downloads/spark-2.2.0-bin-hadoop2.7/",
                    config = config)

1.6 Connect to YARN (version 2)

sc <- 
  spark_connect(master="yarn",
                config = list(
                  default = list(
                    spark.submit.deployMode= "client",
                    spark.executor.instances= 2, 
                    spark.executor.memory= "2G",
                    spark.executor.cores= 2,
                    spark.driver.memory= "1G",
                    "sparklyr.log.console" = TRUE
                    )))

1.7 Connect to H2O

h2o_context(sc, strict_version_check = FALSE)

1.8 Data processing

library(dplyr)
mtcars_tbl <- copy_to(sc, mtcars, "mtcars")
src_tbls(sc) 
# from Hive
data_tbl <- tbl(sc, "table_name") 
data_tbl2 <- tbl(sc, sql("SELECT * from mtcars"))

# H2O
mtcars_h2o <- as_h2o_frame(sc, mtcars_tbl, strict_version_check = FALSE)

partitions <- mtcars_tbl %>%
  filter(hp >= 100) %>%
  mutate(cyl8 = cyl == 8) %>%
  sdf_partition(training = 0.5, test = 0.5, seed = 1099)

1.9 Train model

training <- as_h2o_frame(sc, partitions$training, strict_version_check = FALSE)
test <- as_h2o_frame(sc, partitions$test, strict_version_check = FALSE)

# fit a linear model to the training dataset
glm_model <- h2o.glm(x = c("wt", "cyl"), 
                     y = "mpg", 
                     training_frame = training,
                     lambda_search = TRUE)

glm_model

library(ggplot2)

# compute predicted values on our test dataset
pred <- h2o.predict(glm_model, newdata = test)
# convert from H2O Frame to Spark DataFrame
predicted <- as_spark_dataframe(sc, pred, strict_version_check = FALSE)

# extract the true 'mpg' values from our test dataset
actual <- partitions$test %>%
  select(mpg) %>%
  collect() %>%
  `[[`("mpg")

# produce a data.frame housing our predicted + actual 'mpg' values
data <- data.frame(
  predicted = predicted,
  actual    = actual
)
# a bug in data.frame does not set colnames properly; reset here 
names(data) <- c("predicted", "actual")

# plot predicted vs. actual values
ggplot(data, aes(x = actual, y = predicted)) +
  geom_abline(lty = "dashed", col = "red") +
  geom_point() +
  theme(plot.title = element_text(hjust = 0.5)) +
  coord_fixed(ratio = 1) +
  labs(
    x = "Actual Fuel Consumption",
    y = "Predicted Fuel Consumption",
    title = "Predicted vs. Actual Fuel Consumption"
  )

1.10 Spark disconnect

spark_disconnect(sc)
h2o.shutdown()
# To stop all running H2O sessions
# ps -efww | grep h2o

1.11 Docker file

Docker file

Docker image build and run