[TEST] Run all Travis tests with Arrow enabled

.travis.R

@@ -10,6 +10,9 @@ if (length(args) == 0) {
   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

@@ -14,30 +14,26 @@ r_packages:
 
 matrix:
   include:
-    - name: "Spark 1.6.3 (R 3.2, openjdk7)"
-      r: 3.2
-      env:
-        - SPARK_VERSION="1.6.3"
-        - JAVA_VERSION=openjdk7
-    - name: "Spark 2.2.1 (R oldrel, oraclejdk8)"
-      r: oldrel
-      env:
-        - SPARK_VERSION="2.2.1"
-        - JAVA_VERSION=oraclejdk8
-    - name: "Spark 2.3.1 (R release, openjdk8)"
-      r: release
-      env:
-        - SPARK_VERSION="2.3.1"
-        - JAVA_VERSION=openjdk8
-    - name: "Livy 0.5.0 (R release, openjdk8)"
+    - name: "Arrow (R release, openjdk8)"
       r: release
       env:
-        - LIVY_VERSION="0.5.0"
+        - 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:
   - |

NEWS.md

@@ -1,5 +1,7 @@
 # Sparklyr 0.9.2 (unreleased)
 
+- Support for launching Livy logs from connection pane.
+
 - Removed `overwrite` parameter in `spark_read_table()` (#1698).
 
 - Fix regression preventing using R 3.2 (#1695).

R/arrow_data.R

@@ -0,0 +1,65 @@
+arrow_enabled <- function(sc) {
+  spark_config_value(sc, "sparklyr.arrow", "package:arrow" %in% search())
+}
+
+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 (!record_entry$is_null()) {
+    entries[[length(entries) + 1]] <- tibble::as_tibble(record_entry)
+    record_entry <- read_record_batch(reader)
+  }
+
+  entries
+}
+
+arrow_copy_to <- function(sc, df, parallelism = 8L, serializer = "arrow")
+{
+  # 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_invoke.R

@@ -111,7 +111,7 @@ core_invoke_method <- function(sc, static, object, method, ...)
   backend <- core_invoke_socket(sc)
   connection_name <- core_invoke_socket_name(sc)
 
-  if (!identical(object, "Handler")) {
+  if (!identical(object, "Handler") && getOption("sparklyr.connection.cancellable", TRUE)) {
     # if connection still running, sync to valid state
     if (identical(sc$state$status[[connection_name]], "running"))
       core_invoke_sync(sc)

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,16 @@ 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)
+}
+
 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 +134,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),
+                    "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 +170,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

R/livy_connection.R

@@ -697,6 +697,9 @@ livy_load_scala_sources <- function(sc) {
     "serializer.scala",
     "stream.scala",
     "repartition.scala",
+    "arrowhelper.scala",
+    "arrowbatchstreamwriter.scala",
+    "arrowconverters.scala",
     "applyutils.scala",
     "classutils.scala",
     "fileutils.scala",
@@ -762,15 +765,19 @@ initialize_connection.livy_connection <- function(sc) {
   tryCatch({
     livy_load_scala_sources(sc)
 
-    session <- NULL
-    sc$state$spark_context <- tryCatch({
-      session <<- invoke_static(
+    session <- tryCatch({
+      invoke_static(
         sc,
         "org.apache.spark.sql.SparkSession",
         "builder"
       ) %>%
         invoke("getOrCreate")
+    },
+    error = function(e) {
+      NULL
+    })
 
+    sc$state$spark_context <- tryCatch({
       invoke(session, "sparkContext")
     },
     error = function(e) {

R/livy_sources.R

@@ -11,6 +11,9 @@ livy_sources_included <- function() {
     "/repartition\\.scala",
     "/tracker\\.scala",
     "/livyutils\\.scala",
+    "/arrowhelper\\.scala",
+    "/arrowbatchstreamwriter\\.scala",
+    "/arrowconverters\\.scala",
     "/applyutils\\.scala",
     "/classutils\\.scala",
     "/fileutils\\.scala",

R/sdf_sql.R

@@ -1,11 +1,11 @@
 
-df_from_sql <- function(sc, sql) {
+df_from_sql <- function(sc, sql, arrow = TRUE) {
   sdf <- invoke(hive_context(sc), "sql", as.character(sql))
-  df_from_sdf(sc, sdf)
+  df_from_sdf(sc, sdf, arrow = arrow)
 }
 
-df_from_sdf <- function(sc, sdf, take = -1) {
-  sdf_collect(sdf)
+df_from_sdf <- function(sc, sdf, take = -1, arrow = TRUE) {
+  sdf_collect(sdf, arrow = arrow)
 }
 
 #' Spark DataFrame from SQL

R/sdf_wrapper.R

@@ -93,10 +93,13 @@ sdf_read_column <- function(x, column) {
 #'
 #' @export
 sdf_collect <- function(object, ...) {
+  args <- list(...)
   sc <- spark_connection(object)
 
   if (sdf_is_streaming(object))
     sdf_collect_stream(object, ...)
+  else if (arrow_enabled(sc) && !identical(args$arrow, FALSE))
+    arrow_collect(object, ...)
   else
     sdf_collect_static(object, ...)
 }