aboutsummaryrefslogtreecommitdiff
path: root/streaming
diff options
context:
space:
mode:
authorfreeman <the.freeman.lab@gmail.com>2014-08-19 13:28:57 -0700
committerTathagata Das <tathagata.das1565@gmail.com>2014-08-19 13:28:57 -0700
commit31f0b071efd0b63eb9d6a6a131e5c4fa28237583 (patch)
tree2ce7834721e9296510c91673d67b54528709b2eb /streaming
parentcbfc26ba45f49559e64276c72e3054c6fe30ddd5 (diff)
downloadspark-31f0b071efd0b63eb9d6a6a131e5c4fa28237583.tar.gz
spark-31f0b071efd0b63eb9d6a6a131e5c4fa28237583.tar.bz2
spark-31f0b071efd0b63eb9d6a6a131e5c4fa28237583.zip
[SPARK-3128][MLLIB] Use streaming test suite for StreamingLR
Refactored tests for streaming linear regression to use existing streaming test utilities. Summary of changes: - Made ``mllib`` depend on tests from ``streaming`` - Rewrote accuracy and convergence tests to use ``setupStreams`` and ``runStreams`` - Added new test for the accuracy of predictions generated by ``predictOnValue`` These tests should run faster, be easier to extend/maintain, and provide a reference for new tests. mengxr tdas Author: freeman <the.freeman.lab@gmail.com> Closes #2037 from freeman-lab/streamingLR-predict-tests and squashes the following commits: e851ca7 [freeman] Fixed long lines 50eb0bf [freeman] Refactored tests to use streaming test tools 32c43c2 [freeman] Added test for prediction
Diffstat (limited to 'streaming')
-rw-r--r--streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala4
1 files changed, 3 insertions, 1 deletions
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala b/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala
index cc178fba12..f095da9cb5 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala
@@ -242,7 +242,9 @@ trait TestSuiteBase extends FunSuite with BeforeAndAfter with Logging {
logInfo("numBatches = " + numBatches + ", numExpectedOutput = " + numExpectedOutput)
// Get the output buffer
- val outputStream = ssc.graph.getOutputStreams.head.asInstanceOf[TestOutputStreamWithPartitions[V]]
+ val outputStream = ssc.graph.getOutputStreams.
+ filter(_.isInstanceOf[TestOutputStreamWithPartitions[_]]).
+ head.asInstanceOf[TestOutputStreamWithPartitions[V]]
val output = outputStream.output
try {