Implement Arrow

.gitignore

@@ -48,3 +48,8 @@ flights_model
 flights_pipeline
 new_flights_model
 /checkpoints
+tests/testthat/test.csv
+tests/testthat/test.json
+tests/testthat/batch.csv/
+tests/testthat/iris-in/
+tests/testthat/iris-out/

.travis.R

@@ -3,13 +3,18 @@ args <- commandArgs(trailingOnly=TRUE)
 if (length(args) == 0) {
   stop("Missing arguments")
 } else if (args[[1]] == "--testthat") {
-  parent_dir <- dir("../", full.names = TRUE)
+  parent_dir <- dir(".", full.names = TRUE)
   sparklyr_package <- parent_dir[grepl("sparklyr_", parent_dir)]
   install.packages(sparklyr_package, repos = NULL, type = "source")
 
+  on.exit(setwd(".."))
+  setwd("tests")
   source("testthat.R")
 } else if (args[[1]] == "--coverage") {
   covr::codecov()
+} else if (args[[1]] == "--arrow") {
+  install.packages("devtools")
+  devtools::install_github("apache/arrow", subdir = "r")
 } else {
   stop("Unsupported arguments")
 }

.travis.yml

@@ -33,31 +33,45 @@ matrix:
       env:
         - SPARK_VERSION="2.3.2"
         - JAVA_VERSION=openjdk8
-    - name: "Livy 0.5.0 (R release, openjdk8)"
+    - name: "Livy 0.5.0 (R release, openjdk8, Spark 2.3.0)"
       r: release
       env:
         - LIVY_VERSION="0.5.0"
+        - SPARK_VERSION="2.3.0"
         - JAVA_VERSION=openjdk8
+    - name: "Arrow (R release, openjdk8)"
+      r: release
+      env:
+        - R_ARROW="true"
+        - JAVA_VERSION=openjdk8
+      addons:
+        apt:
+          sources:
+            - sourceline: deb https://packages.red-data-tools.org/ubuntu/ trusty universe
+              key_url: https://packages.red-data-tools.org/ubuntu/red-data-tools-keyring.gpg
+          packages:
+            - apt-transport-https
+            - lsb-release
+            - libarrow-dev
+            - libarrow-glib-dev
 
 before_install:
   - jdk_switcher use $JAVA_VERSION
   - echo $JAVA_HOME
+  - if [[ $R_ARROW == "true" ]]; then Rscript .travis.R --arrow; fi
 
 script:
   - |
     R CMD build .
     export SPARKLYR_LOG_FILE=/tmp/sparklyr.log
-    if [[ $SPARK_VERSION == "2.3.0" ]]; then
+    if [[ $CODE_COVERAGE == "true" ]]; then
       R CMD check --no-build-vignettes --no-manual --no-tests sparklyr*tar.gz
-      travis_wait 45 Rscript .travis.R --coverage
+      Rscript .travis.R --coverage
     else
-      travis_wait 35 R CMD check --no-build-vignettes --no-manual sparklyr*tar.gz
+      travis_wait 35 R CMD check --no-build-vignettes --no-tests --no-manual sparklyr*tar.gz
+      Rscript .travis.R --testthat
     fi
 
 after_failure:
   - |
     grep -B 10 -A 20 ERROR /tmp/sparklyr.log
-    cd tests
-    export NOT_CRAN=true
-    travis_wait 35 Rscript ../.travis.R --testthat
-    sleep 2

DESCRIPTION

@@ -59,6 +59,7 @@ Suggests:
     mlbench,
     nnet,
     nycflights13,
+    R6,
     RCurl,
     reshape2,
     testthat

NEWS.md

@@ -1,5 +1,11 @@
 # Sparklyr 0.9.3 (unreleased)
 
+- Improved memory use in Livy by using string builders and avoid print
+  backs.
+
+ - `sdf_len()`, `sdf_along()` and `sdf_seq()` default to 32 bit integers
+   but allow support for 64 bits through `bits` parameter.
+
 - Removed `stream_read_jdbc()` and `stream_write_jdbc()` since they are 
   not yet implemented in Spark.
 
@@ -25,6 +31,10 @@
 - Fix new connection RStudio selectors colors when running
   under OS X Mojave.
 
