Merge remote-tracking branch 'origin/master' into unsafe-sort

Conflicts: core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala core/src/main/scala/org/apache/spark/serializer/Serializer.scala core/src/test/scala/org/apache/spark/serializer/SerializerPropertiesSuite.scala sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlSerializer2.scala
apache · May 6, 2015 · b674412 · b674412
2 parents aaea17b + 845d1d4
commit b674412
Show file tree

Hide file tree

Showing 452 changed files with 48,544 additions and 3,769 deletions.
diff --git a/.rat-excludes b/.rat-excludes
@@ -15,6 +15,7 @@ TAGS
 RELEASE
 control
 docs
+docker.properties.template
 fairscheduler.xml.template
 spark-defaults.conf.template
 log4j.properties
@@ -29,6 +30,9 @@ spark-env.sh.template
 log4j-defaults.properties
 bootstrap-tooltip.js
 jquery-1.11.1.min.js
+d3.min.js
+dagre-d3.min.js
+graphlib-dot.min.js
 sorttable.js
 vis.min.js
 vis.min.css

diff --git a/LICENSE b/LICENSE
@@ -643,6 +643,36 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 
+========================================================================
+For d3 (core/src/main/resources/org/apache/spark/ui/static/d3.min.js):
+========================================================================
+
+Copyright (c) 2010-2015, Michael Bostock
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+* The name Michael Bostock may not be used to endorse or promote products
+  derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL MICHAEL BOSTOCK BE LIABLE FOR ANY DIRECT,
+INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
+EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 ========================================================================
 For Scala Interpreter classes (all .scala files in repl/src/main/scala

diff --git a/R/pkg/NAMESPACE b/R/pkg/NAMESPACE
@@ -1,117 +1,36 @@
-#exportPattern("^[[:alpha:]]+")
-exportClasses("RDD")
-exportClasses("Broadcast")
-exportMethods(
-              "aggregateByKey",
-              "aggregateRDD",
-              "cache",
-              "cartesian",
-              "checkpoint",
-              "coalesce",
-              "cogroup",
-              "collect",
-              "collectAsMap",
-              "collectPartition",
-              "combineByKey",
-              "count",
-              "countByKey",
-              "countByValue",
-              "distinct",
-              "Filter",
-              "filterRDD",
-              "first",
-              "flatMap",
-              "flatMapValues",
-              "fold",
-              "foldByKey",
-              "foreach",
-              "foreachPartition",
-              "fullOuterJoin",
-              "glom",
-              "groupByKey",
-              "intersection",
-              "join",
-              "keyBy",
-              "keys",
-              "length",
-              "lapply",
-              "lapplyPartition",
-              "lapplyPartitionsWithIndex",
-              "leftOuterJoin",
-              "lookup",
-              "map",
-              "mapPartitions",
-              "mapPartitionsWithIndex",
-              "mapValues",
-              "maximum",
-              "minimum",
-              "numPartitions",
-              "partitionBy",
-              "persist",
-              "pipeRDD",
-              "reduce",
-              "reduceByKey",
-              "reduceByKeyLocally",
-              "repartition",
-              "rightOuterJoin",
-              "sampleByKey",
-              "sampleRDD",
-              "saveAsTextFile",
-              "saveAsObjectFile",
-              "sortBy",
-              "sortByKey",
-              "subtract",
-              "subtractByKey",
-              "sumRDD",
-              "take",
-              "takeOrdered",
-              "takeSample",
-              "top",
-              "unionRDD",
-              "unpersist",
-              "value",
-              "values",
-              "zipPartitions",
-              "zipRDD",
-              "zipWithIndex",
-              "zipWithUniqueId"
-             )
+# Imports from base R
+importFrom(methods, setGeneric, setMethod, setOldClass)
+useDynLib(SparkR, stringHashCode)
 
 # S3 methods exported
-export(
-       "textFile",
-       "objectFile",
-       "parallelize",
-       "hashCode",
-       "includePackage",
-       "broadcast",
-       "setBroadcastValue",
-       "setCheckpointDir"
-      )
 export("sparkR.init")
 export("sparkR.stop")
 export("print.jobj")
-useDynLib(SparkR, stringHashCode)
-importFrom(methods, setGeneric, setMethod, setOldClass)
-
-# SparkRSQL
 
 exportClasses("DataFrame")
 
-exportMethods("columns",
+exportMethods("cache",
+              "collect",
+              "columns",
+              "count",
+              "describe",
               "distinct",
               "dtypes",
               "except",
               "explain",
               "filter",
+              "first",
               "groupBy",
               "head",
               "insertInto",
               "intersect",
               "isLocal",
+              "join",
+              "length",
               "limit",
               "orderBy",
               "names",
+              "persist",
               "printSchema",
               "registerTempTable",
               "repartition",
@@ -125,9 +44,11 @@ exportMethods("columns",
               "show",
               "showDF",
               "sortDF",
+              "take",
               "toJSON",
               "toRDD",
               "unionAll",
+              "unpersist",
               "where",
               "withColumn",
               "withColumnRenamed")

diff --git a/R/pkg/R/DataFrame.R b/R/pkg/R/DataFrame.R
@@ -167,7 +167,7 @@ setMethod("isLocal",
 setMethod("showDF",
           signature(x = "DataFrame"),
           function(x, numRows = 20) {
-            cat(callJMethod(x@sdf, "showString", numToInt(numRows)), "\n")
+            callJMethod(x@sdf, "showString", numToInt(numRows))
           })
 
 #' show
@@ -1276,3 +1276,40 @@ setMethod("saveAsTable",
             callJMethod(df@sdf, "saveAsTable", tableName, source, jmode, options)
           })
 
+#' describe
+#'
+#' Computes statistics for numeric columns.
+#' If no columns are given, this function computes statistics for all numerical columns.
+#'
+#' @param x A DataFrame to be computed.
+#' @param col A string of name
+#' @param ... Additional expressions
+#' @return A DataFrame
+#' @rdname describe 
+#' @export
+#' @examples
+#'\dontrun{
+#' sc <- sparkR.init()
+#' sqlCtx <- sparkRSQL.init(sc)
+#' path <- "path/to/file.json"
+#' df <- jsonFile(sqlCtx, path)
+#' describe(df)
+#' describe(df, "col1")
+#' describe(df, "col1", "col2")
+#' }
+setMethod("describe",
+          signature(x = "DataFrame", col = "character"),
+          function(x, col, ...) {
+            colList <- list(col, ...)
+            sdf <- callJMethod(x@sdf, "describe", listToSeq(colList))
+            dataFrame(sdf)
+          })
+
+#' @rdname describe
+setMethod("describe",
+          signature(x = "DataFrame"),
+          function(x) {
+            colList <- as.list(c(columns(x)))
+            sdf <- callJMethod(x@sdf, "describe", listToSeq(colList))
+            dataFrame(sdf)
+          })
diff --git a/R/pkg/R/RDD.R b/R/pkg/R/RDD.R
@@ -797,7 +797,7 @@ setMethod("first",
 #' @aliases distinct,RDD-method
 setMethod("distinct",
           signature(x = "RDD"),
-          function(x, numPartitions = SparkR::numPartitions(x)) {
+          function(x, numPartitions = SparkR:::numPartitions(x)) {
             identical.mapped <- lapply(x, function(x) { list(x, NULL) })
             reduced <- reduceByKey(identical.mapped,
                                    function(x, y) { x },
@@ -993,7 +993,7 @@ setMethod("coalesce",
            signature(x = "RDD", numPartitions = "numeric"),
            function(x, numPartitions, shuffle = FALSE) {
              numPartitions <- numToInt(numPartitions)
-             if (shuffle || numPartitions > SparkR::numPartitions(x)) {
+             if (shuffle || numPartitions > SparkR:::numPartitions(x)) {
                func <- function(partIndex, part) {
                  set.seed(partIndex)  # partIndex as seed
                  start <- as.integer(sample(numPartitions, 1) - 1)
@@ -1078,7 +1078,7 @@ setMethod("saveAsTextFile",
 #' @aliases sortBy,RDD,RDD-method
 setMethod("sortBy",
           signature(x = "RDD", func = "function"),
-          function(x, func, ascending = TRUE, numPartitions = SparkR::numPartitions(x)) {          
+          function(x, func, ascending = TRUE, numPartitions = SparkR:::numPartitions(x)) {
             values(sortByKey(keyBy(x, func), ascending, numPartitions))
           })
 
@@ -1552,7 +1552,7 @@ setMethod("cartesian",
 #' @aliases subtract,RDD
 setMethod("subtract",
           signature(x = "RDD", other = "RDD"),
-          function(x, other, numPartitions = SparkR::numPartitions(x)) {
+          function(x, other, numPartitions = SparkR:::numPartitions(x)) {
             mapFunction <- function(e) { list(e, NA) }
             rdd1 <- map(x, mapFunction)
             rdd2 <- map(other, mapFunction)
@@ -1583,7 +1583,7 @@ setMethod("subtract",
 #' @aliases intersection,RDD
 setMethod("intersection",
           signature(x = "RDD", other = "RDD"),
-          function(x, other, numPartitions = SparkR::numPartitions(x)) {
+          function(x, other, numPartitions = SparkR:::numPartitions(x)) {
             rdd1 <- map(x, function(v) { list(v, NA) })
             rdd2 <- map(other, function(v) { list(v, NA) })
 

diff --git a/R/pkg/R/generics.R b/R/pkg/R/generics.R
@@ -384,6 +384,10 @@ setGeneric("value", function(bcast) { standardGeneric("value") })
 #' @export
 setGeneric("columns", function(x) {standardGeneric("columns") })
 
+#' @rdname describe
+#' @export
+setGeneric("describe", function(x, col, ...) { standardGeneric("describe") })
+
 #' @rdname schema
 #' @export
 setGeneric("dtypes", function(x) { standardGeneric("dtypes") })

diff --git a/R/pkg/R/pairRDD.R b/R/pkg/R/pairRDD.R
@@ -739,7 +739,7 @@ setMethod("cogroup",
 #' @aliases sortByKey,RDD,RDD-method
 setMethod("sortByKey",
           signature(x = "RDD"),
-          function(x, ascending = TRUE, numPartitions = SparkR::numPartitions(x)) {
+          function(x, ascending = TRUE, numPartitions = SparkR:::numPartitions(x)) {
             rangeBounds <- list()
 
             if (numPartitions > 1) {
@@ -806,7 +806,7 @@ setMethod("sortByKey",
 #' @aliases subtractByKey,RDD
 setMethod("subtractByKey",
           signature(x = "RDD", other = "RDD"),
-          function(x, other, numPartitions = SparkR::numPartitions(x)) {
+          function(x, other, numPartitions = SparkR:::numPartitions(x)) {
             filterFunction <- function(elem) {
               iters <- elem[[2]]
               (length(iters[[1]]) > 0) && (length(iters[[2]]) == 0)

diff --git a/R/pkg/inst/tests/test_broadcast.R b/R/pkg/inst/tests/test_broadcast.R
@@ -29,7 +29,7 @@ test_that("using broadcast variable", {
   randomMatBr <- broadcast(sc, randomMat)
 
   useBroadcast <- function(x) {
-    sum(value(randomMatBr) * x)
+    sum(SparkR:::value(randomMatBr) * x)
   }
   actual <- collect(lapply(rrdd, useBroadcast))
   expected <- list(sum(randomMat) * 1, sum(randomMat) * 2)

diff --git a/R/pkg/inst/tests/test_sparkSQL.R b/R/pkg/inst/tests/test_sparkSQL.R
@@ -641,7 +641,7 @@ test_that("toJSON() returns an RDD of the correct values", {
 
 test_that("showDF()", {
   df <- jsonFile(sqlCtx, jsonPath)
-  expect_output(showDF(df), "age  name   \nnull Michael\n30   Andy   \n19   Justin ")
+  expect_output(showDF(df), "+----+-------+\n| age|   name|\n+----+-------+\n|null|Michael|\n|  30|   Andy|\n|  19| Justin|\n+----+-------+\n")
 })
 
 test_that("isLocal()", {
@@ -705,5 +705,16 @@ test_that("parquetFile works with multiple input paths", {
   expect_true(count(parquetDF) == count(df)*2)
 })
 
+test_that("describe() on a DataFrame", {
+  df <- jsonFile(sqlCtx, jsonPath)
+  stats <- describe(df, "age")
+  expect_true(collect(stats)[1, "summary"] == "count")
+  expect_true(collect(stats)[2, "age"] == 24.5)
+  expect_true(collect(stats)[3, "age"] == 5.5)
+  stats <- describe(df)
+  expect_true(collect(stats)[4, "name"] == "Andy")
+  expect_true(collect(stats)[5, "age"] == 30.0)
+})
+
 unlink(parquetPath)
 unlink(jsonPath)
diff --git a/R/pkg/inst/tests/test_utils.R b/R/pkg/inst/tests/test_utils.R
@@ -92,7 +92,10 @@ test_that("cleanClosure on R functions", {
   }
   newF <- cleanClosure(f)
   env <- environment(newF)
-  expect_equal(length(ls(env)), 3)  # Only "g", "l" and "f". No "base", "field" or "defUse".
+  # TODO(shivaram): length(ls(env)) is 4 here for some reason and `lapply` is included in `env`.
+  # Disabling this test till we debug this.
+  #
+  # expect_equal(length(ls(env)), 3)  # Only "g", "l" and "f". No "base", "field" or "defUse".
   expect_true("g" %in% ls(env))
   expect_true("l" %in% ls(env))
   expect_true("f" %in% ls(env))

diff --git a/R/pkg/inst/worker/worker.R b/R/pkg/inst/worker/worker.R
@@ -72,7 +72,7 @@ if (numBroadcastVars > 0) {
   for (bcast in seq(1:numBroadcastVars)) {
     bcastId <- SparkR:::readInt(inputCon)
     value <- unserialize(SparkR:::readRaw(inputCon))
-    setBroadcastValue(bcastId, value)
+    SparkR:::setBroadcastValue(bcastId, value)
   }
 }
 

diff --git a/conf/docker.properties.template b/conf/docker.properties.template
@@ -0,0 +1,3 @@
+spark.mesos.executor.docker.image: <image built from `../docker/spark-mesos/Dockerfile`>
+spark.mesos.executor.docker.volumes: /usr/local/lib:/host/usr/local/lib:ro
+spark.mesos.executor.home: /opt/spark
diff --git a/core/pom.xml b/core/pom.xml
@@ -273,7 +273,6 @@
     <dependency>
       <groupId>org.apache.ivy</groupId>
       <artifactId>ivy</artifactId>
-      <version>${ivy.version}</version>
     </dependency>
     <dependency>
       <groupId>oro</groupId>
@@ -362,7 +361,7 @@
     <dependency>
       <groupId>org.spark-project</groupId>
       <artifactId>pyrolite</artifactId>
-      <version>2.0.1</version>
+      <version>4.4</version>
     </dependency>
     <dependency>
       <groupId>net.sf.py4j</groupId>