You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by zhengruifeng <gi...@git.apache.org> on 2016/03/17 07:41:09 UTC
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
GitHub user zhengruifeng opened a pull request:
https://github.com/apache/spark/pull/11776
[Minor][DOC] Add JavaStreamingTestExample
## What changes were proposed in this pull request?
Add the java example of StreamingTest
## How was this patch tested?
manual tests in CLI: bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zhengruifeng/spark streaming_je
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11776.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #11776
----
commit 73b6da5a60552277de05b44f0a63646501080a30
Author: Zheng RuiFeng <ru...@foxmail.com>
Date: 2016-03-16T13:43:38Z
create st_je
commit c10d5be2c338e822dd8ba561581e2bd7b4a8849b
Author: Zheng RuiFeng <ru...@foxmail.com>
Date: 2016-03-17T03:28:08Z
fix nits
commit 3458acaf4ca6ae44ae4d5cc740870cd3d7d29f98
Author: Zheng RuiFeng <ru...@foxmail.com>
Date: 2016-03-17T04:12:07Z
add to md
commit ff56ff56d46db9ee64924c44fb18c03c0ff91e4d
Author: Zheng RuiFeng <ru...@foxmail.com>
Date: 2016-03-17T04:13:45Z
add to md
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by zhengruifeng <gi...@git.apache.org>.
Github user zhengruifeng commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56470345
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,123 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+// $example on$
+import org.apache.spark.api.java.function.VoidFunction;
--- End diff --
@MLnick Thinks, I will remove it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197730304
**[Test build #53399 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53399/consoleFull)** for PR 11776 at commit [`ff56ff5`](https://github.com/apache/spark/commit/ff56ff56d46db9ee64924c44fb18c03c0ff91e4d).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197757486
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53405/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-198304584
You'll need to open another one I'm afraid.
On Fri, 18 Mar 2016 at 12:50 Ruifeng Zheng <no...@github.com> wrote:
> @MLnick <https://github.com/MLnick> How can I fix it in this PR? Or I
> have to open another one?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/11776#issuecomment-198302793>
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56481142
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+// $example on$
+import org.apache.spark.mllib.stat.test.BinarySample;
+import org.apache.spark.mllib.stat.test.StreamingTest;
+import org.apache.spark.mllib.stat.test.StreamingTestResult;
+// $example off$
+import org.apache.spark.SparkConf;
+import org.apache.spark.streaming.Duration;
+import org.apache.spark.streaming.Seconds;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.util.Utils;
+
+
+/**
+ * Perform streaming testing using Welch's 2-sample t-test on a stream of data, where the data
+ * stream arrives as text files in a directory. Stops when the two groups are statistically
+ * significant (p-value < 0.05) or after a user-specified timeout in number of batches is exceeded.
+ *
+ * The rows of the text files must be in the form `Boolean, Double`. For example:
+ * false, -3.92
+ * true, 99.32
+ *
+ * Usage:
+ * JavaStreamingTestExample <dataDir> <batchDuration> <numBatchesTimeout>
+ *
+ * To run on your local machine using the directory `dataDir` with 5 seconds between each batch and
+ * a timeout after 100 insignificant batches, call:
+ * $ bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
+ *
+ * As you add text files to `dataDir` the significance test wil continually update every
+ * `batchDuration` seconds until the test becomes significant (p-value < 0.05) or the number of
+ * batches processed exceeds `numBatchesTimeout`.
+ */
+public class JavaStreamingTestExample {
+ public static void main(String[] args) {
+ if (args.length != 3) {
+ System.err.println("Usage: JavaStreamingTestExample " +
+ "<dataDir> <batchDuration> <numBatchesTimeout>");
+ System.exit(1);
+ }
+
+ String dataDir = args[0];
+ Duration batchDuration = Seconds.apply(Long.valueOf(args[1]));
+ int numBatchesTimeout = Integer.valueOf(args[2]);
+
+ SparkConf conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample");
+ JavaStreamingContext ssc = new JavaStreamingContext(conf, batchDuration);
+
+ ssc.checkpoint(Utils.createTempDir(System.getProperty("java.io.tmpdir"), "spark").toString());
+
+ // $example on$
+ JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(
+ new Function<String, BinarySample>() {
+ @Override
+ public BinarySample call(String line) throws Exception {
+ String[] ts = line.split(",");
+ boolean label = Boolean.valueOf(ts[0]);
+ double value = Double.valueOf(ts[1]);
+ return new BinarySample(label, value);
+ }
+ });
+
+ StreamingTest streamingTest = new StreamingTest()
+ .setPeacePeriod(0)
+ .setWindowSize(0)
+ .setTestMethod("welch");
+
+ JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
+ out.print();
+ // $example off$
+
+ // Stop processing if test becomes significant or we time out
+ final Accumulator<Integer> timeoutCounter =
+ ssc.sparkContext().accumulator(numBatchesTimeout);
+
+ out.foreachRDD(new VoidFunction<JavaRDD<StreamingTestResult>>() {
+ @Override
+ public void call(JavaRDD<StreamingTestResult> rdd) throws Exception {
+ timeoutCounter.add(-1);
+
+ long cntSignificant = rdd.filter(new Function<StreamingTestResult, Boolean>() {
+ @Override
+ public Boolean call(StreamingTestResult v) throws Exception {
--- End diff --
`Utils.createTempDir` uses default args so args are required in Java.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56479985
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+// $example on$
+import org.apache.spark.mllib.stat.test.BinarySample;
+import org.apache.spark.mllib.stat.test.StreamingTest;
+import org.apache.spark.mllib.stat.test.StreamingTestResult;
+// $example off$
+import org.apache.spark.SparkConf;
+import org.apache.spark.streaming.Duration;
+import org.apache.spark.streaming.Seconds;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.util.Utils;
+
+
+/**
+ * Perform streaming testing using Welch's 2-sample t-test on a stream of data, where the data
+ * stream arrives as text files in a directory. Stops when the two groups are statistically
+ * significant (p-value < 0.05) or after a user-specified timeout in number of batches is exceeded.
+ *
+ * The rows of the text files must be in the form `Boolean, Double`. For example:
+ * false, -3.92
+ * true, 99.32
+ *
+ * Usage:
+ * JavaStreamingTestExample <dataDir> <batchDuration> <numBatchesTimeout>
+ *
+ * To run on your local machine using the directory `dataDir` with 5 seconds between each batch and
+ * a timeout after 100 insignificant batches, call:
+ * $ bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
+ *
+ * As you add text files to `dataDir` the significance test wil continually update every
+ * `batchDuration` seconds until the test becomes significant (p-value < 0.05) or the number of
+ * batches processed exceeds `numBatchesTimeout`.
+ */
+public class JavaStreamingTestExample {
+ public static void main(String[] args) {
+ if (args.length != 3) {
+ System.err.println("Usage: JavaStreamingTestExample " +
+ "<dataDir> <batchDuration> <numBatchesTimeout>");
+ System.exit(1);
+ }
+
+ String dataDir = args[0];
+ Duration batchDuration = Seconds.apply(Long.valueOf(args[1]));
+ int numBatchesTimeout = Integer.valueOf(args[2]);
+
+ SparkConf conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample");
+ JavaStreamingContext ssc = new JavaStreamingContext(conf, batchDuration);
+
+ ssc.checkpoint(Utils.createTempDir(System.getProperty("java.io.tmpdir"), "spark").toString());
+
+ // $example on$
+ JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(
+ new Function<String, BinarySample>() {
+ @Override
+ public BinarySample call(String line) throws Exception {
+ String[] ts = line.split(",");
+ boolean label = Boolean.valueOf(ts[0]);
+ double value = Double.valueOf(ts[1]);
+ return new BinarySample(label, value);
+ }
+ });
+
+ StreamingTest streamingTest = new StreamingTest()
+ .setPeacePeriod(0)
+ .setWindowSize(0)
+ .setTestMethod("welch");
+
+ JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
+ out.print();
+ // $example off$
+
+ // Stop processing if test becomes significant or we time out
+ final Accumulator<Integer> timeoutCounter =
+ ssc.sparkContext().accumulator(numBatchesTimeout);
+
+ out.foreachRDD(new VoidFunction<JavaRDD<StreamingTestResult>>() {
+ @Override
+ public void call(JavaRDD<StreamingTestResult> rdd) throws Exception {
+ timeoutCounter.add(-1);
+
+ long cntSignificant = rdd.filter(new Function<StreamingTestResult, Boolean>() {
+ @Override
+ public Boolean call(StreamingTestResult v) throws Exception {
--- End diff --
Ex: I see why this call doesn't use "fold" but could have been `!rdd.filter(...).isEmpty`; you don't need an arg to `createTempDir`; `timeoutCounter` actually shouldn't be an accumulator here. The only one that isn't trivial is the last one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-198207419
I think you can actually just make that variable a static int outside the
main definition, and -= it in the foreachRDD block. Similar to Scala.
I tested it and it works - no need even for an atomicinteger.
On Fri, 18 Mar 2016 at 03:33, Ruifeng Zheng <no...@github.com>
wrote:
> @MLnick <https://github.com/MLnick> @srowen <https://github.com/srowen>
> What should I do to refine the PR?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/11776#issuecomment-198153829>
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197757480
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197769698
**[Test build #53409 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53409/consoleFull)** for PR 11776 at commit [`f251229`](https://github.com/apache/spark/commit/f251229b817816dec90ba3019d70367513ea76f2).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by zhengruifeng <gi...@git.apache.org>.
Github user zhengruifeng commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-198309008
@MLnick ok I will open another pr.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197757077
**[Test build #53405 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53405/consoleFull)** for PR 11776 at commit [`7750299`](https://github.com/apache/spark/commit/77502998be1a2b1335f57c878d43d3dcb0044077).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by zhengruifeng <gi...@git.apache.org>.
Github user zhengruifeng commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-198302793
@MLnick How can I fix it in this PR? Or I have to open another one?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197732446
**[Test build #53399 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53399/consoleFull)** for PR 11776 at commit [`ff56ff5`](https://github.com/apache/spark/commit/ff56ff56d46db9ee64924c44fb18c03c0ff91e4d).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56478518
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+// $example on$
+import org.apache.spark.mllib.stat.test.BinarySample;
+import org.apache.spark.mllib.stat.test.StreamingTest;
+import org.apache.spark.mllib.stat.test.StreamingTestResult;
+// $example off$
+import org.apache.spark.SparkConf;
+import org.apache.spark.streaming.Duration;
+import org.apache.spark.streaming.Seconds;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.util.Utils;
+
+
+/**
+ * Perform streaming testing using Welch's 2-sample t-test on a stream of data, where the data
+ * stream arrives as text files in a directory. Stops when the two groups are statistically
+ * significant (p-value < 0.05) or after a user-specified timeout in number of batches is exceeded.
+ *
+ * The rows of the text files must be in the form `Boolean, Double`. For example:
+ * false, -3.92
+ * true, 99.32
+ *
+ * Usage:
+ * JavaStreamingTestExample <dataDir> <batchDuration> <numBatchesTimeout>
+ *
+ * To run on your local machine using the directory `dataDir` with 5 seconds between each batch and
+ * a timeout after 100 insignificant batches, call:
+ * $ bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
+ *
+ * As you add text files to `dataDir` the significance test wil continually update every
+ * `batchDuration` seconds until the test becomes significant (p-value < 0.05) or the number of
+ * batches processed exceeds `numBatchesTimeout`.
+ */
+public class JavaStreamingTestExample {
+ public static void main(String[] args) {
+ if (args.length != 3) {
+ System.err.println("Usage: JavaStreamingTestExample " +
+ "<dataDir> <batchDuration> <numBatchesTimeout>");
+ System.exit(1);
+ }
+
+ String dataDir = args[0];
+ Duration batchDuration = Seconds.apply(Long.valueOf(args[1]));
+ int numBatchesTimeout = Integer.valueOf(args[2]);
+
+ SparkConf conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample");
+ JavaStreamingContext ssc = new JavaStreamingContext(conf, batchDuration);
+
+ ssc.checkpoint(Utils.createTempDir(System.getProperty("java.io.tmpdir"), "spark").toString());
+
+ // $example on$
+ JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(
+ new Function<String, BinarySample>() {
+ @Override
+ public BinarySample call(String line) throws Exception {
+ String[] ts = line.split(",");
+ boolean label = Boolean.valueOf(ts[0]);
+ double value = Double.valueOf(ts[1]);
+ return new BinarySample(label, value);
+ }
+ });
+
+ StreamingTest streamingTest = new StreamingTest()
+ .setPeacePeriod(0)
+ .setWindowSize(0)
+ .setTestMethod("welch");
+
+ JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
+ out.print();
+ // $example off$
+
+ // Stop processing if test becomes significant or we time out
+ final Accumulator<Integer> timeoutCounter =
+ ssc.sparkContext().accumulator(numBatchesTimeout);
+
+ out.foreachRDD(new VoidFunction<JavaRDD<StreamingTestResult>>() {
+ @Override
+ public void call(JavaRDD<StreamingTestResult> rdd) throws Exception {
+ timeoutCounter.add(-1);
+
+ long cntSignificant = rdd.filter(new Function<StreamingTestResult, Boolean>() {
+ @Override
+ public Boolean call(StreamingTestResult v) throws Exception {
--- End diff --
Sorry Sean - I admit I made fairly a quick pass. What variances from the Scala example do you see?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197765606
**[Test build #53409 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53409/consoleFull)** for PR 11776 at commit [`f251229`](https://github.com/apache/spark/commit/f251229b817816dec90ba3019d70367513ea76f2).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197732525
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by zhengruifeng <gi...@git.apache.org>.
Github user zhengruifeng commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-198153829
@MLnick @srowen What should I do to refine the PR?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197769816
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by zhengruifeng <gi...@git.apache.org>.
Github user zhengruifeng commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56468065
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,119 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.mllib.stat.test.BinarySample;
--- End diff --
@MLnick OK, I add required imports according to your comments
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56482053
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+// $example on$
+import org.apache.spark.mllib.stat.test.BinarySample;
+import org.apache.spark.mllib.stat.test.StreamingTest;
+import org.apache.spark.mllib.stat.test.StreamingTestResult;
+// $example off$
+import org.apache.spark.SparkConf;
+import org.apache.spark.streaming.Duration;
+import org.apache.spark.streaming.Seconds;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.util.Utils;
+
+
+/**
+ * Perform streaming testing using Welch's 2-sample t-test on a stream of data, where the data
+ * stream arrives as text files in a directory. Stops when the two groups are statistically
+ * significant (p-value < 0.05) or after a user-specified timeout in number of batches is exceeded.
+ *
+ * The rows of the text files must be in the form `Boolean, Double`. For example:
+ * false, -3.92
+ * true, 99.32
+ *
+ * Usage:
+ * JavaStreamingTestExample <dataDir> <batchDuration> <numBatchesTimeout>
+ *
+ * To run on your local machine using the directory `dataDir` with 5 seconds between each batch and
+ * a timeout after 100 insignificant batches, call:
+ * $ bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
+ *
+ * As you add text files to `dataDir` the significance test wil continually update every
+ * `batchDuration` seconds until the test becomes significant (p-value < 0.05) or the number of
+ * batches processed exceeds `numBatchesTimeout`.
+ */
+public class JavaStreamingTestExample {
+ public static void main(String[] args) {
+ if (args.length != 3) {
+ System.err.println("Usage: JavaStreamingTestExample " +
+ "<dataDir> <batchDuration> <numBatchesTimeout>");
+ System.exit(1);
+ }
+
+ String dataDir = args[0];
+ Duration batchDuration = Seconds.apply(Long.valueOf(args[1]));
+ int numBatchesTimeout = Integer.valueOf(args[2]);
+
+ SparkConf conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample");
+ JavaStreamingContext ssc = new JavaStreamingContext(conf, batchDuration);
+
+ ssc.checkpoint(Utils.createTempDir(System.getProperty("java.io.tmpdir"), "spark").toString());
+
+ // $example on$
+ JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(
+ new Function<String, BinarySample>() {
+ @Override
+ public BinarySample call(String line) throws Exception {
+ String[] ts = line.split(",");
+ boolean label = Boolean.valueOf(ts[0]);
+ double value = Double.valueOf(ts[1]);
+ return new BinarySample(label, value);
+ }
+ });
+
+ StreamingTest streamingTest = new StreamingTest()
+ .setPeacePeriod(0)
+ .setWindowSize(0)
+ .setTestMethod("welch");
+
+ JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
+ out.print();
+ // $example off$
+
+ // Stop processing if test becomes significant or we time out
+ final Accumulator<Integer> timeoutCounter =
+ ssc.sparkContext().accumulator(numBatchesTimeout);
+
+ out.foreachRDD(new VoidFunction<JavaRDD<StreamingTestResult>>() {
+ @Override
+ public void call(JavaRDD<StreamingTestResult> rdd) throws Exception {
+ timeoutCounter.add(-1);
+
+ long cntSignificant = rdd.filter(new Function<StreamingTestResult, Boolean>() {
+ @Override
+ public Boolean call(StreamingTestResult v) throws Exception {
--- End diff --
Fair enough - if one tries to just use a local var similar to the Scala example it won't compile as it needs a static var, maybe this was the workaround. Agreed AtomicInteger is cleaner. Happy to clean that up
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56476337
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+// $example on$
+import org.apache.spark.mllib.stat.test.BinarySample;
+import org.apache.spark.mllib.stat.test.StreamingTest;
+import org.apache.spark.mllib.stat.test.StreamingTestResult;
+// $example off$
+import org.apache.spark.SparkConf;
+import org.apache.spark.streaming.Duration;
+import org.apache.spark.streaming.Seconds;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.util.Utils;
+
+
+/**
+ * Perform streaming testing using Welch's 2-sample t-test on a stream of data, where the data
+ * stream arrives as text files in a directory. Stops when the two groups are statistically
+ * significant (p-value < 0.05) or after a user-specified timeout in number of batches is exceeded.
+ *
+ * The rows of the text files must be in the form `Boolean, Double`. For example:
+ * false, -3.92
+ * true, 99.32
+ *
+ * Usage:
+ * JavaStreamingTestExample <dataDir> <batchDuration> <numBatchesTimeout>
+ *
+ * To run on your local machine using the directory `dataDir` with 5 seconds between each batch and
+ * a timeout after 100 insignificant batches, call:
+ * $ bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
+ *
+ * As you add text files to `dataDir` the significance test wil continually update every
+ * `batchDuration` seconds until the test becomes significant (p-value < 0.05) or the number of
+ * batches processed exceeds `numBatchesTimeout`.
+ */
+public class JavaStreamingTestExample {
+ public static void main(String[] args) {
+ if (args.length != 3) {
+ System.err.println("Usage: JavaStreamingTestExample " +
+ "<dataDir> <batchDuration> <numBatchesTimeout>");
+ System.exit(1);
+ }
+
+ String dataDir = args[0];
+ Duration batchDuration = Seconds.apply(Long.valueOf(args[1]));
+ int numBatchesTimeout = Integer.valueOf(args[2]);
+
+ SparkConf conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample");
+ JavaStreamingContext ssc = new JavaStreamingContext(conf, batchDuration);
+
+ ssc.checkpoint(Utils.createTempDir(System.getProperty("java.io.tmpdir"), "spark").toString());
+
+ // $example on$
+ JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(
+ new Function<String, BinarySample>() {
+ @Override
+ public BinarySample call(String line) throws Exception {
+ String[] ts = line.split(",");
+ boolean label = Boolean.valueOf(ts[0]);
+ double value = Double.valueOf(ts[1]);
+ return new BinarySample(label, value);
+ }
+ });
+
+ StreamingTest streamingTest = new StreamingTest()
+ .setPeacePeriod(0)
+ .setWindowSize(0)
+ .setTestMethod("welch");
+
+ JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
+ out.print();
+ // $example off$
+
+ // Stop processing if test becomes significant or we time out
+ final Accumulator<Integer> timeoutCounter =
+ ssc.sparkContext().accumulator(numBatchesTimeout);
+
+ out.foreachRDD(new VoidFunction<JavaRDD<StreamingTestResult>>() {
+ @Override
+ public void call(JavaRDD<StreamingTestResult> rdd) throws Exception {
+ timeoutCounter.add(-1);
+
+ long cntSignificant = rdd.filter(new Function<StreamingTestResult, Boolean>() {
+ @Override
+ public Boolean call(StreamingTestResult v) throws Exception {
--- End diff --
There are some minor style sub-optimalities here, like some indentation issues, unneeded "throws Exception", and some variances from the Scala example. No big deal but this could have been left open more than a couple hours to get some eyes on it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56464958
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,119 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.mllib.stat.test.BinarySample;
--- End diff --
Please add `// $example on$` and `// $example off$` for the required imports.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197751636
**[Test build #53405 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53405/consoleFull)** for PR 11776 at commit [`7750299`](https://github.com/apache/spark/commit/77502998be1a2b1335f57c878d43d3dcb0044077).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197732526
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53399/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56469940
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,123 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+// $example on$
+import org.apache.spark.api.java.function.VoidFunction;
--- End diff --
Don't think this import is actually required, as that code is after the final `$example off$`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197778301
Merged to master. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11776#discussion_r56481486
--- Diff: examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java ---
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.examples.mllib;
+
+
+import org.apache.spark.Accumulator;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.Function;
+// $example on$
+import org.apache.spark.mllib.stat.test.BinarySample;
+import org.apache.spark.mllib.stat.test.StreamingTest;
+import org.apache.spark.mllib.stat.test.StreamingTestResult;
+// $example off$
+import org.apache.spark.SparkConf;
+import org.apache.spark.streaming.Duration;
+import org.apache.spark.streaming.Seconds;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.util.Utils;
+
+
+/**
+ * Perform streaming testing using Welch's 2-sample t-test on a stream of data, where the data
+ * stream arrives as text files in a directory. Stops when the two groups are statistically
+ * significant (p-value < 0.05) or after a user-specified timeout in number of batches is exceeded.
+ *
+ * The rows of the text files must be in the form `Boolean, Double`. For example:
+ * false, -3.92
+ * true, 99.32
+ *
+ * Usage:
+ * JavaStreamingTestExample <dataDir> <batchDuration> <numBatchesTimeout>
+ *
+ * To run on your local machine using the directory `dataDir` with 5 seconds between each batch and
+ * a timeout after 100 insignificant batches, call:
+ * $ bin/run-example mllib.JavaStreamingTestExample dataDir 5 100
+ *
+ * As you add text files to `dataDir` the significance test wil continually update every
+ * `batchDuration` seconds until the test becomes significant (p-value < 0.05) or the number of
+ * batches processed exceeds `numBatchesTimeout`.
+ */
+public class JavaStreamingTestExample {
+ public static void main(String[] args) {
+ if (args.length != 3) {
+ System.err.println("Usage: JavaStreamingTestExample " +
+ "<dataDir> <batchDuration> <numBatchesTimeout>");
+ System.exit(1);
+ }
+
+ String dataDir = args[0];
+ Duration batchDuration = Seconds.apply(Long.valueOf(args[1]));
+ int numBatchesTimeout = Integer.valueOf(args[2]);
+
+ SparkConf conf = new SparkConf().setMaster("local").setAppName("StreamingTestExample");
+ JavaStreamingContext ssc = new JavaStreamingContext(conf, batchDuration);
+
+ ssc.checkpoint(Utils.createTempDir(System.getProperty("java.io.tmpdir"), "spark").toString());
+
+ // $example on$
+ JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(
+ new Function<String, BinarySample>() {
+ @Override
+ public BinarySample call(String line) throws Exception {
+ String[] ts = line.split(",");
+ boolean label = Boolean.valueOf(ts[0]);
+ double value = Double.valueOf(ts[1]);
+ return new BinarySample(label, value);
+ }
+ });
+
+ StreamingTest streamingTest = new StreamingTest()
+ .setPeacePeriod(0)
+ .setWindowSize(0)
+ .setTestMethod("welch");
+
+ JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
+ out.print();
+ // $example off$
+
+ // Stop processing if test becomes significant or we time out
+ final Accumulator<Integer> timeoutCounter =
+ ssc.sparkContext().accumulator(numBatchesTimeout);
+
+ out.foreachRDD(new VoidFunction<JavaRDD<StreamingTestResult>>() {
+ @Override
+ public void call(JavaRDD<StreamingTestResult> rdd) throws Exception {
+ timeoutCounter.add(-1);
+
+ long cntSignificant = rdd.filter(new Function<StreamingTestResult, Boolean>() {
+ @Override
+ public Boolean call(StreamingTestResult v) throws Exception {
--- End diff --
Ah right, scratch that one. The only thing that might be an issue here is counting with the accumulator. It's all actually local, so, I'm guessing it works OK anyway. It could be an AtomicInteger or anything that can be decremented and referred to inside the function, which will only ever run on the driver.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11776#issuecomment-197769822
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53409/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [Minor][DOC] Add JavaStreamingTestExample
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11776
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org