OSCL-LXR spark-3.0.0/​R/​pkg/​R/​utils.R	Source navigation Diff markup Identifier search General search
	Version: spark-3.0.0 hadoop-3.2.0-src hadoop-3.3.0-src spark-2.4.7 linux-4.15.13 Revert   Change

 #
 # Licensed to the Apache Software Foundation (ASF) under one or more
 # contributor license agreements.  See the NOTICE file distributed with
 # this work for additional information regarding copyright ownership.
 # The ASF licenses this file to You under the Apache License, Version 2.0
 # (the "License"); you may not use this file except in compliance with
 # the License.  You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
 
 # Utilities and Helpers
 
 # Given a JList<T>, returns an R list containing the same elements, the number
 # of which is optionally upper bounded by `logicalUpperBound` (by default,
 # return all elements).  Takes care of deserializations and type conversions.
 convertJListToRList <- function(jList, flatten, logicalUpperBound = NULL,
   serializedMode = "byte") {
   arrSize <- callJMethod(jList, "size")
 
   # Datasets with serializedMode == "string" (such as an RDD directly generated by textFile()):
   # each partition is not dense-packed into one Array[Byte], and `arrSize`
   # here corresponds to number of logical elements. Thus we can prune here.
   if (serializedMode == "string" && !is.null(logicalUpperBound)) {
     arrSize <- min(arrSize, logicalUpperBound)
   }
 
   results <- if (arrSize > 0) {
     lapply(0 : (arrSize - 1),
           function(index) {
             obj <- callJMethod(jList, "get", as.integer(index))
 
             # Assume it is either an R object or a Java obj ref.
             if (inherits(obj, "jobj")) {
               if (isInstanceOf(obj, "scala.Tuple2")) {
                 # JavaPairRDD[Array[Byte], Array[Byte]].
 
                 keyBytes <- callJMethod(obj, "_1")
                 valBytes <- callJMethod(obj, "_2")
                 res <- list(unserialize(keyBytes),
                   unserialize(valBytes))
               } else {
                 stop("utils.R: convertJListToRList only supports ",
                   "RDD[Array[Byte]] and ",
                   "JavaPairRDD[Array[Byte], Array[Byte]] for now")
               }
             } else {
               if (inherits(obj, "raw")) {
                 if (serializedMode == "byte") {
                   # RDD[Array[Byte]]. `obj` is a whole partition.
                   res <- unserialize(obj)
                   # For serialized datasets, `obj` (and `rRaw`) here corresponds to
                   # one whole partition dense-packed together. We deserialize the
                   # whole partition first, then cap the number of elements to be returned.
                 } else if (serializedMode == "row") {
                   res <- readRowList(obj)
                   # For DataFrames that have been converted to RRDDs, we call readRowList
                   # which will read in each row of the RRDD as a list and deserialize
                   # each element.
                   flatten <<- FALSE
                   # Use global assignment to change the flatten flag. This means
                   # we don't have to worry about the default argument in other functions
                   # e.g. collect
                 }
                 # TODO: is it possible to distinguish element boundary so that we can
                 # unserialize only what we need?
                 if (!is.null(logicalUpperBound)) {
                   res <- head(res, n = logicalUpperBound)
                 }
               } else {
                 # obj is of a primitive Java type, is simplified to R's
                 # corresponding type.
                 res <- list(obj)
               }
             }
             res
           })
   } else {
     list()
   }
 
   if (flatten) {
     as.list(unlist(results, recursive = FALSE))
   } else {
     as.list(results)
   }
 }
 
 # Returns TRUE if `name` refers to an RDD in the given environment `env`
 isRDD <- function(name, env) {
   obj <- get(name, envir = env)
   inherits(obj, "RDD")
 }
 
 #' Compute the hashCode of an object
 #'
 #' Java-style function to compute the hashCode for the given object. Returns
 #' an integer value.
 #'
 #' @details
 #' This only works for integer, numeric and character types right now.
 #'
 #' @param key the object to be hashed
 #' @return the hash code as an integer
 #' @examples
 #'\dontrun{
 #' hashCode(1L) # 1
 #' hashCode(1.0) # 1072693248
 #' hashCode("1") # 49
 #'}
 #' @note hashCode since 1.4.0
 hashCode <- function(key) {
   if (class(key) == "integer") {
     as.integer(key[[1]])
   } else if (class(key) == "numeric") {
     # Convert the double to long and then calculate the hash code
     rawVec <- writeBin(key[[1]], con = raw())
     intBits <- packBits(rawToBits(rawVec), "integer")
     as.integer(bitwXor(intBits[2], intBits[1]))
   } else if (class(key) == "character") {
     # TODO: SPARK-7839 means we might not have the native library available
     n <- nchar(key)
     if (n == 0) {
       0L
     } else {
       asciiVals <- sapply(charToRaw(key), function(x) { strtoi(x, 16L) })
       hashC <- 0
       for (k in seq_len(length(asciiVals))) {
         hashC <- mult31AndAdd(hashC, asciiVals[k])
       }
       as.integer(hashC)
     }
   } else {
     warning("Could not hash object, returning 0")
     as.integer(0)
   }
 }
 
 # Helper function used to wrap a 'numeric' value to integer bounds.
 # Useful for implementing C-like integer arithmetic
 wrapInt <- function(value) {
   if (value > .Machine$integer.max) {
     value <- value - 2 * .Machine$integer.max - 2
   } else if (value < -1 * .Machine$integer.max) {
     value <- 2 * .Machine$integer.max + value + 2
   }
   value
 }
 
 # Multiply `val` by 31 and add `addVal` to the result. Ensures that
 # integer-overflows are handled at every step.
 #
 # TODO: this function does not handle integer overflow well
 mult31AndAdd <- function(val, addVal) {
   vec <- c(bitwShiftL(val, c(4, 3, 2, 1, 0)), addVal)
   vec[is.na(vec)] <- 0
   Reduce(function(a, b) {
           wrapInt(as.numeric(a) + as.numeric(b))
          },
          vec)
 }
 
 # Create a new RDD with serializedMode == "byte".
 # Return itself if already in "byte" format.
 serializeToBytes <- function(rdd) {
   if (!inherits(rdd, "RDD")) {
     stop("Argument 'rdd' is not an RDD type.")
   }
   if (getSerializedMode(rdd) != "byte") {
     ser.rdd <- lapply(rdd, function(x) { x })
     return(ser.rdd)
   } else {
     return(rdd)
   }
 }
 
 # Create a new RDD with serializedMode == "string".
 # Return itself if already in "string" format.
 serializeToString <- function(rdd) {
   if (!inherits(rdd, "RDD")) {
     stop("Argument 'rdd' is not an RDD type.")
   }
   if (getSerializedMode(rdd) != "string") {
     ser.rdd <- lapply(rdd, function(x) { toString(x) })
     # force it to create jrdd using "string"
     getJRDD(ser.rdd, serializedMode = "string")
     return(ser.rdd)
   } else {
     return(rdd)
   }
 }
 
 # Fast append to list by using an accumulator.
 # http://stackoverflow.com/questions/17046336/here-we-go-again-append-an-element-to-a-list-in-r
 #
 # The accumulator should has three fields size, counter and data.
 # This function amortizes the allocation cost by doubling
 # the size of the list every time it fills up.
 addItemToAccumulator <- function(acc, item) {
   if (acc$counter == acc$size) {
     acc$size <- acc$size * 2
     length(acc$data) <- acc$size
   }
   acc$counter <- acc$counter + 1
   acc$data[[acc$counter]] <- item
 }
 
 initAccumulator <- function() {
   acc <- new.env()
   acc$counter <- 0
   acc$data <- list(NULL)
   acc$size <- 1
   acc
 }
 
 # Utility function to sort a list of key value pairs
 # Used in unit tests
 sortKeyValueList <- function(kv_list, decreasing = FALSE) {
   keys <- sapply(kv_list, function(x) x[[1]])
   kv_list[order(keys, decreasing = decreasing)]
 }
 
 # Utility function to generate compact R lists from grouped rdd
 # Used in Join-family functions
 # param:
 #   tagged_list R list generated via groupByKey with tags(1L, 2L, ...)
 #   cnull Boolean list where each element determines whether the corresponding list should
 #         be converted to list(NULL)
 genCompactLists <- function(tagged_list, cnull) {
   len <- length(tagged_list)
   lists <- list(vector("list", len), vector("list", len))
   index <- list(1, 1)
 
   for (x in tagged_list) {
     tag <- x[[1]]
     idx <- index[[tag]]
     lists[[tag]][[idx]] <- x[[2]]
     index[[tag]] <- idx + 1
   }
 
   len <- lapply(index, function(x) x - 1)
   for (i in (1:2)) {
     if (cnull[[i]] && len[[i]] == 0) {
       lists[[i]] <- list(NULL)
     } else {
       length(lists[[i]]) <- len[[i]]
     }
   }
 
   lists
 }
 
 # Utility function to merge compact R lists
 # Used in Join-family functions
 # param:
 #   left/right Two compact lists ready for Cartesian product
 mergeCompactLists <- function(left, right) {
   result <- list()
   length(result) <- length(left) * length(right)
   index <- 1
   for (i in left) {
     for (j in right) {
       result[[index]] <- list(i, j)
       index <- index + 1
     }
   }
   result
 }
 
 # Utility function to wrapper above two operations
 # Used in Join-family functions
 # param (same as genCompactLists):
 #   tagged_list R list generated via groupByKey with tags(1L, 2L, ...)
 #   cnull Boolean list where each element determines whether the corresponding list should
 #         be converted to list(NULL)
 joinTaggedList <- function(tagged_list, cnull) {
   lists <- genCompactLists(tagged_list, cnull)
   mergeCompactLists(lists[[1]], lists[[2]])
 }
 
 # Utility function to reduce a key-value list with predicate
 # Used in *ByKey functions
 # param
 #   pair key-value pair
 #   keys/vals env of key/value with hashes
 #   updateOrCreatePred predicate function
 #   updateFn update or merge function for existing pair, similar with `mergeVal` @combineByKey
 #   createFn create function for new pair, similar with `createCombiner` @combinebykey
 updateOrCreatePair <- function(pair, keys, vals, updateOrCreatePred, updateFn, createFn) {
   # assume hashVal bind to `$hash`, key/val with index 1/2
   hashVal <- pair$hash
   key <- pair[[1]]
   val <- pair[[2]]
   if (updateOrCreatePred(pair)) {
     assign(hashVal, do.call(updateFn, list(get(hashVal, envir = vals), val)), envir = vals)
   } else {
     assign(hashVal, do.call(createFn, list(val)), envir = vals)
     assign(hashVal, key, envir = keys)
   }
 }
 
 # Utility function to convert key&values envs into key-val list
 convertEnvsToList <- function(keys, vals) {
   lapply(ls(keys),
          function(name) {
            list(keys[[name]], vals[[name]])
          })
 }
 
 # Utility function to merge 2 environments with the second overriding values in the first
 # env1 is changed in place
 overrideEnvs <- function(env1, env2) {
   lapply(ls(env2),
          function(name) {
            env1[[name]] <- env2[[name]]
          })
 }
 
 # Utility function to capture the varargs into environment object
 varargsToEnv <- function(...) {
   # Based on http://stackoverflow.com/a/3057419/4577954
   pairs <- list(...)
   env <- new.env()
   for (name in names(pairs)) {
     env[[name]] <- pairs[[name]]
   }
   env
 }
 
 # Utility function to capture the varargs into environment object but all values are converted
 # into string.
 varargsToStrEnv <- function(...) {
   pairs <- list(...)
   nameList <- names(pairs)
   env <- new.env()
   ignoredNames <- list()
 
   if (is.null(nameList)) {
     # When all arguments are not named, names(..) returns NULL.
     ignoredNames <- pairs
   } else {
     for (i in seq_along(pairs)) {
       name <- nameList[i]
       value <- pairs[i]
       if (identical(name, "")) {
         # When some of arguments are not named, name is "".
         ignoredNames <- append(ignoredNames, value)
       } else {
         value <- pairs[[name]]
         if (!(is.logical(value) || is.numeric(value) || is.character(value) || is.null(value))) {
           stop("Unsupported type for ", name, " : ", toString(class(value)), ". ",
                "Supported types are logical, numeric, character and NULL.", call. = FALSE)
         }
         if (is.logical(value)) {
           env[[name]] <- tolower(as.character(value))
         } else if (is.null(value)) {
           env[[name]] <- value
         } else {
           env[[name]] <- as.character(value)
         }
       }
     }
   }
 
   if (length(ignoredNames) != 0) {
     warning("Unnamed arguments ignored: ", toString(ignoredNames), ".", call. = FALSE)
   }
   env
 }
 
 getStorageLevel <- function(newLevel = c("DISK_ONLY",
                                          "DISK_ONLY_2",
                                          "MEMORY_AND_DISK",
                                          "MEMORY_AND_DISK_2",
                                          "MEMORY_AND_DISK_SER",
                                          "MEMORY_AND_DISK_SER_2",
                                          "MEMORY_ONLY",
                                          "MEMORY_ONLY_2",
                                          "MEMORY_ONLY_SER",
                                          "MEMORY_ONLY_SER_2",
                                          "OFF_HEAP")) {
   match.arg(newLevel)
   storageLevelClass <- "org.apache.spark.storage.StorageLevel"
   storageLevel <- switch(newLevel,
                          "DISK_ONLY" = callJStatic(storageLevelClass, "DISK_ONLY"),
                          "DISK_ONLY_2" = callJStatic(storageLevelClass, "DISK_ONLY_2"),
                          "MEMORY_AND_DISK" = callJStatic(storageLevelClass, "MEMORY_AND_DISK"),
                          "MEMORY_AND_DISK_2" = callJStatic(storageLevelClass, "MEMORY_AND_DISK_2"),
                          "MEMORY_AND_DISK_SER" = callJStatic(storageLevelClass,
                                                              "MEMORY_AND_DISK_SER"),
                          "MEMORY_AND_DISK_SER_2" = callJStatic(storageLevelClass,
                                                                "MEMORY_AND_DISK_SER_2"),
                          "MEMORY_ONLY" = callJStatic(storageLevelClass, "MEMORY_ONLY"),
                          "MEMORY_ONLY_2" = callJStatic(storageLevelClass, "MEMORY_ONLY_2"),
                          "MEMORY_ONLY_SER" = callJStatic(storageLevelClass, "MEMORY_ONLY_SER"),
                          "MEMORY_ONLY_SER_2" = callJStatic(storageLevelClass, "MEMORY_ONLY_SER_2"),
                          "OFF_HEAP" = callJStatic(storageLevelClass, "OFF_HEAP"))
 }
 
 storageLevelToString <- function(levelObj) {
   useDisk <- callJMethod(levelObj, "useDisk")
   useMemory <- callJMethod(levelObj, "useMemory")
   useOffHeap <- callJMethod(levelObj, "useOffHeap")
   deserialized <- callJMethod(levelObj, "deserialized")
   replication <- callJMethod(levelObj, "replication")
   shortName <- if (!useDisk && !useMemory && !useOffHeap && !deserialized && replication == 1) {
     "NONE"
   } else if (useDisk && !useMemory && !useOffHeap && !deserialized && replication == 1) {
     "DISK_ONLY"
   } else if (useDisk && !useMemory && !useOffHeap && !deserialized && replication == 2) {
     "DISK_ONLY_2"
   } else if (!useDisk && useMemory && !useOffHeap && deserialized && replication == 1) {
     "MEMORY_ONLY"
   } else if (!useDisk && useMemory && !useOffHeap && deserialized && replication == 2) {
     "MEMORY_ONLY_2"
   } else if (!useDisk && useMemory && !useOffHeap && !deserialized && replication == 1) {
     "MEMORY_ONLY_SER"
   } else if (!useDisk && useMemory && !useOffHeap && !deserialized && replication == 2) {
     "MEMORY_ONLY_SER_2"
   } else if (useDisk && useMemory && !useOffHeap && deserialized && replication == 1) {
     "MEMORY_AND_DISK"
   } else if (useDisk && useMemory && !useOffHeap && deserialized && replication == 2) {
     "MEMORY_AND_DISK_2"
   } else if (useDisk && useMemory && !useOffHeap && !deserialized && replication == 1) {
     "MEMORY_AND_DISK_SER"
   } else if (useDisk && useMemory && !useOffHeap && !deserialized && replication == 2) {
     "MEMORY_AND_DISK_SER_2"
   } else if (useDisk && useMemory && useOffHeap && !deserialized && replication == 1) {
     "OFF_HEAP"
   } else {
     NULL
   }
   fullInfo <- callJMethod(levelObj, "toString")
   if (is.null(shortName)) {
     fullInfo
   } else {
     paste(shortName, "-", fullInfo)
   }
 }
 
 # Utility function for functions where an argument needs to be integer but we want to allow
 # the user to type (for example) `5` instead of `5L` to avoid a confusing error message.
 numToInt <- function(num) {
   if (as.integer(num) != num) {
     warning("Coercing ", as.list(sys.call())[[2L]], " to integer.")
   }
   as.integer(num)
 }
 
 # Utility function to recursively traverse the Abstract Syntax Tree (AST) of a
 # user defined function (UDF), and to examine variables in the UDF to decide
 # if their values should be included in the new function environment.
 # param
 #   node The current AST node in the traversal.
 #   oldEnv The original function environment.
 #   defVars An Accumulator of variables names defined in the function's calling environment,
 #           including function argument and local variable names.
 #   checkedFunc An environment of function objects examined during cleanClosure. It can
 #               be considered as a "name"-to-"list of functions" mapping.
 #   newEnv A new function environment to store necessary function dependencies, an output argument.
 processClosure <- function(node, oldEnv, defVars, checkedFuncs, newEnv) {
   nodeLen <- length(node)
 
   if (nodeLen > 1 && typeof(node) == "language") {
     # Recursive case: current AST node is an internal node, check for its children.
     if (length(node[[1]]) > 1) {
       for (i in 1:nodeLen) {
         processClosure(node[[i]], oldEnv, defVars, checkedFuncs, newEnv)
       }
     } else {
       # if node[[1]] is length of 1, check for some R special functions.
       nodeChar <- as.character(node[[1]])
       if (nodeChar == "{" || nodeChar == "(") {
         # Skip start symbol.
         for (i in 2:nodeLen) {
           processClosure(node[[i]], oldEnv, defVars, checkedFuncs, newEnv)
         }
       } else if (nodeChar == "<-" || nodeChar == "=" ||
                    nodeChar == "<<-") {
         # Assignment Ops.
         defVar <- node[[2]]
         if (length(defVar) == 1 && typeof(defVar) == "symbol") {
           # Add the defined variable name into defVars.
           addItemToAccumulator(defVars, as.character(defVar))
         } else {
           processClosure(node[[2]], oldEnv, defVars, checkedFuncs, newEnv)
         }
         for (i in 3:nodeLen) {
           processClosure(node[[i]], oldEnv, defVars, checkedFuncs, newEnv)
         }
       } else if (nodeChar == "function") {
         # Function definition.
         # Add parameter names.
         newArgs <- names(node[[2]])
         lapply(newArgs, function(arg) { addItemToAccumulator(defVars, arg) })
         for (i in 3:nodeLen) {
           processClosure(node[[i]], oldEnv, defVars, checkedFuncs, newEnv)
         }
       } else if (nodeChar == "$") {
         # Skip the field.
         processClosure(node[[2]], oldEnv, defVars, checkedFuncs, newEnv)
       } else if (nodeChar == "::" || nodeChar == ":::") {
         processClosure(node[[3]], oldEnv, defVars, checkedFuncs, newEnv)
       } else {
         for (i in 1:nodeLen) {
           processClosure(node[[i]], oldEnv, defVars, checkedFuncs, newEnv)
         }
       }
     }
   } else if (nodeLen == 1 &&
                (typeof(node) == "symbol" || typeof(node) == "language")) {
     # Base case: current AST node is a leaf node and a symbol or a function call.
     nodeChar <- as.character(node)
     if (!nodeChar %in% defVars$data) {
       # Not a function parameter or local variable.
       func.env <- oldEnv
       topEnv <- parent.env(.GlobalEnv)
       # Search in function environment, and function's enclosing environments
       # up to global environment. There is no need to look into package environments
       # above the global or namespace environment that is not SparkR below the global,
       # as they are assumed to be loaded on workers.
       while (!identical(func.env, topEnv)) {
         # Namespaces other than "SparkR" will not be searched.
         if (!isNamespace(func.env) ||
             (getNamespaceName(func.env) == "SparkR" &&
                !(nodeChar %in% getNamespaceExports("SparkR")))) {
           # Only include SparkR internals.
 
           # Set parameter 'inherits' to FALSE since we do not need to search in
           # attached package environments.
           if (tryCatch(exists(nodeChar, envir = func.env, inherits = FALSE),
                        error = function(e) { FALSE })) {
             obj <- get(nodeChar, envir = func.env, inherits = FALSE)
             if (is.function(obj)) {
               # If the node is a function call.
               funcList <- mget(nodeChar, envir = checkedFuncs, inherits = F,
                                ifnotfound = list(list(NULL)))[[1]]
               found <- sapply(funcList, function(func) {
                 ifelse(
                   identical(func, obj) &&
                     # Also check if the parent environment is identical to current parent
                     identical(parent.env(environment(func)), func.env),
                   TRUE, FALSE)
               })
               if (sum(found) > 0) {
                 # If function has been examined ignore
                 break
               }
               # Function has not been examined, record it and recursively clean its closure.
               assign(nodeChar,
                      if (is.null(funcList[[1]])) {
                        list(obj)
                      } else {
                        append(funcList, obj)
                      },
                      envir = checkedFuncs)
               obj <- cleanClosure(obj, checkedFuncs)
             }
             assign(nodeChar, obj, envir = newEnv)
             break
           }
         }
 
         # Continue to search in enclosure.
         func.env <- parent.env(func.env)
       }
     }
   }
 }
 
 # Utility function to get user defined function (UDF) dependencies (closure).
 # More specifically, this function captures the values of free variables defined
 # outside a UDF, and stores them in the function's environment.
 # param
 #   func A function whose closure needs to be captured.
 #   checkedFunc An environment of function objects examined during cleanClosure. It can be
 #               considered as a "name"-to-"list of functions" mapping.
 # return value
 #   a new version of func that has a correct environment (closure).
 cleanClosure <- function(func, checkedFuncs = new.env()) {
   if (is.function(func)) {
     newEnv <- new.env(parent = .GlobalEnv)
     func.body <- body(func)
     oldEnv <- environment(func)
     # defVars is an Accumulator of variables names defined in the function's calling
     # environment. First, function's arguments are added to defVars.
     defVars <- initAccumulator()
     argNames <- names(as.list(args(func)))
     for (i in 1:(length(argNames) - 1)) {
       # Remove the ending NULL in pairlist.
       addItemToAccumulator(defVars, argNames[i])
     }
     # Recursively examine variables in the function body.
     processClosure(func.body, oldEnv, defVars, checkedFuncs, newEnv)
     environment(func) <- newEnv
   }
   func
 }
 
 # Append partition lengths to each partition in two input RDDs if needed.
 # param
 #   x An RDD.
 #   Other An RDD.
 # return value
 #   A list of two result RDDs.
 appendPartitionLengths <- function(x, other) {
   if (getSerializedMode(x) != getSerializedMode(other) ||
       getSerializedMode(x) == "byte") {
     # Append the number of elements in each partition to that partition so that we can later
     # know the boundary of elements from x and other.
     #
     # Note that this appending also serves the purpose of reserialization, because even if
     # any RDD is serialized, we need to reserialize it to make sure its partitions are encoded
     # as a single byte array. For example, partitions of an RDD generated from partitionBy()
     # may be encoded as multiple byte arrays.
     appendLength <- function(part) {
       len <- length(part)
       part[[len + 1]] <- len + 1
       part
     }
     x <- lapplyPartition(x, appendLength)
     other <- lapplyPartition(other, appendLength)
   }
   list(x, other)
 }
 
 # Perform zip or cartesian between elements from two RDDs in each partition
 # param
 #   rdd An RDD.
 #   zip A boolean flag indicating this call is for zip operation or not.
 # return value
 #   A result RDD.
 mergePartitions <- function(rdd, zip) {
   serializerMode <- getSerializedMode(rdd)
   partitionFunc <- function(partIndex, part) {
     len <- length(part)
     if (len > 0) {
       if (serializerMode == "byte") {
         lengthOfValues <- part[[len]]
         lengthOfKeys <- part[[len - lengthOfValues]]
         stopifnot(len == lengthOfKeys + lengthOfValues)
 
         # For zip operation, check if corresponding partitions
         # of both RDDs have the same number of elements.
         if (zip && lengthOfKeys != lengthOfValues) {
           stop("Can only zip RDDs with same number of elements ",
                "in each pair of corresponding partitions.")
         }
 
         if (lengthOfKeys > 1) {
           keys <- part[1 : (lengthOfKeys - 1)]
         } else {
           keys <- list()
         }
         if (lengthOfValues > 1) {
           values <- part[(lengthOfKeys + 1) : (len - 1)]
         } else {
           values <- list()
         }
 
         if (!zip) {
           return(mergeCompactLists(keys, values))
         }
       } else {
         keys <- part[c(TRUE, FALSE)]
         values <- part[c(FALSE, TRUE)]
       }
       mapply(
         function(k, v) { list(k, v) },
         keys,
         values,
         SIMPLIFY = FALSE,
         USE.NAMES = FALSE)
     } else {
       part
     }
   }
 
   PipelinedRDD(rdd, partitionFunc)
 }
 
 # Convert a named list to struct so that
 # SerDe won't confuse between a normal named list and struct
 listToStruct <- function(list) {
   stopifnot(class(list) == "list")
   stopifnot(!is.null(names(list)))
   class(list) <- "struct"
   list
 }
 
 # Convert a struct to a named list
 structToList <- function(struct) {
   stopifnot(class(list) == "struct")
 
   class(struct) <- "list"
   struct
 }
 
 # Convert a named list to an environment to be passed to JVM
 convertNamedListToEnv <- function(namedList) {
   # Make sure each item in the list has a name
   names <- names(namedList)
   stopifnot(
     if (is.null(names)) {
       length(namedList) == 0
     } else {
       !any(is.na(names))
     })
 
   env <- new.env()
   for (name in names) {
     env[[name]] <- namedList[[name]]
   }
   env
 }
 
 # Assign a new environment for attach() and with() methods
 assignNewEnv <- function(data) {
   stopifnot(class(data) == "SparkDataFrame")
   cols <- columns(data)
   stopifnot(length(cols) > 0)
 
   env <- new.env()
   for (i in seq_len(length(cols))) {
     assign(x = cols[i], value = data[, cols[i], drop = F], envir = env)
   }
   env
 }
 
 # Utility function to split by ',' and whitespace, remove empty tokens
 splitString <- function(input) {
   Filter(nzchar, unlist(strsplit(input, ",|\\s")))
 }
 
 varargsToJProperties <- function(...) {
   pairs <- list(...)
   props <- newJObject("java.util.Properties")
   if (length(pairs) > 0) {
     lapply(ls(pairs), function(k) {
       callJMethod(props, "setProperty", as.character(k), as.character(pairs[[k]]))
     })
   }
   props
 }
 
 launchScript <- function(script, combinedArgs, wait = FALSE, stdout = "", stderr = "") {
   if (.Platform$OS.type == "windows") {
     scriptWithArgs <- paste(script, combinedArgs, sep = " ")
     # on Windows, intern = F seems to mean output to the console. (documentation on this is missing)
     shell(scriptWithArgs, translate = TRUE, wait = wait, intern = wait)
   } else {
     # http://stat.ethz.ch/R-manual/R-devel/library/base/html/system2.html
     # stdout = F means discard output
     # stdout = "" means to its console (default)
     # Note that the console of this child process might not be the same as the running R process.
     system2(script, combinedArgs, stdout = stdout, wait = wait, stderr = stderr)
   }
 }
 
 getSparkContext <- function() {
   if (!exists(".sparkRjsc", envir = .sparkREnv)) {
     stop("SparkR has not been initialized. Please call sparkR.session()")
   }
   sc <- get(".sparkRjsc", envir = .sparkREnv)
   sc
 }
 
 isMasterLocal <- function(master) {
   grepl("^local(\\[([0-9]+|\\*)\\])?$", master, perl = TRUE)
 }
 
 isClientMode <- function(master) {
   grepl("([a-z]+)-client$", master, perl = TRUE)
 }
 
 isSparkRShell <- function() {
   grepl(".*shell\\.R$", Sys.getenv("R_PROFILE_USER"), perl = TRUE)
 }
 
 # Works identically with `callJStatic(...)` but throws a pretty formatted exception.
 handledCallJStatic <- function(cls, method, ...) {
   result <- tryCatch(callJStatic(cls, method, ...),
                      error = function(e) {
                        captureJVMException(e, method)
                      })
   result
 }
 
 # Works identically with `callJMethod(...)` but throws a pretty formatted exception.
 handledCallJMethod <- function(obj, method, ...) {
   result <- tryCatch(callJMethod(obj, method, ...),
                      error = function(e) {
                        captureJVMException(e, method)
                      })
   result
 }
 
 captureJVMException <- function(e, method) {
   rawmsg <- as.character(e)
   if (any(grepl("^Error in .*?: ", rawmsg))) {
     # If the exception message starts with "Error in ...", this is possibly
     # "Error in invokeJava(...)". Here, it replaces the characters to
     # `paste("Error in", method, ":")` in order to identify which function
     # was called in JVM side.
     stacktrace <- strsplit(rawmsg, "Error in .*?: ")[[1]]
     rmsg <- paste("Error in", method, ":")
     stacktrace <- paste(rmsg[1], stacktrace[2])
   } else {
     # Otherwise, do not convert the error message just in case.
     stacktrace <- rawmsg
   }
 
   # StreamingQueryException could wrap an IllegalArgumentException, so look for that first
   if (any(grepl("org.apache.spark.sql.streaming.StreamingQueryException: ",
                 stacktrace, fixed = TRUE))) {
     msg <- strsplit(stacktrace, "org.apache.spark.sql.streaming.StreamingQueryException: ",
                     fixed = TRUE)[[1]]
     # Extract "Error in ..." message.
     rmsg <- msg[1]
     # Extract the first message of JVM exception.
     first <- strsplit(msg[2], "\r?\n\tat")[[1]][1]
     stop(rmsg, "streaming query error - ", first, call. = FALSE)
   } else if (any(grepl("java.lang.IllegalArgumentException: ", stacktrace, fixed = TRUE))) {
     msg <- strsplit(stacktrace, "java.lang.IllegalArgumentException: ", fixed = TRUE)[[1]]
     # Extract "Error in ..." message.
     rmsg <- msg[1]
     # Extract the first message of JVM exception.
     first <- strsplit(msg[2], "\r?\n\tat")[[1]][1]
     stop(rmsg, "illegal argument - ", first, call. = FALSE)
   } else if (any(grepl("org.apache.spark.sql.AnalysisException: ", stacktrace, fixed = TRUE))) {
     msg <- strsplit(stacktrace, "org.apache.spark.sql.AnalysisException: ", fixed = TRUE)[[1]]
     # Extract "Error in ..." message.
     rmsg <- msg[1]
     # Extract the first message of JVM exception.
     first <- strsplit(msg[2], "\r?\n\tat")[[1]][1]
     stop(rmsg, "analysis error - ", first, call. = FALSE)
   } else
     if (any(grepl("org.apache.spark.sql.catalyst.analysis.NoSuchDatabaseException: ",
                   stacktrace, fixed = TRUE))) {
     msg <- strsplit(stacktrace, "org.apache.spark.sql.catalyst.analysis.NoSuchDatabaseException: ",
                     fixed = TRUE)[[1]]
     # Extract "Error in ..." message.
     rmsg <- msg[1]
     # Extract the first message of JVM exception.
     first <- strsplit(msg[2], "\r?\n\tat")[[1]][1]
     stop(rmsg, "no such database - ", first, call. = FALSE)
   } else
     if (any(grepl("org.apache.spark.sql.catalyst.analysis.NoSuchTableException: ",
                   stacktrace, fixed = TRUE))) {
     msg <- strsplit(stacktrace, "org.apache.spark.sql.catalyst.analysis.NoSuchTableException: ",
                     fixed = TRUE)[[1]]
     # Extract "Error in ..." message.
     rmsg <- msg[1]
     # Extract the first message of JVM exception.
     first <- strsplit(msg[2], "\r?\n\tat")[[1]][1]
     stop(rmsg, "no such table - ", first, call. = FALSE)
   } else if (any(grepl("org.apache.spark.sql.catalyst.parser.ParseException: ",
                        stacktrace, fixed = TRUE))) {
     msg <- strsplit(stacktrace, "org.apache.spark.sql.catalyst.parser.ParseException: ",
                     fixed = TRUE)[[1]]
     # Extract "Error in ..." message.
     rmsg <- msg[1]
     # Extract the first message of JVM exception.
     first <- strsplit(msg[2], "\r?\n\tat")[[1]][1]
     stop(rmsg, "parse error - ", first, call. = FALSE)
   } else {
     stop(stacktrace, call. = FALSE)
   }
 }
 
 # rbind a list of rows with raw (binary) columns
 #
 # @param inputData a list of rows, with each row a list
 # @return data.frame with raw columns as lists
 rbindRaws <- function(inputData) {
   row1 <- inputData[[1]]
   rawcolumns <- ("raw" == sapply(row1, class))
 
   listmatrix <- do.call(rbind, inputData)
   # A dataframe with all list columns
   out <- as.data.frame(listmatrix)
   out[!rawcolumns] <- lapply(out[!rawcolumns], unlist)
   out
 }
 
 # Get basename without extension from URL
 basenameSansExtFromUrl <- function(url) {
   # split by '/'
   splits <- unlist(strsplit(url, "^.+/"))
   last <- tail(splits, 1)
   # this is from file_path_sans_ext
   # first, remove any compression extension
   filename <- sub("[.](gz|bz2|xz)$", "", last)
   # then, strip extension by the last '.'
   sub("([^.]+)\\.[[:alnum:]]+$", "\\1", filename)
 }
 
 isAtomicLengthOne <- function(x) {
   is.atomic(x) && length(x) == 1
 }
 
 is_windows <- function() {
   .Platform$OS.type == "windows"
 }
 
 hadoop_home_set <- function() {
   !identical(Sys.getenv("HADOOP_HOME"), "")
 }
 
 windows_with_hadoop <- function() {
   !is_windows() || hadoop_home_set()
 }
 
 # get0 not supported before R 3.2.0
 getOne <- function(x, envir, inherits = TRUE, ifnotfound = NULL) {
   mget(x[1L], envir = envir, inherits = inherits, ifnotfound = list(ifnotfound))[[1L]]
 }
 
 # Returns a vector of parent directories, traversing up count times, starting with a full path
 # eg. traverseParentDirs("/Users/user/Library/Caches/spark/spark2.2", 1) should return
 # this "/Users/user/Library/Caches/spark/spark2.2"
 # and  "/Users/user/Library/Caches/spark"
 traverseParentDirs <- function(x, count) {
   if (dirname(x) == x || count <= 0) x else c(x, Recall(dirname(x), count - 1))
 }

This page was automatically generated by the 2.1.0 LXR engine. The LXR team