+- Support for launching Livy logs from connection pane.
+
+# Sparklyr 0.9.2
+
 - Removed `overwrite` parameter in `spark_read_table()` (#1698).
 
 - Fix regression preventing using R 3.2 (#1695).

R/arrow_data.R

@@ -0,0 +1,105 @@
+arrow_enabled <- function(sc, object) {
+  enabled <- spark_config_value(sc, "sparklyr.arrow", "package:arrow" %in% search())
+  if (!enabled) {
+    enabled
+  }
+  else {
+    arrow_enabled_object(object)
+  }
+}
+
+arrow_enabled_object <- function(object) {
+  UseMethod("arrow_enabled_object")
+}
+
+arrow_enabled_object.tbl_spark <- function(object) {
+  sdf <- spark_dataframe(object)
+  arrow_enabled_object(sdf)
+}
+
+arrow_enabled_object.spark_jobj <- function(object) {
+  unsupported_expr <- ".Vector|ArrayType|TimestampType|StructType"
+  unsupported <- object %>%
+    sdf_schema() %>%
+    Filter(function(x) grepl(unsupported_expr, x$type), .)
+  enabled <- length(unsupported) == 0
+  if (!enabled) warning("Arrow disabled due to columns: ", paste(names(unsupported), collapse = ", "))
+
+  enabled
+}
+
+arrow_enabled_dataframe_schema <- function(types) {
+  unsupported_expr <- "POSIXct"
+  unsupported <- Filter(function(e) grepl(unsupported_expr , e), types)
+
+  enabled <- length(unsupported) == 0
+  if (!enabled) warning("Arrow disabled due to columns: ", paste(names(unsupported), collapse = ", "))
+
+  enabled
+}
+
+arrow_enabled_object.data.frame <- function(object) {
+  arrow_enabled_dataframe_schema(sapply(object, function(e) class(e)[[1]]))
+}
+
+arrow_batch <- function(df)
+{
+  record_batch <- get("record_batch", envir = as.environment(asNamespace("arrow")))
+  write_record_batch <- get("write_record_batch", envir = as.environment(asNamespace("arrow")))
+
+  record <- record_batch(df)
+  write_record_batch(record, raw())
+}
+
+arrow_read_stream <- function(stream)
+{
+  record_batch_stream_reader <- get("record_batch_stream_reader", envir = as.environment(asNamespace("arrow")))
+  read_record_batch <- get("read_record_batch", envir = as.environment(asNamespace("arrow")))
+
+  reader <- record_batch_stream_reader(stream)
+  record_entry <- read_record_batch(reader)
+
+  entries <- list()
+  while (!is.null(record_entry)) {
+    entries[[length(entries) + 1]] <- tibble::as_tibble(record_entry)
+    record_entry <- read_record_batch(reader)
+  }
+
+  entries
+}
+
+arrow_copy_to <- function(sc, df, parallelism)
+{
+  # replace factors with characters
+  if (any(sapply(df, is.factor))) {
+    df <- dplyr::as_data_frame(lapply(df, function(x) if(is.factor(x)) as.character(x) else x))
+  }
+
+  # serialize to arrow
+  bytes <- arrow_batch(df)
+
+  # create batches data frame
+  batches <- list(bytes)
+
+  # build schema
+  schema <- spark_data_build_types(sc, lapply(df, class))
+
+  # load arrow file in scala
+  rdd <- invoke_static(sc, "sparklyr.ArrowHelper", "javaRddFromBinaryBatches", spark_context(sc), batches, parallelism)
+  sdf <- invoke_static(sc, "sparklyr.ArrowConverters", "toDataFrame", rdd, schema, spark_session(sc))
+
+  sdf
+}
+
+arrow_collect <- function(tbl, ...)
+{
+  sc <- spark_connection(tbl)
+  sdf <- spark_dataframe(tbl)
+  session <- spark_session(sc)
+
+  time_zone <- spark_session(sc) %>% invoke("sessionState") %>% invoke("conf") %>% invoke("sessionLocalTimeZone")
+
+  invoke_static(sc, "sparklyr.ArrowConverters", "toArrowBatchRdd", sdf, session, time_zone) %>%
+    arrow_read_stream() %>%
+    dplyr::bind_rows()
+}

R/config_settings.R

@@ -8,6 +8,7 @@ spark_config_settings <- function() {
     sparklyr.apply.packages = "Configures default value for packages parameter in spark_apply().",
     sparklyr.apply.rlang = "Experimental feature. Turns on improved serialization for spark_apply().",
     sparklyr.apply.schema.infer = "Number of rows collected to infer schema when column types specified in spark_apply().",
+    sparklyr.arrow = "Use Apache Arrow to serialize data?",
     sparklyr.backend.interval = "Total seconds sparklyr will check on a backend operation.",
     sparklyr.backend.timeout = "Total seconds before sparklyr will give up waiting for a backend operation to complete.",
     sparklyr.connect.aftersubmit = "R function to call after spark-submit executes.",

R/connection_viewer.R

@@ -23,6 +23,12 @@ spark_actions <- function(scon) {
           callback = function() {
             utils::browseURL(file.path(scon$master, "ui"))
           }
+        ),
+        "Log" = list(
+          icon = file.path(icons, "spark-log.png"),
+          callback = function() {
+            utils::browseURL(file.path(scon$master, "ui", "session", scon$sessionId, "log"))
+          }
         )
       )
     )

R/core_worker_config.R

@@ -5,6 +5,7 @@ worker_config_serialize <- function(config) {
     spark_config_value(config, "sparklyr.worker.gateway.address", "localhost"),
     if (isTRUE(config$profile)) "TRUE" else "FALSE",
     if (isTRUE(config$schema)) "TRUE" else "FALSE",
+    if (isTRUE(config$arrow)) "TRUE" else "FALSE",
     sep = ";"
   )
 }
