You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2017/11/08 11:17:55 UTC
spark git commit: [SPARK-22446][SQL][ML] Declare StringIndexerModel
indexer udf as nondeterministic
Repository: spark
Updated Branches:
refs/heads/master 51debf8b1 -> 87343e155
[SPARK-22446][SQL][ML] Declare StringIndexerModel indexer udf as nondeterministic
## What changes were proposed in this pull request?
UDFs that can cause runtime exception on invalid data are not safe to pushdown, because its behavior depends on its position in the query plan. Pushdown of it will risk to change its original behavior.
The example reported in the JIRA and taken as test case shows this issue. We should declare UDFs that can cause runtime exception on invalid data as non-determinstic.
This updates the document of `deterministic` property in `Expression` and states clearly an UDF that can cause runtime exception on some specific input, should be declared as non-determinstic.
## How was this patch tested?
Added test. Manually test.
Author: Liang-Chi Hsieh <vi...@gmail.com>
Closes #19662 from viirya/SPARK-22446.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/87343e15
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/87343e15
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/87343e15
Branch: refs/heads/master
Commit: 87343e15566da870d1b8e49a78ef16e08ddfd406
Parents: 51debf8
Author: Liang-Chi Hsieh <vi...@gmail.com>
Authored: Wed Nov 8 12:17:52 2017 +0100
Committer: Wenchen Fan <we...@databricks.com>
Committed: Wed Nov 8 12:17:52 2017 +0100
----------------------------------------------------------------------
.../apache/spark/ml/feature/StringIndexer.scala | 2 +-
.../spark/ml/feature/VectorAssembler.scala | 2 +-
.../spark/ml/feature/StringIndexerSuite.scala | 17 +++++++++++++++
.../spark/ml/feature/VectorAssemblerSuite.scala | 23 +++++++++++++++++++-
.../sql/catalyst/expressions/Expression.scala | 1 +
5 files changed, 42 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/87343e15/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
----------------------------------------------------------------------
diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
index 2679ec3..1cdcdfc 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
@@ -261,7 +261,7 @@ class StringIndexerModel (
s"set Param handleInvalid to ${StringIndexer.KEEP_INVALID}.")
}
}
- }
+ }.asNondeterministic()
filteredDataset.select(col("*"),
indexer(dataset($(inputCol)).cast(StringType)).as($(outputCol), metadata))
http://git-wip-us.apache.org/repos/asf/spark/blob/87343e15/mllib/src/main/scala/org/apache/spark/ml/feature/VectorAssembler.scala
----------------------------------------------------------------------
diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/VectorAssembler.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/VectorAssembler.scala
index 73f27d1..b373ae9 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/feature/VectorAssembler.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/feature/VectorAssembler.scala
@@ -97,7 +97,7 @@ class VectorAssembler @Since("1.4.0") (@Since("1.4.0") override val uid: String)
// Data transformation.
val assembleFunc = udf { r: Row =>
VectorAssembler.assemble(r.toSeq: _*)
- }
+ }.asNondeterministic()
val args = $(inputCols).map { c =>
schema(c).dataType match {
case DoubleType => dataset(c)
http://git-wip-us.apache.org/repos/asf/spark/blob/87343e15/mllib/src/test/scala/org/apache/spark/ml/feature/StringIndexerSuite.scala
----------------------------------------------------------------------
diff --git a/mllib/src/test/scala/org/apache/spark/ml/feature/StringIndexerSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/feature/StringIndexerSuite.scala
index 027b1fb..775a04d 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/feature/StringIndexerSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/feature/StringIndexerSuite.scala
@@ -314,4 +314,21 @@ class StringIndexerSuite
idx += 1
}
}
+
+ test("SPARK-22446: StringIndexerModel's indexer UDF should not apply on filtered data") {
+ val df = List(
+ ("A", "London", "StrA"),
+ ("B", "Bristol", null),
+ ("C", "New York", "StrC")).toDF("ID", "CITY", "CONTENT")
+
+ val dfNoBristol = df.filter($"CONTENT".isNotNull)
+
+ val model = new StringIndexer()
+ .setInputCol("CITY")
+ .setOutputCol("CITYIndexed")
+ .fit(dfNoBristol)
+
+ val dfWithIndex = model.transform(dfNoBristol)
+ assert(dfWithIndex.filter($"CITYIndexed" === 1.0).count == 1)
+ }
}
http://git-wip-us.apache.org/repos/asf/spark/blob/87343e15/mllib/src/test/scala/org/apache/spark/ml/feature/VectorAssemblerSuite.scala
----------------------------------------------------------------------
diff --git a/mllib/src/test/scala/org/apache/spark/ml/feature/VectorAssemblerSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/feature/VectorAssemblerSuite.scala
index 6aef1c6..eca065f 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/feature/VectorAssemblerSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/feature/VectorAssemblerSuite.scala
@@ -24,7 +24,7 @@ import org.apache.spark.ml.param.ParamsSuite
import org.apache.spark.ml.util.DefaultReadWriteTest
import org.apache.spark.mllib.util.MLlibTestSparkContext
import org.apache.spark.sql.Row
-import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.functions.{col, udf}
class VectorAssemblerSuite
extends SparkFunSuite with MLlibTestSparkContext with DefaultReadWriteTest {
@@ -126,4 +126,25 @@ class VectorAssemblerSuite
.setOutputCol("myOutputCol")
testDefaultReadWrite(t)
}
+
+ test("SPARK-22446: VectorAssembler's UDF should not apply on filtered data") {
+ val df = Seq(
+ (0, 0.0, Vectors.dense(1.0, 2.0), "a", Vectors.sparse(2, Array(1), Array(3.0)), 10L),
+ (0, 1.0, null, "b", null, 20L)
+ ).toDF("id", "x", "y", "name", "z", "n")
+
+ val assembler = new VectorAssembler()
+ .setInputCols(Array("x", "z", "n"))
+ .setOutputCol("features")
+
+ val filteredDF = df.filter($"y".isNotNull)
+
+ val vectorUDF = udf { vector: Vector =>
+ vector.numActives
+ }
+
+ assert(assembler.transform(filteredDF).select("features")
+ .filter(vectorUDF($"features") > 1)
+ .count() == 1)
+ }
}
http://git-wip-us.apache.org/repos/asf/spark/blob/87343e15/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
index 0e75ac8..a3b722a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
@@ -75,6 +75,7 @@ abstract class Expression extends TreeNode[Expression] {
* - it relies on some mutable internal state, or
* - it relies on some implicit input that is not part of the children expression list.
* - it has non-deterministic child or children.
+ * - it assumes the input satisfies some certain condition via the child operator.
*
* An example would be `SparkPartitionID` that relies on the partition id returned by TaskContext.
* By default leaf expressions are deterministic as Nil.forall(_.deterministic) returns true.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org