@@ -17,6 +18,7 @@ worker_config_deserialize <- function(raw) {
     sparklyr.gateway.port = as.integer(parts[[2]]),
     sparklyr.gateway.address = parts[[3]],
     profile = as.logical(parts[[4]]),
-    schema = as.logical(parts[[5]])
+    schema = as.logical(parts[[5]]),
+    arrow = as.logical(parts[[6]])
   )
 }

R/data_copy.R

@@ -115,12 +115,20 @@ spark_serialize_csv_scala <- function(sc, df, columns, repartition) {
   invoke(hive_context(sc), "createDataFrame", rdd, structType)
 }
 
+spark_serialize_arrow <- function(sc, df, columns, repartition) {
+  arrow_copy_to(
+    sc,
+    df,
+    as.integer(if (repartition <= 0) 1 else repartition)
+  )
+}
+
 spark_data_copy <- function(
   sc,
   df,
   name,
   repartition,
-  serializer = getOption("sparklyr.copy.serializer", "csv_file")) {
+  serializer = NULL) {
 
   if (!is.numeric(repartition)) {
     stop("The repartition parameter must be an integer")
@@ -130,12 +138,19 @@ spark_data_copy <- function(
     stop("Using a local file to copy data is not supported for remote clusters")
   }
 
-  serializer <- ifelse(is.null(serializer),
-                       ifelse(spark_connection_is_local(sc) ||
-                              spark_connection_is_yarn_client(sc),
-                              "csv_file_scala",
-                              "csv_string"),
-                       serializer)
+  serializer <- ifelse(
+                  is.null(serializer),
+                  ifelse(
+                    arrow_enabled(sc, df),
+                    "arrow",
+                    ifelse(
+                      spark_connection_is_local(sc) || spark_connection_is_yarn_client(sc),
+                      "csv_file_scala",
+                      getOption("sparklyr.copy.serializer", "csv_string")
+                    )
+                  ),
+                  serializer
+                )
 
   # Spark unfortunately has a number of issues with '.'s in column names, e.g.
   #
@@ -159,7 +174,8 @@ spark_data_copy <- function(
   serializers <- list(
     "csv_file" = spark_serialize_csv_file,
     "csv_string" = spark_serialize_csv_string,
-    "csv_file_scala" = spark_serialize_csv_scala
+    "csv_file_scala" = spark_serialize_csv_scala,
+    "arrow" = spark_serialize_arrow
   )
 
   df <- serializers[[serializer]](sc, df, columns, repartition)

R/dbi_spark_table.R

@@ -30,7 +30,7 @@ setMethod("dbReadTable", c("spark_connection", "character"),
 
 
 setMethod("dbListTables", "spark_connection", function(conn) {
-  df <- df_from_sql(conn, "SHOW TABLES")
+  df <- df_from_sql(conn, "SHOW TABLES", arrow = FALSE)
 
   tableNames <- df$tableName
   filtered <- grep("^sparklyr_tmp_", tableNames, invert = TRUE, value = TRUE)

R/install_spark_versions.R

@@ -124,7 +124,7 @@ spark_versions <- function(latest = TRUE) {
              if (dir.exists(maybeDir)) {
                fileName <- basename(maybeDir)
                m <- regmatches(fileName, regexec(spark_versions_file_pattern(), fileName))[[1]]
-               if (length(m) > 2) list(spark = m[[2]], hadoop = m[[3]]) else NULL
+               if (length(m) > 2) list(spark = m[[2]], hadoop = m[[3]], pattern = fileName) else NULL
              }
            })
     ),
@@ -134,7 +134,6 @@ spark_versions <- function(latest = TRUE) {
 
       newRow <- c(row, installed = TRUE)
       newRow$base <- ""
-      newRow$pattern <- ""
       newRow$download <- ""
       newRow$default <- FALSE
       newRow$hadoop_default <- FALSE