[GitHub] incubator-beam pull request #794: Compressed source test and cleanups
GitHub user dhalperi opened a pull request: https://github.com/apache/incubator-beam/pull/794 Compressed source test and cleanups R: @jkff @bjchambers You can merge this pull request into a Git repository by running: $ git pull https://github.com/dhalperi/incubator-beam compressed-source-test Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/794.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 #794 commit 960e8db24aa96be812d81dc86c75ededddcc9a0d Author: Dan Halperin Date: 2016-08-05T15:51:15Z TextIO: test compression with larger files It is important for correctness that we test with large files because otherwise the compressed file may be larger than the uncompressed file, which could mask bugs commit 8a83165fff9e419fd828499243b3055079791bed Author: Dan Halperin Date: 2016-08-05T16:05:55Z CompressedSource: CompressedReader is never splittable The only way it's safe to split a compressed file is if the file is not compressed. This can only happen when the source itself is splittable, and that in turn will result in the inner source's reader being returned. A CompressedReader will only be created in the event that the file is NOT splittable. So remove all the logic handling splittable compressed readers, and instead go with the logic when we know/assume the file is compressed. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-beam pull request #795: Minor Cleanups in the Direct Runner
GitHub user tgroh opened a pull request: https://github.com/apache/incubator-beam/pull/795 Minor Cleanups in the Direct Runner Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [ ] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [ ] Replace `` in the title with the actual Jira issue number, if there is one. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- Consistent use of clock. Add a new test for DirectRunner Pipeline Reuse Remove unnecessary assignment. You can merge this pull request into a Git repository by running: $ git pull https://github.com/tgroh/incubator-beam direct_cleanups Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/795.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 #795 commit 7585cfc3693800b00c4ccc799c27f0311e9b0cc1 Author: Thomas Groh Date: 2016-08-05T16:58:05Z Remove References to Instant#now in the DirectRunner The DirectRunner should use exclusively the configured clock to determine the processing time. commit 4546fd9c5e073eb33787faa302b8695dfd6e04aa Author: Thomas Groh Date: 2016-08-05T16:58:59Z Add DirectRunner Reuse Test Two calls to run using the Direct Runner should be independent and succeed independently. commit acf71d313a43c8f39213ce19277ffebadcc40a77 Author: Thomas Groh Date: 2016-08-05T16:59:46Z Remove unneccssary Assignment in TransformExecutor --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[2/2] incubator-beam git commit: This closes #757
This closes #757 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/2b5c6bcb Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/2b5c6bcb Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/2b5c6bcb Branch: refs/heads/master Commit: 2b5c6bcb2173d55f6b45f4134cd500d9cc9c1965 Parents: 8daf518 4ac5caf Author: Kenneth Knowles Authored: Fri Aug 5 10:09:25 2016 -0700 Committer: Kenneth Knowles Committed: Fri Aug 5 10:09:25 2016 -0700 -- .../beam/sdk/transforms/FlatMapElements.java| 126 +-- .../apache/beam/sdk/transforms/MapElements.java | 60 + .../beam/sdk/transforms/SimpleFunction.java | 34 + .../sdk/transforms/FlatMapElementsTest.java | 48 +++ .../beam/sdk/transforms/MapElementsTest.java| 84 + 5 files changed, 288 insertions(+), 64 deletions(-) --
[1/2] incubator-beam git commit: Use input type in coder inference for MapElements and FlatMapElements
Repository: incubator-beam
Updated Branches:
refs/heads/master 8daf518bc -> 2b5c6bcb2
Use input type in coder inference for MapElements and FlatMapElements
Previously, the input TypeDescriptor was unknown, so we would fail
to infer a coder for things like MapElements.of(SimpleFunction)
even if the input PCollection provided a coder for T.
Now, the input type is plumbed appropriately and the coder is inferred.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/4ac5cafe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/4ac5cafe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/4ac5cafe
Branch: refs/heads/master
Commit: 4ac5cafe90a371cf616f97cb202d5016b68616d1
Parents: 8daf518
Author: Kenneth Knowles
Authored: Fri Jul 29 10:35:01 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 20:18:59 2016 -0700
--
.../beam/sdk/transforms/FlatMapElements.java| 126 +--
.../apache/beam/sdk/transforms/MapElements.java | 60 +
.../beam/sdk/transforms/SimpleFunction.java | 34 +
.../sdk/transforms/FlatMapElementsTest.java | 48 +++
.../beam/sdk/transforms/MapElementsTest.java| 84 +
5 files changed, 288 insertions(+), 64 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4ac5cafe/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
index 694592e..04d993c 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
@@ -17,8 +17,10 @@
*/
package org.apache.beam.sdk.transforms;
+import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
import java.lang.reflect.ParameterizedType;
@@ -45,8 +47,16 @@ extends PTransform,
PCollection> {
* descriptor need not be provided.
*/
public static MissingOutputTypeDescriptor
- via(SerializableFunction> fn) {
-return new MissingOutputTypeDescriptor<>(fn);
+ via(SerializableFunction> fn) {
+
+// TypeDescriptor interacts poorly with the wildcards needed to correctly
express
+// covariance and contravariance in Java, so instead we cast it to an
invariant
+// function here.
+@SuppressWarnings("unchecked") // safe covariant cast
+SerializableFunction> simplerFn =
+(SerializableFunction>) fn;
+
+return new MissingOutputTypeDescriptor<>(simplerFn);
}
/**
@@ -72,16 +82,15 @@ extends PTransform,
PCollection> {
* To use a Java 8 lambda, see {@link #via(SerializableFunction)}.
*/
public static FlatMapElements
- via(SimpleFunction> fn) {
-
-@SuppressWarnings({"rawtypes", "unchecked"}) // safe by static typing
-TypeDescriptor> iterableType = (TypeDescriptor)
fn.getOutputTypeDescriptor();
-
-@SuppressWarnings("unchecked") // safe by correctness of
getIterableElementType
-TypeDescriptor outputType =
-(TypeDescriptor) getIterableElementType(iterableType);
-
-return new FlatMapElements<>(fn, outputType);
+ via(SimpleFunction> fn) {
+// TypeDescriptor interacts poorly with the wildcards needed to correctly
express
+// covariance and contravariance in Java, so instead we cast it to an
invariant
+// function here.
+@SuppressWarnings("unchecked") // safe covariant cast
+SimpleFunction> simplerFn =
+(SimpleFunction>) fn;
+
+return new FlatMapElements<>(simplerFn, fn.getClass());
}
/**
@@ -91,18 +100,80 @@ extends PTransform,
PCollection> {
*/
public static final class MissingOutputTypeDescriptor {
-private final SerializableFunction> fn;
+private final SerializableFunction> fn;
private MissingOutputTypeDescriptor(
-SerializableFunction> fn) {
+SerializableFunction> fn) {
this.fn = fn;
}
public FlatMapElements
withOutputType(TypeDescriptor outputType) {
- return new FlatMapElements<>(fn, outputType);
+ TypeDescriptor> iterableOutputType =
TypeDescriptors.iterables(outputType);
+
+ return new FlatMapElements<>(
+ SimpleFunction.fromSerializableFunctionWithOutputType(fn,
+ iterableOutputType),
+ fn.getClass());
}
}
+
//
+
+ private final SimpleFunct
[GitHub] incubator-beam pull request #757: Use input type in coder inference for MapE...
Github user asfgit closed the pull request at: https://github.com/apache/incubator-beam/pull/757 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[2/4] incubator-beam git commit: Add DirectRunner Reuse Test
Add DirectRunner Reuse Test
Two calls to run using the Direct Runner should be independent and
succeed independently.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/4546fd9c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/4546fd9c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/4546fd9c
Branch: refs/heads/master
Commit: 4546fd9c5e073eb33787faa302b8695dfd6e04aa
Parents: 7585cfc
Author: Thomas Groh
Authored: Fri Aug 5 09:58:59 2016 -0700
Committer: Thomas Groh
Committed: Fri Aug 5 10:04:21 2016 -0700
--
.../beam/runners/direct/DirectRunnerTest.java | 31
1 file changed, 31 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4546fd9c/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
--
diff --git
a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index 29dea32..1e73ec0 100644
---
a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -109,6 +109,37 @@ public class DirectRunnerTest implements Serializable {
result.awaitCompletion();
}
+ @Test
+ public void reusePipelineSucceeds() throws Throwable {
+Pipeline p = getPipeline();
+
+PCollection> counts =
+p.apply(Create.of("foo", "bar", "foo", "baz", "bar", "foo"))
+.apply(MapElements.via(new SimpleFunction() {
+ @Override
+ public String apply(String input) {
+return input;
+ }
+}))
+.apply(Count.perElement());
+PCollection countStrs =
+counts.apply(MapElements.via(new SimpleFunction,
String>() {
+ @Override
+ public String apply(KV input) {
+String str = String.format("%s: %s", input.getKey(),
input.getValue());
+return str;
+ }
+}));
+
+PAssert.that(countStrs).containsInAnyOrder("baz: 1", "bar: 2", "foo: 3");
+
+DirectPipelineResult result = ((DirectPipelineResult) p.run());
+result.awaitCompletion();
+
+DirectPipelineResult otherResult = ((DirectPipelineResult) p.run());
+otherResult.awaitCompletion();
+ }
+
@Test(timeout = 5000L)
public void byteArrayCountShouldSucceed() {
Pipeline p = getPipeline();
[1/4] incubator-beam git commit: Remove unneccssary Assignment in TransformExecutor
Repository: incubator-beam
Updated Branches:
refs/heads/master 2b5c6bcb2 -> c584b37b8
Remove unneccssary Assignment in TransformExecutor
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/acf71d31
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/acf71d31
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/acf71d31
Branch: refs/heads/master
Commit: acf71d313a43c8f39213ce19277ffebadcc40a77
Parents: 4546fd9
Author: Thomas Groh
Authored: Fri Aug 5 09:59:46 2016 -0700
Committer: Thomas Groh
Committed: Fri Aug 5 10:04:21 2016 -0700
--
.../java/org/apache/beam/runners/direct/TransformExecutor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/acf71d31/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
--
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
index 3db941d..d873bf5 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
@@ -118,7 +118,7 @@ class TransformExecutor implements Runnable {
processElements(evaluator, enforcements);
- TransformResult result = finishBundle(evaluator, enforcements);
+ finishBundle(evaluator, enforcements);
} catch (Throwable t) {
onComplete.handleThrowable(inputBundle, t);
if (t instanceof RuntimeException) {
[3/4] incubator-beam git commit: Remove References to Instant#now in the DirectRunner
Remove References to Instant#now in the DirectRunner
The DirectRunner should use exclusively the configured clock to
determine the processing time.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7585cfc3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7585cfc3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7585cfc3
Branch: refs/heads/master
Commit: 7585cfc3693800b00c4ccc799c27f0311e9b0cc1
Parents: fcf6b1d
Author: Thomas Groh
Authored: Fri Aug 5 09:58:05 2016 -0700
Committer: Thomas Groh
Committed: Fri Aug 5 10:04:21 2016 -0700
--
.../apache/beam/runners/direct/EvaluationContext.java | 14 ++
.../direct/ExecutorServiceParallelExecutor.java | 5 ++---
2 files changed, 12 insertions(+), 7 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7585cfc3/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
--
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
index 23c139d..94f28e2 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
@@ -48,6 +48,8 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.MoreExecutors;
+import org.joda.time.Instant;
+
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
@@ -81,6 +83,7 @@ class EvaluationContext {
/** The options that were used to create this {@link Pipeline}. */
private final DirectOptions options;
+ private final Clock clock;
private final BundleFactory bundleFactory;
/** The current processing time and event time watermarks and timers. */
@@ -116,6 +119,7 @@ class EvaluationContext {
Map, String> stepNames,
Collection> views) {
this.options = checkNotNull(options);
+this.clock = options.getClock();
this.bundleFactory = checkNotNull(bundleFactory);
checkNotNull(rootTransforms);
checkNotNull(valueToConsumers);
@@ -123,9 +127,7 @@ class EvaluationContext {
checkNotNull(views);
this.stepNames = stepNames;
-this.watermarkManager =
-WatermarkManager.create(
-NanosOffsetClock.create(), rootTransforms, valueToConsumers);
+this.watermarkManager = WatermarkManager.create(clock, rootTransforms,
valueToConsumers);
this.sideInputContainer = SideInputContainer.create(this, views);
this.applicationStateInternals = new ConcurrentHashMap<>();
@@ -314,7 +316,7 @@ class EvaluationContext {
AppliedPTransform application, StructuralKey key) {
StepAndKey stepAndKey = StepAndKey.of(application, key);
return new DirectExecutionContext(
-options.getClock(),
+clock,
key,
(CopyOnAccessInMemoryStateInternals)
applicationStateInternals.get(stepAndKey),
watermarkManager.getWatermarks(application));
@@ -427,4 +429,8 @@ class EvaluationContext {
}
return true;
}
+
+ public Instant now() {
+return clock.now();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7585cfc3/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
--
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
index 64836d8..a0a5ec0 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
@@ -40,7 +40,6 @@ import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
-import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -433,9 +432,9 @@ final class ExecutorServiceParallelExecutor implements
PipelineExecutor {
.createKeyedBundle(
null, keyTimers.getKey(), (PCollection)
transform.getInput())
.add(WindowedValue.valueInEmptyWindows(work))
- .commit(Instant.now());
- state.set(ExecutorState.ACTIVE);
+ .commit(evaluationContext.now());
[4/4] incubator-beam git commit: This closes #795
This closes #795 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c584b37b Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c584b37b Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c584b37b Branch: refs/heads/master Commit: c584b37b8ac4e863bad83a766b6871ccc7135334 Parents: 2b5c6bc acf71d3 Author: Kenneth Knowles Authored: Fri Aug 5 10:40:17 2016 -0700 Committer: Kenneth Knowles Committed: Fri Aug 5 10:40:17 2016 -0700 -- .../beam/runners/direct/EvaluationContext.java | 14 ++--- .../direct/ExecutorServiceParallelExecutor.java | 5 ++-- .../beam/runners/direct/TransformExecutor.java | 2 +- .../beam/runners/direct/DirectRunnerTest.java | 31 4 files changed, 44 insertions(+), 8 deletions(-) --
[GitHub] incubator-beam pull request #795: Minor Cleanups in the Direct Runner
Github user asfgit closed the pull request at: https://github.com/apache/incubator-beam/pull/795 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Resolved] (BEAM-479) Move RunnableOnService test executions to postcommit
[ https://issues.apache.org/jira/browse/BEAM-479?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kenneth Knowles resolved BEAM-479. -- Resolution: Fixed Fix Version/s: Not applicable > Move RunnableOnService test executions to postcommit > > > Key: BEAM-479 > URL: https://issues.apache.org/jira/browse/BEAM-479 > Project: Beam > Issue Type: Improvement > Components: runner-dataflow, runner-flink, runner-gearpump, > runner-spark >Reporter: Kenneth Knowles >Assignee: Kenneth Knowles > Fix For: Not applicable > > > The Spark and Flink RunnableOnService tests only use a local endpoint, so > they _can_ run as unit tests, but our test duration is getting out of hand. > For Gearpump, the tests timeout. So this ticket tracks getting everyone to a > symmetric configuration. > Later, we can re-enable a select few local endpoint tests for the various > runners, to smoke test, and hopefully get actual cluster-based integration > tests running on Jenkins. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (BEAM-535) Cloud Datastore naming clean-up
Vikas Kedigehalli created BEAM-535: -- Summary: Cloud Datastore naming clean-up Key: BEAM-535 URL: https://issues.apache.org/jira/browse/BEAM-535 Project: Beam Issue Type: Improvement Components: sdk-java-gcp Affects Versions: 0.2.0-incubating Reporter: Vikas Kedigehalli Assignee: Vikas Kedigehalli Priority: Minor Clean up documentation to be always use "Cloud Datastore" instead of "Datastore" and make it consistent. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] incubator-beam pull request #796: [BEAM-535] Cloud Datastore naming clean-up
GitHub user vikkyrk opened a pull request: https://github.com/apache/incubator-beam/pull/796 [BEAM-535] Cloud Datastore naming clean-up Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [ ] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [ ] Replace `` in the title with the actual Jira issue number, if there is one. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- You can merge this pull request into a Git repository by running: $ git pull https://github.com/vikkyrk/incubator-beam vikasrk/ds_doc_cleanup Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/796.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 #796 commit 6e52edf5da681123f6b40de081d993e5a5517d95 Author: Vikas Kedigehalli Date: 2016-08-05T18:02:41Z Cloud Datastore naming clean-up commit fa6baf859aaa93daa3829b170029016777ad47ba Author: Vikas Kedigehalli Date: 2016-08-05T18:22:59Z Include examples as well --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Commented] (BEAM-535) Cloud Datastore naming clean-up
[ https://issues.apache.org/jira/browse/BEAM-535?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15409898#comment-15409898 ] ASF GitHub Bot commented on BEAM-535: - GitHub user vikkyrk opened a pull request: https://github.com/apache/incubator-beam/pull/796 [BEAM-535] Cloud Datastore naming clean-up Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [ ] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [ ] Replace `` in the title with the actual Jira issue number, if there is one. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- You can merge this pull request into a Git repository by running: $ git pull https://github.com/vikkyrk/incubator-beam vikasrk/ds_doc_cleanup Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/796.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 #796 commit 6e52edf5da681123f6b40de081d993e5a5517d95 Author: Vikas Kedigehalli Date: 2016-08-05T18:02:41Z Cloud Datastore naming clean-up commit fa6baf859aaa93daa3829b170029016777ad47ba Author: Vikas Kedigehalli Date: 2016-08-05T18:22:59Z Include examples as well > Cloud Datastore naming clean-up > --- > > Key: BEAM-535 > URL: https://issues.apache.org/jira/browse/BEAM-535 > Project: Beam > Issue Type: Improvement > Components: sdk-java-gcp >Affects Versions: 0.2.0-incubating >Reporter: Vikas Kedigehalli >Assignee: Vikas Kedigehalli >Priority: Minor > > Clean up documentation to be always use "Cloud Datastore" instead of > "Datastore" and make it consistent. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] incubator-beam pull request #797: [BEAM-498] Port more easy cases to new DoF...
GitHub user kennknowles opened a pull request: https://github.com/apache/incubator-beam/pull/797 [BEAM-498] Port more easy cases to new DoFn Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [x] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [x] Replace `` in the title with the actual Jira issue number, if there is one. - [x] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- You can merge this pull request into a Git repository by running: $ git pull https://github.com/kennknowles/incubator-beam port-more-DoFn Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/797.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 #797 commit c5a2593f9b91fa2fbb53f933857f7d3ca61a795e Author: Kenneth Knowles Date: 2016-08-05T18:46:53Z Port Filter to the new DoFn commit 6587a222e196fd6787a0add46f84d3e450d01140 Author: Kenneth Knowles Date: 2016-08-05T18:47:23Z Port Reshuffle to new DoFn commit bd10c123cf4115959d19d7a64a9ba39413b4160a Author: Kenneth Knowles Date: 2016-08-05T18:47:57Z Port Window transform to new DoFn commit 2383a1bec774cd449a3a2eabd7575b7758111d14 Author: Kenneth Knowles Date: 2016-08-05T18:49:37Z Port Write to new DoFn commit bb6a1e3215496fb1014f2952f88757e2abcf6ad0 Author: Kenneth Knowles Date: 2016-08-05T18:54:57Z Remove references to OldDoFn from DoFnTest commit 0669e746641112a091c5e7c5aaabd6e6499f3710 Author: Kenneth Knowles Date: 2016-08-05T19:07:28Z Port ViewTest to new DoFn commit 1afe9d1406b320ef55005cb1e2f059d00a16fe11 Author: Kenneth Knowles Date: 2016-08-05T19:11:11Z Port most of Combine to new DoFn commit 813d68c5463a2081c8081b878625304004772c8b Author: Kenneth Knowles Date: 2016-08-05T19:12:12Z Port Java 8 tests to new DoFn commit 148af203cb724c9a4a48937f472b6f5380996d0a Author: Kenneth Knowles Date: 2016-08-05T19:17:20Z Port Flink integration tests to new DoFn commit 23848bc1667b0f83489717e4d92144e328c5b214 Author: Kenneth Knowles Date: 2016-08-05T19:24:24Z Port Flink fork of examples to new DoFn commit 6cec56c765bfb34b6a5451957c75e1f4ed3b8635 Author: Kenneth Knowles Date: 2016-08-05T19:26:53Z Port BigQueryIO to new DoFn commit c9ccd31648405f57a0696642205dcda7fd4c4d25 Author: Kenneth Knowles Date: 2016-08-05T19:31:07Z Port various Spark runner tests to new DoFn --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Commented] (BEAM-498) Make DoFnWithContext the new DoFn
[ https://issues.apache.org/jira/browse/BEAM-498?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15409945#comment-15409945 ] ASF GitHub Bot commented on BEAM-498: - GitHub user kennknowles opened a pull request: https://github.com/apache/incubator-beam/pull/797 [BEAM-498] Port more easy cases to new DoFn Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [x] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [x] Replace `` in the title with the actual Jira issue number, if there is one. - [x] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- You can merge this pull request into a Git repository by running: $ git pull https://github.com/kennknowles/incubator-beam port-more-DoFn Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/797.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 #797 commit c5a2593f9b91fa2fbb53f933857f7d3ca61a795e Author: Kenneth Knowles Date: 2016-08-05T18:46:53Z Port Filter to the new DoFn commit 6587a222e196fd6787a0add46f84d3e450d01140 Author: Kenneth Knowles Date: 2016-08-05T18:47:23Z Port Reshuffle to new DoFn commit bd10c123cf4115959d19d7a64a9ba39413b4160a Author: Kenneth Knowles Date: 2016-08-05T18:47:57Z Port Window transform to new DoFn commit 2383a1bec774cd449a3a2eabd7575b7758111d14 Author: Kenneth Knowles Date: 2016-08-05T18:49:37Z Port Write to new DoFn commit bb6a1e3215496fb1014f2952f88757e2abcf6ad0 Author: Kenneth Knowles Date: 2016-08-05T18:54:57Z Remove references to OldDoFn from DoFnTest commit 0669e746641112a091c5e7c5aaabd6e6499f3710 Author: Kenneth Knowles Date: 2016-08-05T19:07:28Z Port ViewTest to new DoFn commit 1afe9d1406b320ef55005cb1e2f059d00a16fe11 Author: Kenneth Knowles Date: 2016-08-05T19:11:11Z Port most of Combine to new DoFn commit 813d68c5463a2081c8081b878625304004772c8b Author: Kenneth Knowles Date: 2016-08-05T19:12:12Z Port Java 8 tests to new DoFn commit 148af203cb724c9a4a48937f472b6f5380996d0a Author: Kenneth Knowles Date: 2016-08-05T19:17:20Z Port Flink integration tests to new DoFn commit 23848bc1667b0f83489717e4d92144e328c5b214 Author: Kenneth Knowles Date: 2016-08-05T19:24:24Z Port Flink fork of examples to new DoFn commit 6cec56c765bfb34b6a5451957c75e1f4ed3b8635 Author: Kenneth Knowles Date: 2016-08-05T19:26:53Z Port BigQueryIO to new DoFn commit c9ccd31648405f57a0696642205dcda7fd4c4d25 Author: Kenneth Knowles Date: 2016-08-05T19:31:07Z Port various Spark runner tests to new DoFn > Make DoFnWithContext the new DoFn > - > > Key: BEAM-498 > URL: https://issues.apache.org/jira/browse/BEAM-498 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core >Reporter: Kenneth Knowles >Assignee: Kenneth Knowles > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Resolved] (BEAM-382) Audit BoundedReaders for BEAM-381
[ https://issues.apache.org/jira/browse/BEAM-382?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Daniel Halperin resolved BEAM-382. -- Resolution: Fixed Fix Version/s: Not applicable > Audit BoundedReaders for BEAM-381 > - > > Key: BEAM-382 > URL: https://issues.apache.org/jira/browse/BEAM-382 > Project: Beam > Issue Type: Bug > Components: sdk-java-core >Reporter: Daniel Halperin >Assignee: Daniel Halperin > Fix For: Not applicable > > > Prevent similar issues -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] incubator-beam pull request #798: [BEAM-523] Fixed typo in aggregator_test.p...
GitHub user shlee0605 opened a pull request: https://github.com/apache/incubator-beam/pull/798 [BEAM-523] Fixed typo in aggregator_test.py Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [x] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [x] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [x] Replace `` in the title with the actual Jira issue number, if there is one. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- You can merge this pull request into a Git repository by running: $ git pull https://github.com/shlee0605/incubator-beam aggregator-typo Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/798.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 #798 commit 378f67e1cb5082a3869ce8fbf17adeba30924ead Author: Seunghyun Lee Date: 2016-08-05T20:35:12Z fixed typo in aggregator_test.py --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Commented] (BEAM-523) Minor typo in aggregator_test.py
[ https://issues.apache.org/jira/browse/BEAM-523?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15410033#comment-15410033 ] ASF GitHub Bot commented on BEAM-523: - GitHub user shlee0605 opened a pull request: https://github.com/apache/incubator-beam/pull/798 [BEAM-523] Fixed typo in aggregator_test.py Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [x] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [x] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [x] Replace `` in the title with the actual Jira issue number, if there is one. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- You can merge this pull request into a Git repository by running: $ git pull https://github.com/shlee0605/incubator-beam aggregator-typo Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/798.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 #798 commit 378f67e1cb5082a3869ce8fbf17adeba30924ead Author: Seunghyun Lee Date: 2016-08-05T20:35:12Z fixed typo in aggregator_test.py > Minor typo in aggregator_test.py > > > Key: BEAM-523 > URL: https://issues.apache.org/jira/browse/BEAM-523 > Project: Beam > Issue Type: Bug > Components: sdk-py >Reporter: Frank Yellin >Priority: Trivial > Labels: starter > Original Estimate: 10m > Remaining Estimate: 10m > > aggregators is repeatedly misspelled as aggeregators. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] incubator-beam pull request #799: Use EqualsTester in ProxyInvocationHandler...
GitHub user tgroh opened a pull request: https://github.com/apache/incubator-beam/pull/799 Use EqualsTester in ProxyInvocationHandlerTest Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [ ] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [ ] Replace `` in the title with the actual Jira issue number, if there is one. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- This tests the equality groups instead of manual assertions, including same-reference equality. You can merge this pull request into a Git repository by running: $ git pull https://github.com/tgroh/incubator-beam equals_test Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/799.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 #799 commit 6b5cb3b83ffb0bb348122767a7d8a65bdf89a45e Author: Thomas Groh Date: 2016-08-05T22:21:12Z Use EqualsTester in ProxyInvocationHandlerTest This tests the equality groups instead of manual assertions, including same-reference equality. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-beam pull request #800: [BEAM-498] Add OldDoFn -> DoFn adapter and...
GitHub user kennknowles opened a pull request: https://github.com/apache/incubator-beam/pull/800 [BEAM-498] Add OldDoFn -> DoFn adapter and expose raw DoFn in ParDo Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [x] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [x] Replace `` in the title with the actual Jira issue number, if there is one. - [x] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- Previously, `ParDo.of(DoFn)` would immediately wrap it into an `OldDoFn` so the new `DoFn` could not be recovered, since After this change, the new `DoFn` is temporarily available via `#getNewFn()`, which will eventually overwrite `#getFn()`. When `ParDo` is instantiated via `ParDo.of(OldDoFn)`, a wrapper in the reverse direction will create a new `DoFn` that can be accessed via the same method. So a `ParDo` transform temporarily always carries both, just so one need not reason about which is present. This allows consumers of the `fn` inside a `ParDo` to being porting their logic to unconditionally use `DoFn`. R: @bjchambers You can merge this pull request into a Git repository by running: $ git pull https://github.com/kennknowles/incubator-beam ParDoTest Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/800.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 #800 commit 22e629d3ae647b0a32aa8367a0e711e44c7ce31a Author: Kenneth Knowles Date: 2016-08-05T22:21:23Z Add OldDoFn -> DoFn adapter and expose raw DoFn in ParDo Previously, ParDo.of(DoFn) would immediately wrap it into an OldDoFn so the new DoFn could not be recovered, since After this change, the new DoFn is temporarily available via #getNewFn(), which will eventually overwrite #getFn(). When ParDo is instantiated via ParDo.of(OldDoFn), a wrapper in the reverse direction will create a new DoFn that can be accessed via the same method. So a ParDo transform temporarily always carries both. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Commented] (BEAM-498) Make DoFnWithContext the new DoFn
[ https://issues.apache.org/jira/browse/BEAM-498?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15410192#comment-15410192 ] ASF GitHub Bot commented on BEAM-498: - GitHub user kennknowles opened a pull request: https://github.com/apache/incubator-beam/pull/800 [BEAM-498] Add OldDoFn -> DoFn adapter and expose raw DoFn in ParDo Be sure to do all of the following to help us incorporate your contribution quickly and easily: - [x] Make sure the PR title is formatted like: `[BEAM-] Description of pull request` - [ ] Make sure tests pass via `mvn clean verify`. (Even better, enable Travis-CI on your fork and ensure the whole test matrix passes). - [x] Replace `` in the title with the actual Jira issue number, if there is one. - [x] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.txt). --- Previously, `ParDo.of(DoFn)` would immediately wrap it into an `OldDoFn` so the new `DoFn` could not be recovered, since After this change, the new `DoFn` is temporarily available via `#getNewFn()`, which will eventually overwrite `#getFn()`. When `ParDo` is instantiated via `ParDo.of(OldDoFn)`, a wrapper in the reverse direction will create a new `DoFn` that can be accessed via the same method. So a `ParDo` transform temporarily always carries both, just so one need not reason about which is present. This allows consumers of the `fn` inside a `ParDo` to being porting their logic to unconditionally use `DoFn`. R: @bjchambers You can merge this pull request into a Git repository by running: $ git pull https://github.com/kennknowles/incubator-beam ParDoTest Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/800.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 #800 commit 22e629d3ae647b0a32aa8367a0e711e44c7ce31a Author: Kenneth Knowles Date: 2016-08-05T22:21:23Z Add OldDoFn -> DoFn adapter and expose raw DoFn in ParDo Previously, ParDo.of(DoFn) would immediately wrap it into an OldDoFn so the new DoFn could not be recovered, since After this change, the new DoFn is temporarily available via #getNewFn(), which will eventually overwrite #getFn(). When ParDo is instantiated via ParDo.of(OldDoFn), a wrapper in the reverse direction will create a new DoFn that can be accessed via the same method. So a ParDo transform temporarily always carries both. > Make DoFnWithContext the new DoFn > - > > Key: BEAM-498 > URL: https://issues.apache.org/jira/browse/BEAM-498 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core >Reporter: Kenneth Knowles >Assignee: Kenneth Knowles > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] incubator-beam pull request #801: Added unit tests and error handling in rem...
GitHub user ianzhou1 opened a pull request: https://github.com/apache/incubator-beam/pull/801 Added unit tests and error handling in removeTemporaryTables You can merge this pull request into a Git repository by running: $ git pull https://github.com/ianzhou1/incubator-beam BigQueryExtra Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-beam/pull/801.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 #801 commit 3214c3812bd07a10c3093cdf06e140f2ddc2ba63 Author: Ian Zhou Date: 2016-08-05T22:31:59Z Added unit tests and error handling in removeTemporaryTables --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-beam pull request #640: Prototype simplified all-dynamic display d...
Github user swegner closed the pull request at: https://github.com/apache/incubator-beam/pull/640 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-beam pull request #577: Job Messages Prototypes
Github user swegner closed the pull request at: https://github.com/apache/incubator-beam/pull/577 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Created] (BEAM-536) Aggregator.py. More misleading documentation. More bad documentation
Frank Yellin created BEAM-536: - Summary: Aggregator.py. More misleading documentation. More bad documentation Key: BEAM-536 URL: https://issues.apache.org/jira/browse/BEAM-536 Project: Beam Issue Type: Bug Reporter: Frank Yellin Priority: Minor The last paragraph of the documentation for Aggregator is: You can also query the combined value(s) of an aggregator by calling aggregated_value() or aggregated_values() on the result object returned after running a pipeline. There are multiple problems in this one sentence! #1) There is no such method aggregated_value() that I can find anywhere. #2) DirectRunner implements aggregated_values(), but DirectPipelineRunner does not. The latter is the far more interesting case. #3) When I use a BlockingDirectPipelineRunner and ask for its aggregated_values(), I get an error message indicating that this is not implemented in DirectPipelineRunner. Very confusing since I never asked for a DirectPipelineRunner. It is clear that this is because BlockingDirectPipelineRunner is a method rather than a class. Is this really the right thing? Will there be other confusing error messages. #4) The documentation for aggregated_values() says "returns a dict of step names to values of the aggregator." I have no idea what a "step" means in this context. In practice, it seems to be a single-element dictionary whose key is 'user--' prefixed onto the aggregator name. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (BEAM-536) Aggregator.py. More misleading documentation. More bad documentation
[ https://issues.apache.org/jira/browse/BEAM-536?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15410353#comment-15410353 ] Ahmet Altay commented on BEAM-536: -- #1 - The comment should be cleaned. #2 - Tracking issue: https://issues.apache.org/jira/browse/BEAM-531 #3 - BlockingDataflowPipelineRunner is being removed for Java (https://github.com/apache/incubator-beam/pull/762) . It is being replaced with an optional set of wait...() methods on the result. We should do the same thing in the Python SDK. __str__ and __repr__ methods of DataflowPipelineRunner also use class name (https://github.com/aaltay/incubator-beam/blob/python-sdk/sdks/python/apache_beam/runners/dataflow_runner.py#L651). So printing the BlockingDataflowPipelineRunner object will use the wrong name. #4 - This also needs doc improvements. (related javadoc https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/runners/AggregatorPipelineExtractor.html#getAggregatorSteps--) User counters are by default prefixed with user-, there might be non user- aggregators once DataflowPipelineRunner implements this. I believe #1 and #4 can be tracked here for documentation changes. #3 requires a new bug of its own. > Aggregator.py. More misleading documentation. More bad documentation > -- > > Key: BEAM-536 > URL: https://issues.apache.org/jira/browse/BEAM-536 > Project: Beam > Issue Type: Bug >Reporter: Frank Yellin >Priority: Minor > > The last paragraph of the documentation for Aggregator is: > You can also query the combined value(s) of an aggregator by calling > aggregated_value() or aggregated_values() on the result object returned after > running a pipeline. > There are multiple problems in this one sentence! > #1) There is no such method aggregated_value() that I can find anywhere. > #2) DirectRunner implements aggregated_values(), but DirectPipelineRunner > does not. The latter is the far more interesting case. > #3) When I use a BlockingDirectPipelineRunner and ask for its > aggregated_values(), I get an error message indicating that this is not > implemented in DirectPipelineRunner. Very confusing since I never asked for > a DirectPipelineRunner. > It is clear that this is because BlockingDirectPipelineRunner is a method > rather than a class. Is this really the right thing? Will there be other > confusing error messages. > #4) The documentation for aggregated_values() says "returns a dict of step > names to values of the aggregator." I have no idea what a "step" means in > this context. In practice, it seems to be a single-element dictionary whose > key is 'user--' prefixed onto the aggregator name. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (BEAM-536) Aggregator.py. More misleading documentation. More bad documentation
[ https://issues.apache.org/jira/browse/BEAM-536?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ahmet Altay updated BEAM-536: - Component/s: sdk-py > Aggregator.py. More misleading documentation. More bad documentation > -- > > Key: BEAM-536 > URL: https://issues.apache.org/jira/browse/BEAM-536 > Project: Beam > Issue Type: Bug > Components: sdk-py >Reporter: Frank Yellin >Priority: Minor > > The last paragraph of the documentation for Aggregator is: > You can also query the combined value(s) of an aggregator by calling > aggregated_value() or aggregated_values() on the result object returned after > running a pipeline. > There are multiple problems in this one sentence! > #1) There is no such method aggregated_value() that I can find anywhere. > #2) DirectRunner implements aggregated_values(), but DirectPipelineRunner > does not. The latter is the far more interesting case. > #3) When I use a BlockingDirectPipelineRunner and ask for its > aggregated_values(), I get an error message indicating that this is not > implemented in DirectPipelineRunner. Very confusing since I never asked for > a DirectPipelineRunner. > It is clear that this is because BlockingDirectPipelineRunner is a method > rather than a class. Is this really the right thing? Will there be other > confusing error messages. > #4) The documentation for aggregated_values() says "returns a dict of step > names to values of the aggregator." I have no idea what a "step" means in > this context. In practice, it seems to be a single-element dictionary whose > key is 'user--' prefixed onto the aggregator name. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[27/51] [abbrv] incubator-beam git commit: The new DoFn is not @Experimental
The new DoFn is not @Experimental
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e73d163d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e73d163d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e73d163d
Branch: refs/heads/python-sdk
Commit: e73d163d5a220e91e88da4055bc45c18bfc9e930
Parents: b8d7559
Author: Kenneth Knowles
Authored: Thu Aug 4 10:01:28 2016 -0700
Committer: Dan Halperin
Committed: Thu Aug 4 11:47:31 2016 -0700
--
.../core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java| 2 --
1 file changed, 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e73d163d/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
index eb6753c..6f9a6b6 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
@@ -21,7 +21,6 @@ import static
com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
-import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import org.apache.beam.sdk.transforms.OldDoFn.DelegatingAggregator;
@@ -83,7 +82,6 @@ import java.util.Map;
* @param the type of the (main) input elements
* @param the type of the (main) output elements
*/
-@Experimental
public abstract class DoFn implements Serializable,
HasDisplayData {
/** Information accessible to all methods in this {@code DoFn}. */
[42/51] [abbrv] incubator-beam git commit: Port easy transforms to new DoFn
Port easy transforms to new DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/47341e11
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/47341e11
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/47341e11
Branch: refs/heads/python-sdk
Commit: 47341e113334827101ddbf775c69ae34d178cd8f
Parents: 269fbf3
Author: Kenneth Knowles
Authored: Wed Aug 3 20:27:28 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../java/org/apache/beam/sdk/transforms/Count.java | 4 ++--
.../java/org/apache/beam/sdk/transforms/Create.java | 4 ++--
.../apache/beam/sdk/transforms/FlatMapElements.java | 4 ++--
.../org/apache/beam/sdk/transforms/Flatten.java | 4 ++--
.../java/org/apache/beam/sdk/transforms/Keys.java | 4 ++--
.../java/org/apache/beam/sdk/transforms/KvSwap.java | 4 ++--
.../org/apache/beam/sdk/transforms/MapElements.java | 4 ++--
.../org/apache/beam/sdk/transforms/Partition.java | 4 ++--
.../beam/sdk/transforms/RemoveDuplicates.java | 4 ++--
.../java/org/apache/beam/sdk/transforms/Sample.java | 6 +++---
.../java/org/apache/beam/sdk/transforms/Values.java | 4 ++--
.../java/org/apache/beam/sdk/transforms/View.java | 8
.../org/apache/beam/sdk/transforms/WithKeys.java| 4 ++--
.../apache/beam/sdk/transforms/WithTimestamps.java | 6 +++---
.../beam/sdk/transforms/join/CoGroupByKey.java | 16
15 files changed, 40 insertions(+), 40 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/47341e11/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Count.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Count.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Count.java
index 7601ffc..ac59c76 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Count.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Count.java
@@ -107,8 +107,8 @@ public class Count {
public PCollection> apply(PCollection input) {
return
input
- .apply("Init", ParDo.of(new OldDoFn>() {
-@Override
+ .apply("Init", ParDo.of(new DoFn>() {
+@ProcessElement
public void processElement(ProcessContext c) {
c.output(KV.of(c.element(), (Void) null));
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/47341e11/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
index fb7f784..08d0a7a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
@@ -486,8 +486,8 @@ public class Create {
this.elementCoder = elementCoder;
}
-private static class ConvertTimestamps extends
OldDoFn, T> {
- @Override
+private static class ConvertTimestamps extends
DoFn, T> {
+ @ProcessElement
public void processElement(ProcessContext c) {
c.outputWithTimestamp(c.element().getValue(),
c.element().getTimestamp());
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/47341e11/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
index b48da38..694592e 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
@@ -133,9 +133,9 @@ extends PTransform,
PCollection> {
@Override
public PCollection apply(PCollection input) {
-return input.apply("Map", ParDo.of(new OldDoFn() {
+return input.apply("Map", ParDo.of(new DoFn() {
private static final long serialVersionUID = 0L;
- @Override
+ @ProcessElement
public void processElement(ProcessContext c) {
for (OutputT element : fn.apply(c.element())) {
c.output(element);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/47341e11/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/bea
[33/51] [abbrv] incubator-beam git commit: Closes #781
Closes #781 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/74c5e5e1 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/74c5e5e1 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/74c5e5e1 Branch: refs/heads/python-sdk Commit: 74c5e5e1b546759a082d4e4785d46f9d56fa1052 Parents: 734bfb9 71e027d Author: Dan Halperin Authored: Thu Aug 4 11:54:31 2016 -0700 Committer: Dan Halperin Committed: Thu Aug 4 11:54:31 2016 -0700 -- .../examples/complete/StreamingWordExtract.java | 18 ++-- .../apache/beam/examples/complete/TfIdf.java| 28 +- .../examples/complete/TopWikipediaSessions.java | 31 +--- .../examples/complete/TrafficMaxLaneFlow.java | 16 +- .../beam/examples/complete/TrafficRoutes.java | 20 ++--- .../examples/cookbook/BigQueryTornadoes.java| 10 +++ .../cookbook/CombinePerKeyExamples.java | 10 +++ .../examples/cookbook/DatastoreWordCount.java | 14 - .../beam/examples/cookbook/FilterExamples.java | 20 ++--- .../beam/examples/cookbook/JoinExamples.java| 18 ++-- .../examples/cookbook/MaxPerKeyExamples.java| 10 +++ .../beam/examples/cookbook/TriggerExample.java | 25 .../org/apache/beam/examples/WordCountTest.java | 3 +- .../examples/complete/AutoCompleteTest.java | 6 ++-- .../examples/cookbook/TriggerExampleTest.java | 6 ++-- .../beam/examples/complete/game/GameStats.java | 22 +++--- .../beam/examples/complete/game/UserScore.java | 6 ++-- .../examples/complete/game/UserScoreTest.java | 2 +- 18 files changed, 130 insertions(+), 135 deletions(-) --
[24/51] [abbrv] incubator-beam git commit: Exclude "uncallable method of anonymous class" errors from findbugs
Exclude "uncallable method of anonymous class" errors from findbugs These errors are spurious in the presence of reflective invocations, such as that for the new DoFn. Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8a2cf609 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8a2cf609 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8a2cf609 Branch: refs/heads/python-sdk Commit: 8a2cf609ec0f3ac74a4c33a89cd852aee87cd0a1 Parents: 34d5012 Author: Kenneth Knowles Authored: Wed Aug 3 21:23:22 2016 -0700 Committer: Kenneth Knowles Committed: Thu Aug 4 11:06:10 2016 -0700 -- sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml | 2 ++ 1 file changed, 2 insertions(+) -- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8a2cf609/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml -- diff --git a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml index 28fd0df..f117fbf 100644 --- a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml +++ b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml @@ -23,6 +23,8 @@ + +
[41/51] [abbrv] incubator-beam git commit: Port easy I/O transforms to new DoFn
Port easy I/O transforms to new DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/269fbf38
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/269fbf38
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/269fbf38
Branch: refs/heads/python-sdk
Commit: 269fbf386454ea77845e54764a125edba7039b03
Parents: ef5e31f
Author: Kenneth Knowles
Authored: Wed Aug 3 20:22:26 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../beam/runners/dataflow/DataflowRunner.java | 3 +-
.../java/org/apache/beam/sdk/io/PubsubIO.java | 14
.../apache/beam/sdk/io/PubsubUnboundedSink.java | 17 +
.../beam/sdk/io/PubsubUnboundedSource.java | 7 ++--
.../beam/sdk/io/gcp/bigquery/BigQueryIO.java| 36 +---
.../beam/sdk/io/gcp/bigtable/BigtableIO.java| 12 +++
.../beam/sdk/io/gcp/datastore/V1Beta3.java | 18 +-
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 10 +++---
.../sdk/io/gcp/bigtable/BigtableWriteIT.java| 6 ++--
.../sdk/io/gcp/datastore/V1Beta3TestUtil.java | 9 +++--
.../java/org/apache/beam/sdk/io/jms/JmsIO.java | 10 +++---
.../org/apache/beam/sdk/io/kafka/KafkaIO.java | 19 +--
.../apache/beam/sdk/io/kafka/KafkaIOTest.java | 10 +++---
13 files changed, 82 insertions(+), 89 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/269fbf38/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
--
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index abcf415..fadd9c7 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -78,6 +78,7 @@ import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.OldDoFn;
@@ -2715,7 +2716,7 @@ public class DataflowRunner extends
PipelineRunner {
@Nullable
private PTransform transform;
@Nullable
-private OldDoFn doFn;
+private DoFn doFn;
/**
* Builds an instance of this class from the overridden transform.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/269fbf38/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
--
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
index 1902bca..2b27175 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
@@ -25,7 +25,7 @@ import org.apache.beam.sdk.coders.VoidCoder;
import org.apache.beam.sdk.options.PubsubOptions;
import org.apache.beam.sdk.runners.PipelineRunner;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.display.DisplayData;
@@ -709,11 +709,11 @@ public class PubsubIO {
*
* Public so can be suppressed by runners.
*/
- public class PubsubBoundedReader extends OldDoFn {
+ public class PubsubBoundedReader extends DoFn {
private static final int DEFAULT_PULL_SIZE = 100;
private static final int ACK_TIMEOUT_SEC = 60;
-@Override
+@ProcessElement
public void processElement(ProcessContext c) throws IOException {
try (PubsubClient pubsubClient =
FACTORY.newClient(timestampLabel, idLabel,
@@ -998,12 +998,12 @@ public class PubsubIO {
*
* Public so can be suppressed by runners.
*/
- public class PubsubBoundedWriter extends OldDoFn {
+ public class PubsubBoundedWriter extends DoFn {
private static final int MAX_PUBLISH_BATCH_SIZE = 100;
private transient List output;
private transient PubsubClient pubsubClient;
-@Override
+@StartBundle
public void startBund
[31/51] [abbrv] incubator-beam git commit: Closes #767
Closes #767 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/734bfb9a Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/734bfb9a Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/734bfb9a Branch: refs/heads/python-sdk Commit: 734bfb9afc29a4d760cb8d996638021f0d55b8c6 Parents: be2758c 8942c0a Author: Dan Halperin Authored: Thu Aug 4 11:50:09 2016 -0700 Committer: Dan Halperin Committed: Thu Aug 4 11:50:09 2016 -0700 -- examples/java/pom.xml | 7 +++ 1 file changed, 7 insertions(+) --
[29/51] [abbrv] incubator-beam git commit: add back unnecessary changes
add back unnecessary changes
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8942c0a9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8942c0a9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8942c0a9
Branch: refs/heads/python-sdk
Commit: 8942c0a90eb12a1f1f04c2fc47e9cbe1ed2be5b8
Parents: d7a02a1
Author: Mark Liu
Authored: Thu Aug 4 10:26:04 2016 -0700
Committer: Dan Halperin
Committed: Thu Aug 4 11:50:08 2016 -0700
--
examples/java/pom.xml | 1 +
1 file changed, 1 insertion(+)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8942c0a9/examples/java/pom.xml
--
diff --git a/examples/java/pom.xml b/examples/java/pom.xml
index dca2318..10e7503 100644
--- a/examples/java/pom.xml
+++ b/examples/java/pom.xml
@@ -275,6 +275,7 @@
beam-runners-direct-java
${project.version}
runtime
+ true
[40/51] [abbrv] incubator-beam git commit: Propagate getAllowedTimestampSkew from DoFn to its adapter
Propagate getAllowedTimestampSkew from DoFn to its adapter
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a1c06d71
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a1c06d71
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a1c06d71
Branch: refs/heads/python-sdk
Commit: a1c06d71876384722982ec24da1607e41af653d9
Parents: 0b18652
Author: Kenneth Knowles
Authored: Thu Aug 4 14:56:19 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../java/org/apache/beam/sdk/transforms/DoFnReflector.java | 6 ++
1 file changed, 6 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a1c06d71/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
index 9bdfde8..c6168b3 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnReflector.java
@@ -72,6 +72,7 @@ import net.bytebuddy.jar.asm.Label;
import net.bytebuddy.jar.asm.MethodVisitor;
import net.bytebuddy.jar.asm.Opcodes;
import net.bytebuddy.matcher.ElementMatchers;
+import org.joda.time.Duration;
import org.joda.time.Instant;
import java.io.IOException;
@@ -731,6 +732,11 @@ public abstract class DoFnReflector {
}
@Override
+public Duration getAllowedTimestampSkew() {
+ return fn.getAllowedTimestampSkew();
+}
+
+@Override
public void populateDisplayData(DisplayData.Builder builder) {
builder.include(fn);
}
[50/51] [abbrv] incubator-beam git commit: Merge remote-tracking branch 'origin/master' into python-sdk
Merge remote-tracking branch 'origin/master' into python-sdk Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c3988118 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c3988118 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c3988118 Branch: refs/heads/python-sdk Commit: c39881186d4625d935efd9a717d9978cd1252979 Parents: 65152ca c584b37 Author: Ahmet Altay Authored: Fri Aug 5 15:41:30 2016 -0700 Committer: Ahmet Altay Committed: Fri Aug 5 15:41:30 2016 -0700 -- .travis.yml |6 +- KEYS| 141 - NOTICE |1 - README.md | 32 +- examples/java/pom.xml | 73 +- .../beam/examples/DebuggingWordCount.java |4 +- .../apache/beam/examples/MinimalWordCount.java | 30 +- .../apache/beam/examples/WindowedWordCount.java | 91 +- .../org/apache/beam/examples/WordCount.java |6 +- .../examples/common/DataflowExampleOptions.java | 37 - .../examples/common/DataflowExampleUtils.java | 491 -- .../common/ExampleBigQueryTableOptions.java | 10 +- .../beam/examples/common/ExampleOptions.java| 83 + ...xamplePubsubTopicAndSubscriptionOptions.java | 10 +- .../common/ExamplePubsubTopicOptions.java | 12 +- .../beam/examples/common/ExampleUtils.java | 390 ++ .../examples/common/PubsubFileInjector.java | 10 +- .../beam/examples/complete/AutoComplete.java| 91 +- .../examples/complete/StreamingWordExtract.java | 56 +- .../apache/beam/examples/complete/TfIdf.java| 28 +- .../examples/complete/TopWikipediaSessions.java | 38 +- .../examples/complete/TrafficMaxLaneFlow.java | 90 +- .../beam/examples/complete/TrafficRoutes.java | 90 +- .../examples/cookbook/BigQueryTornadoes.java|6 +- .../cookbook/CombinePerKeyExamples.java |6 +- .../examples/cookbook/DatastoreWordCount.java | 67 +- .../beam/examples/cookbook/DeDupExample.java|5 +- .../beam/examples/cookbook/FilterExamples.java | 17 +- .../beam/examples/cookbook/JoinExamples.java| 14 +- .../examples/cookbook/MaxPerKeyExamples.java|6 +- .../beam/examples/cookbook/TriggerExample.java | 97 +- .../org/apache/beam/examples/WordCountIT.java | 118 +- .../org/apache/beam/examples/WordCountTest.java |3 +- .../examples/complete/AutoCompleteTest.java |2 +- .../examples/cookbook/TriggerExampleTest.java |2 +- examples/java8/pom.xml | 81 +- .../beam/examples/MinimalWordCountJava8.java| 30 +- .../beam/examples/complete/game/GameStats.java | 57 +- .../examples/complete/game/HourlyTeamScore.java |9 +- .../examples/complete/game/LeaderBoard.java | 23 +- .../beam/examples/complete/game/UserScore.java |4 +- .../complete/game/injector/InjectorUtils.java |6 +- .../injector/RetryHttpInitializerWrapper.java |5 +- .../complete/game/utils/WriteToBigQuery.java| 20 +- .../game/utils/WriteWindowedToBigQuery.java | 16 +- .../complete/game/HourlyTeamScoreTest.java |2 +- .../examples/complete/game/UserScoreTest.java |2 +- examples/pom.xml|4 +- pom.xml | 265 +- runners/core-java/pom.xml | 55 +- .../core/GroupAlsoByWindowViaWindowSetDoFn.java | 28 +- .../core/UnboundedReadFromBoundedSource.java| 542 ++ .../apache/beam/runners/core/package-info.java | 22 + .../org/apache/beam/sdk/util/AssignWindows.java | 46 + .../apache/beam/sdk/util/AssignWindowsDoFn.java | 80 + .../beam/sdk/util/BatchTimerInternals.java | 141 + .../org/apache/beam/sdk/util/DoFnRunner.java| 63 + .../apache/beam/sdk/util/DoFnRunnerBase.java| 551 ++ .../org/apache/beam/sdk/util/DoFnRunners.java | 222 + .../beam/sdk/util/GroupAlsoByWindowsDoFn.java | 63 + .../GroupAlsoByWindowsViaOutputBufferDoFn.java | 100 + .../sdk/util/GroupByKeyViaGroupByKeyOnly.java | 269 + .../sdk/util/LateDataDroppingDoFnRunner.java| 147 + .../org/apache/beam/sdk/util/NonEmptyPanes.java | 150 + .../apache/beam/sdk/util/PaneInfoTracker.java | 158 + .../sdk/util/PushbackSideInputDoFnRunner.java | 115 + .../java/org/apache/beam/sdk/util/ReduceFn.java | 130 + .../beam/sdk/util/ReduceFnContextFactory.java | 497 ++ .../apache/beam/sdk/util/ReduceFnRunner.java| 988 .../apache/beam/sdk/util/SimpleDoFnRunner.java | 55 + .../apache/beam/sdk/util/SystemReduceFn.java| 139 + .../org/apache/beam/sdk/util/TriggerRunner.java | 245 + .../org/apache/beam/sdk/util/WatermarkHold.java | 540 ++ .../util/common/ElementByteSizeObservable.java |
[49/51] [abbrv] incubator-beam git commit: This closes #795
This closes #795 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c584b37b Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c584b37b Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c584b37b Branch: refs/heads/python-sdk Commit: c584b37b8ac4e863bad83a766b6871ccc7135334 Parents: 2b5c6bc acf71d3 Author: Kenneth Knowles Authored: Fri Aug 5 10:40:17 2016 -0700 Committer: Kenneth Knowles Committed: Fri Aug 5 10:40:17 2016 -0700 -- .../beam/runners/direct/EvaluationContext.java | 14 ++--- .../direct/ExecutorServiceParallelExecutor.java | 5 ++-- .../beam/runners/direct/TransformExecutor.java | 2 +- .../beam/runners/direct/DirectRunnerTest.java | 31 4 files changed, 44 insertions(+), 8 deletions(-) --
[18/51] [abbrv] incubator-beam git commit: SparkRunner calls pipeline.run
SparkRunner calls pipeline.run
* Remove SparkStreamingPipelineOptions.
* Run pipeline with Pipeline.run().
* Better EmbeddedKafka.
* Avoid NPE if factory wasn't created.
* Let EmbeddedKafka/Zookeeper discover ports on their own.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ac0875de
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ac0875de
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ac0875de
Branch: refs/heads/python-sdk
Commit: ac0875de84085e1298575d0887e83e5deee5f418
Parents: c314e67
Author: Sela
Authored: Wed Jul 27 23:11:37 2016 +0300
Committer: Dan Halperin
Committed: Wed Aug 3 22:31:48 2016 -0700
--
.../runners/spark/SparkPipelineOptions.java | 6
.../apache/beam/runners/spark/SparkRunner.java | 14 +++--
.../runners/spark/SparkRunnerRegistrar.java | 6 ++--
.../spark/SparkStreamingPipelineOptions.java| 32
.../beam/runners/spark/TestSparkRunner.java | 2 --
.../apache/beam/runners/spark/DeDupTest.java| 2 +-
.../beam/runners/spark/EmptyInputTest.java | 2 +-
.../beam/runners/spark/SimpleWordCountTest.java | 4 +--
.../runners/spark/SparkRunnerRegistrarTest.java | 2 +-
.../apache/beam/runners/spark/TfIdfTest.java| 2 +-
.../beam/runners/spark/io/AvroPipelineTest.java | 2 +-
.../beam/runners/spark/io/NumShardsTest.java| 2 +-
.../io/hadoop/HadoopFileFormatPipelineTest.java | 2 +-
.../spark/translation/CombineGloballyTest.java | 2 +-
.../spark/translation/CombinePerKeyTest.java| 2 +-
.../spark/translation/DoFnOutputTest.java | 6 ++--
.../translation/MultiOutputWordCountTest.java | 2 +-
.../spark/translation/SerializationTest.java| 2 +-
.../spark/translation/SideEffectsTest.java | 8 ++---
.../streaming/FlattenStreamingTest.java | 8 ++---
.../streaming/KafkaStreamingTest.java | 13
.../streaming/SimpleStreamingWordCountTest.java | 8 ++---
.../streaming/utils/EmbeddedKafkaCluster.java | 4 ++-
23 files changed, 49 insertions(+), 84 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac0875de/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
--
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
index 4bb2a57..6ef3741 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
@@ -33,4 +33,10 @@ public interface SparkPipelineOptions extends
PipelineOptions, StreamingOptions,
@Default.String("local[1]")
String getSparkMaster();
void setSparkMaster(String master);
+
+ @Description("Timeout to wait (in msec) for a streaming execution to stop,
-1 runs until "
+ + "execution is stopped")
+ @Default.Long(-1)
+ Long getTimeout();
+ void setTimeout(Long batchInterval);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac0875de/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
--
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
index dfda987..d994ec4 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
@@ -69,8 +69,6 @@ import org.slf4j.LoggerFactory;
* options.setSparkMaster("spark://host:port");
* EvaluationResult result = SparkRunner.create(options).run(p);
* }
- *
- * To create a Spark streaming pipeline runner use {@link
SparkStreamingPipelineOptions}
*/
public final class SparkRunner extends PipelineRunner {
@@ -146,12 +144,6 @@ public final class SparkRunner extends
PipelineRunner {
@Override
public EvaluationResult run(Pipeline pipeline) {
try {
- // validate streaming configuration
- if (mOptions.isStreaming() && !(mOptions instanceof
SparkStreamingPipelineOptions)) {
-throw new RuntimeException("A streaming job must be configured with "
-+ SparkStreamingPipelineOptions.class.getSimpleName() + ", found "
-+ mOptions.getClass().getSimpleName());
- }
LOG.info("Executing pipeline using the SparkRunner.");
JavaSparkContext jsc = SparkContextFactory.getSparkContext(mOptions
.getSparkMaster(), mOptions.getAppName());
@@ -179,6 +171,9 @@ public final class SparkRun
[09/51] [abbrv] incubator-beam git commit: Closes #758
Closes #758 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/9a329aad Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/9a329aad Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/9a329aad Branch: refs/heads/python-sdk Commit: 9a329aada82d3ca7f619b88eddea04bdd329d992 Parents: 388816a 3466a0e Author: Dan Halperin Authored: Wed Aug 3 18:25:53 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 18:25:53 2016 -0700 -- .../beam/examples/DebuggingWordCount.java | 2 +- .../apache/beam/examples/MinimalWordCount.java | 2 +- .../apache/beam/examples/WindowedWordCount.java | 4 +- .../org/apache/beam/examples/WordCount.java | 2 +- .../examples/common/PubsubFileInjector.java | 6 +- .../beam/examples/complete/AutoComplete.java| 14 +- .../examples/complete/StreamingWordExtract.java | 12 +- .../apache/beam/examples/complete/TfIdf.java| 16 +- .../examples/complete/TopWikipediaSessions.java | 12 +- .../examples/complete/TrafficMaxLaneFlow.java | 10 +- .../beam/examples/complete/TrafficRoutes.java | 12 +- .../examples/cookbook/BigQueryTornadoes.java| 6 +- .../cookbook/CombinePerKeyExamples.java | 6 +- .../examples/cookbook/DatastoreWordCount.java | 11 +- .../beam/examples/cookbook/FilterExamples.java | 12 +- .../beam/examples/cookbook/JoinExamples.java| 10 +- .../examples/cookbook/MaxPerKeyExamples.java| 6 +- .../beam/examples/cookbook/TriggerExample.java | 12 +- .../org/apache/beam/examples/WordCountTest.java | 2 +- .../examples/complete/AutoCompleteTest.java | 4 +- .../examples/cookbook/TriggerExampleTest.java | 4 +- .../beam/examples/complete/game/GameStats.java | 10 +- .../beam/examples/complete/game/UserScore.java | 4 +- .../complete/game/utils/WriteToBigQuery.java| 12 +- .../game/utils/WriteWindowedToBigQuery.java | 8 +- .../examples/complete/game/UserScoreTest.java | 2 +- .../core/GroupAlsoByWindowViaWindowSetDoFn.java | 12 +- .../core/UnboundedReadFromBoundedSource.java| 2 +- .../apache/beam/sdk/util/AssignWindowsDoFn.java | 10 +- .../org/apache/beam/sdk/util/DoFnRunner.java| 21 +- .../apache/beam/sdk/util/DoFnRunnerBase.java| 54 +- .../org/apache/beam/sdk/util/DoFnRunners.java | 24 +- .../beam/sdk/util/GroupAlsoByWindowsDoFn.java | 6 +- .../GroupAlsoByWindowsViaOutputBufferDoFn.java | 4 +- .../sdk/util/GroupByKeyViaGroupByKeyOnly.java | 6 +- .../sdk/util/LateDataDroppingDoFnRunner.java| 4 +- .../apache/beam/sdk/util/PaneInfoTracker.java | 1 - .../apache/beam/sdk/util/ReduceFnRunner.java| 4 +- .../apache/beam/sdk/util/SimpleDoFnRunner.java | 12 +- .../org/apache/beam/sdk/util/WatermarkHold.java | 1 - .../beam/sdk/util/ReduceFnRunnerTest.java | 1 + .../apache/beam/sdk/util/ReduceFnTester.java| 1 + .../beam/sdk/util/SimpleDoFnRunnerTest.java | 6 +- .../GroupAlsoByWindowEvaluatorFactory.java | 4 +- .../ImmutabilityCheckingBundleFactory.java | 4 +- .../beam/runners/direct/ParDoEvaluator.java | 4 +- .../direct/ParDoMultiEvaluatorFactory.java | 11 +- .../direct/ParDoSingleEvaluatorFactory.java | 11 +- .../direct/TransformEvaluatorFactory.java | 6 +- .../direct/WriteWithShardingFactory.java| 4 +- .../ConsumerTrackingPipelineVisitorTest.java| 22 +- .../beam/runners/direct/DirectRunnerTest.java | 24 +- .../ImmutabilityCheckingBundleFactoryTest.java | 6 +- .../ImmutabilityEnforcementFactoryTest.java | 6 +- .../direct/KeyedPValueTrackingVisitorTest.java | 6 +- .../beam/runners/direct/ParDoEvaluatorTest.java | 6 +- .../direct/ParDoMultiEvaluatorFactoryTest.java | 10 +- .../direct/ParDoSingleEvaluatorFactoryTest.java | 10 +- .../runners/direct/WatermarkManagerTest.java| 7 +- .../beam/runners/flink/examples/TFIDF.java | 16 +- .../beam/runners/flink/examples/WordCount.java | 4 +- .../flink/examples/streaming/AutoComplete.java | 16 +- .../flink/examples/streaming/JoinExamples.java | 8 +- .../examples/streaming/KafkaIOExamples.java | 4 +- .../KafkaWindowedWordCountExample.java | 6 +- .../examples/streaming/WindowedWordCount.java | 6 +- .../FlinkBatchTransformTranslators.java | 12 +- .../FlinkStreamingTransformTranslators.java | 9 +- .../functions/FlinkDoFnFunction.java| 10 +- .../FlinkMergingNonShuffleReduceFunction.java | 8 +- .../functions/FlinkMultiOutputDoFnFunction.java | 10 +- .../FlinkMultiOutputProcessContext.java | 6 +- .../functions/FlinkNoElementAssignContext.java | 8 +- .../functions/FlinkPartialReduceFunction.java | 8 +- .../functions/FlinkProcessContext.java | 16 +- .../functions/FlinkReduceFunction.java
[21/51] [abbrv] incubator-beam git commit: Closes #775
Closes #775 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/595d2d4b Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/595d2d4b Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/595d2d4b Branch: refs/heads/python-sdk Commit: 595d2d4baf1f539cbd51d53ae2dffd8fdbd825f7 Parents: 3144363 032e1fa Author: Dan Halperin Authored: Wed Aug 3 23:03:54 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 23:03:54 2016 -0700 -- .../src/main/resources/beam/checkstyle.xml | 5 + .../main/java/org/apache/beam/sdk/Pipeline.java | 2 +- .../org/apache/beam/sdk/coders/JAXBCoder.java | 2 +- .../java/org/apache/beam/sdk/io/AvroIO.java | 3 +-- .../org/apache/beam/sdk/io/FileBasedSink.java | 2 +- .../java/org/apache/beam/sdk/io/PubsubIO.java | 6 +++--- .../org/apache/beam/sdk/io/package-info.java| 3 +-- .../beam/sdk/options/PipelineOptions.java | 2 +- .../sdk/options/PipelineOptionsFactory.java | 2 +- .../beam/sdk/testing/SerializableMatchers.java | 22 +++- .../apache/beam/sdk/transforms/CombineFns.java | 2 +- .../beam/sdk/transforms/DoFnReflector.java | 4 ++-- .../apache/beam/sdk/transforms/DoFnTester.java | 2 +- .../org/apache/beam/sdk/transforms/Filter.java | 5 ++--- .../beam/sdk/transforms/SimpleFunction.java | 2 +- .../windowing/AfterProcessingTime.java | 5 +++-- .../beam/sdk/transforms/windowing/PaneInfo.java | 2 +- .../beam/sdk/transforms/windowing/Window.java | 2 +- .../org/apache/beam/sdk/util/TimeDomain.java| 2 +- .../apache/beam/sdk/util/WindowingStrategy.java | 6 +++--- .../util/common/ElementByteSizeObserver.java| 2 +- .../CopyOnAccessInMemoryStateInternals.java | 4 ++-- .../sdk/util/state/StateInternalsFactory.java | 4 ++-- .../beam/sdk/util/state/StateNamespaces.java| 2 +- .../apache/beam/sdk/util/state/StateTags.java | 6 +++--- .../apache/beam/sdk/values/PCollectionList.java | 2 +- .../org/apache/beam/sdk/values/POutput.java | 2 +- .../org/apache/beam/sdk/values/TupleTag.java| 2 +- .../beam/sdk/io/gcp/bigquery/BigQueryIO.java| 1 - .../java/org/apache/beam/sdk/io/jms/JmsIO.java | 4 +--- .../org/apache/beam/sdk/io/kafka/KafkaIO.java | 2 +- 31 files changed, 57 insertions(+), 55 deletions(-) --
[34/51] [abbrv] incubator-beam git commit: This closes #711
This closes #711 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/fcf6b1d3 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/fcf6b1d3 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/fcf6b1d3 Branch: refs/heads/python-sdk Commit: fcf6b1d34e8b9da284e9857fecced2deb5a5ab1d Parents: 74c5e5e b9543b9 Author: Kenneth Knowles Authored: Thu Aug 4 12:49:08 2016 -0700 Committer: Kenneth Knowles Committed: Thu Aug 4 12:49:08 2016 -0700 -- runners/spark/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) --
[20/51] [abbrv] incubator-beam git commit: [BEAM-475] Javadoc has invalid references after the recent refactorings + other fixes
[BEAM-475] Javadoc has invalid references after the recent refactorings + other fixes * Remove useless semicolons * Remove unneeded java keywords/validations and fix Filter style * Fix invalid Javadoc references and some other documentation issues * Add rules for unused semicolons and overcomplicated boolean expresions Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/032e1fa6 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/032e1fa6 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/032e1fa6 Branch: refs/heads/python-sdk Commit: 032e1fa6b11a7474057df8ef553a5ba2d6cad63d Parents: 3144363 Author: Ismaël MejÃÂa Authored: Wed Aug 3 11:03:43 2016 +0200 Committer: Dan Halperin Committed: Wed Aug 3 23:03:30 2016 -0700 -- .../src/main/resources/beam/checkstyle.xml | 5 + .../main/java/org/apache/beam/sdk/Pipeline.java | 2 +- .../org/apache/beam/sdk/coders/JAXBCoder.java | 2 +- .../java/org/apache/beam/sdk/io/AvroIO.java | 3 +-- .../org/apache/beam/sdk/io/FileBasedSink.java | 2 +- .../java/org/apache/beam/sdk/io/PubsubIO.java | 6 +++--- .../org/apache/beam/sdk/io/package-info.java| 3 +-- .../beam/sdk/options/PipelineOptions.java | 2 +- .../sdk/options/PipelineOptionsFactory.java | 2 +- .../beam/sdk/testing/SerializableMatchers.java | 22 +++- .../apache/beam/sdk/transforms/CombineFns.java | 2 +- .../beam/sdk/transforms/DoFnReflector.java | 4 ++-- .../apache/beam/sdk/transforms/DoFnTester.java | 2 +- .../org/apache/beam/sdk/transforms/Filter.java | 5 ++--- .../beam/sdk/transforms/SimpleFunction.java | 2 +- .../windowing/AfterProcessingTime.java | 5 +++-- .../beam/sdk/transforms/windowing/PaneInfo.java | 2 +- .../beam/sdk/transforms/windowing/Window.java | 2 +- .../org/apache/beam/sdk/util/TimeDomain.java| 2 +- .../apache/beam/sdk/util/WindowingStrategy.java | 6 +++--- .../util/common/ElementByteSizeObserver.java| 2 +- .../CopyOnAccessInMemoryStateInternals.java | 4 ++-- .../sdk/util/state/StateInternalsFactory.java | 4 ++-- .../beam/sdk/util/state/StateNamespaces.java| 2 +- .../apache/beam/sdk/util/state/StateTags.java | 6 +++--- .../apache/beam/sdk/values/PCollectionList.java | 2 +- .../org/apache/beam/sdk/values/POutput.java | 2 +- .../org/apache/beam/sdk/values/TupleTag.java| 2 +- .../beam/sdk/io/gcp/bigquery/BigQueryIO.java| 1 - .../java/org/apache/beam/sdk/io/jms/JmsIO.java | 4 +--- .../org/apache/beam/sdk/io/kafka/KafkaIO.java | 2 +- 31 files changed, 57 insertions(+), 55 deletions(-) -- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/032e1fa6/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml -- diff --git a/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml b/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml index a437561..4bb7428 100644 --- a/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml +++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle.xml @@ -316,6 +316,11 @@ page at http://checkstyle.sourceforge.net/config.html --> + + + + +
[30/51] [abbrv] incubator-beam git commit: [BEAM-124] Flink Running WordCountIT Example
[BEAM-124] Flink Running WordCountIT Example
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d7a02a17
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d7a02a17
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d7a02a17
Branch: refs/heads/python-sdk
Commit: d7a02a17b494b6ca3ffa5ce10ef1c3572738a875
Parents: be2758c
Author: Mark Liu
Authored: Tue Aug 2 11:19:03 2016 -0700
Committer: Dan Halperin
Committed: Thu Aug 4 11:50:08 2016 -0700
--
examples/java/pom.xml | 8 +++-
1 file changed, 7 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d7a02a17/examples/java/pom.xml
--
diff --git a/examples/java/pom.xml b/examples/java/pom.xml
index d0a6b34..dca2318 100644
--- a/examples/java/pom.xml
+++ b/examples/java/pom.xml
@@ -275,7 +275,6 @@
beam-runners-direct-java
${project.version}
runtime
- true
@@ -285,6 +284,13 @@
+ org.apache.beam
+ beam-runners-flink_2.10
+ ${project.version}
+ runtime
+
+
+
org.slf4j
slf4j-jdk14
runtime
[28/51] [abbrv] incubator-beam git commit: Deprecate OldDoFn
Deprecate OldDoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/fb6d2c2e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/fb6d2c2e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/fb6d2c2e
Branch: refs/heads/python-sdk
Commit: fb6d2c2e3b97ca67450ea0ca84bbb40667b48a92
Parents: e73d163
Author: Kenneth Knowles
Authored: Thu Aug 4 10:02:22 2016 -0700
Committer: Dan Halperin
Committed: Thu Aug 4 11:47:31 2016 -0700
--
.../core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java | 2 ++
1 file changed, 2 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/fb6d2c2e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java
index f640442..443599a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java
@@ -72,7 +72,9 @@ import java.util.UUID;
*
* @param the type of the (main) input elements
* @param the type of the (main) output elements
+ * @deprecated Uses of {@link OldDoFn} should be replaced by the new {@link
DoFn}.
*/
+@Deprecated
public abstract class OldDoFn implements Serializable,
HasDisplayData {
/**
[10/51] [abbrv] incubator-beam git commit: Update Dataflow runner to worker container supporting OldDoFn
Update Dataflow runner to worker container supporting OldDoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/3466a0e7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/3466a0e7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/3466a0e7
Branch: refs/heads/python-sdk
Commit: 3466a0e72d1f025c2d4562635a20730778b497b0
Parents: e07c339
Author: Kenneth Knowles
Authored: Wed Aug 3 17:47:23 2016 -0700
Committer: Dan Halperin
Committed: Wed Aug 3 18:25:53 2016 -0700
--
.../java/org/apache/beam/runners/dataflow/DataflowRunner.java| 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3466a0e7/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
--
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index d762d50..abcf415 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -212,9 +212,9 @@ public class DataflowRunner extends
PipelineRunner {
// Default Docker container images that execute Dataflow worker harness,
residing in Google
// Container Registry, separately for Batch and Streaming.
public static final String BATCH_WORKER_HARNESS_CONTAINER_IMAGE =
- "dataflow.gcr.io/v1beta3/beam-java-batch:beam-master-20160714";
+ "dataflow.gcr.io/v1beta3/beam-java-batch:beam-master-20160804-dofn";
public static final String STREAMING_WORKER_HARNESS_CONTAINER_IMAGE =
- "dataflow.gcr.io/v1beta3/beam-java-streaming:beam-master-20160714";
+ "dataflow.gcr.io/v1beta3/beam-java-streaming:beam-master-20160804-dofn";
// The limit of CreateJob request size.
private static final int CREATE_JOB_REQUEST_LIMIT_BYTES = 10 * 1024 * 1024;
[48/51] [abbrv] incubator-beam git commit: This closes #757
This closes #757 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/2b5c6bcb Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/2b5c6bcb Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/2b5c6bcb Branch: refs/heads/python-sdk Commit: 2b5c6bcb2173d55f6b45f4134cd500d9cc9c1965 Parents: 8daf518 4ac5caf Author: Kenneth Knowles Authored: Fri Aug 5 10:09:25 2016 -0700 Committer: Kenneth Knowles Committed: Fri Aug 5 10:09:25 2016 -0700 -- .../beam/sdk/transforms/FlatMapElements.java| 126 +-- .../apache/beam/sdk/transforms/MapElements.java | 60 + .../beam/sdk/transforms/SimpleFunction.java | 34 + .../sdk/transforms/FlatMapElementsTest.java | 48 +++ .../beam/sdk/transforms/MapElementsTest.java| 84 + 5 files changed, 288 insertions(+), 64 deletions(-) --
[25/51] [abbrv] incubator-beam git commit: Closes #786
Closes #786 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/b8d7559f Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/b8d7559f Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/b8d7559f Branch: refs/heads/python-sdk Commit: b8d7559f6737a0cc6cb747fb9907237ca1215a6d Parents: 34d5012 8a2cf60 Author: Dan Halperin Authored: Thu Aug 4 11:46:18 2016 -0700 Committer: Dan Halperin Committed: Thu Aug 4 11:46:18 2016 -0700 -- sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml | 2 ++ 1 file changed, 2 insertions(+) --
[46/51] [abbrv] incubator-beam git commit: Add DirectRunner Reuse Test
Add DirectRunner Reuse Test
Two calls to run using the Direct Runner should be independent and
succeed independently.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/4546fd9c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/4546fd9c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/4546fd9c
Branch: refs/heads/python-sdk
Commit: 4546fd9c5e073eb33787faa302b8695dfd6e04aa
Parents: 7585cfc
Author: Thomas Groh
Authored: Fri Aug 5 09:58:59 2016 -0700
Committer: Thomas Groh
Committed: Fri Aug 5 10:04:21 2016 -0700
--
.../beam/runners/direct/DirectRunnerTest.java | 31
1 file changed, 31 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4546fd9c/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
--
diff --git
a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index 29dea32..1e73ec0 100644
---
a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -109,6 +109,37 @@ public class DirectRunnerTest implements Serializable {
result.awaitCompletion();
}
+ @Test
+ public void reusePipelineSucceeds() throws Throwable {
+Pipeline p = getPipeline();
+
+PCollection> counts =
+p.apply(Create.of("foo", "bar", "foo", "baz", "bar", "foo"))
+.apply(MapElements.via(new SimpleFunction() {
+ @Override
+ public String apply(String input) {
+return input;
+ }
+}))
+.apply(Count.perElement());
+PCollection countStrs =
+counts.apply(MapElements.via(new SimpleFunction,
String>() {
+ @Override
+ public String apply(KV input) {
+String str = String.format("%s: %s", input.getKey(),
input.getValue());
+return str;
+ }
+}));
+
+PAssert.that(countStrs).containsInAnyOrder("baz: 1", "bar: 2", "foo: 3");
+
+DirectPipelineResult result = ((DirectPipelineResult) p.run());
+result.awaitCompletion();
+
+DirectPipelineResult otherResult = ((DirectPipelineResult) p.run());
+otherResult.awaitCompletion();
+ }
+
@Test(timeout = 5000L)
public void byteArrayCountShouldSucceed() {
Pipeline p = getPipeline();
[17/51] [abbrv] incubator-beam git commit: Closes #783
Closes #783 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c314e670 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c314e670 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c314e670 Branch: refs/heads/python-sdk Commit: c314e670e0113cddc40af680f5ce8a5134d61e9a Parents: 9a329aa 14c6d99 Author: Dan Halperin Authored: Wed Aug 3 22:28:17 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 22:28:17 2016 -0700 -- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) --
[08/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
Rename DoFn to OldDoFn Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a64baf48 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a64baf48 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a64baf48 Branch: refs/heads/python-sdk Commit: a64baf4878f28e98da696dacc587c1151d0cdb9e Parents: 388816a Author: Kenneth Knowles Authored: Fri Jul 22 13:00:10 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 18:25:52 2016 -0700 -- .../beam/examples/DebuggingWordCount.java | 6 +- .../apache/beam/examples/MinimalWordCount.java | 7 +- .../apache/beam/examples/WindowedWordCount.java | 10 +- .../org/apache/beam/examples/WordCount.java | 8 +- .../examples/common/PubsubFileInjector.java | 6 +- .../beam/examples/complete/AutoComplete.java| 16 +- .../examples/complete/StreamingWordExtract.java | 12 +- .../apache/beam/examples/complete/TfIdf.java| 16 +- .../examples/complete/TopWikipediaSessions.java | 12 +- .../examples/complete/TrafficMaxLaneFlow.java | 10 +- .../beam/examples/complete/TrafficRoutes.java | 12 +- .../examples/cookbook/BigQueryTornadoes.java| 6 +- .../cookbook/CombinePerKeyExamples.java | 6 +- .../examples/cookbook/DatastoreWordCount.java | 11 +- .../beam/examples/cookbook/FilterExamples.java | 12 +- .../beam/examples/cookbook/JoinExamples.java| 10 +- .../examples/cookbook/MaxPerKeyExamples.java| 6 +- .../beam/examples/cookbook/TriggerExample.java | 12 +- .../org/apache/beam/examples/WordCountTest.java | 2 +- .../examples/complete/AutoCompleteTest.java | 4 +- .../examples/cookbook/TriggerExampleTest.java | 4 +- .../beam/examples/complete/game/GameStats.java | 10 +- .../beam/examples/complete/game/UserScore.java | 4 +- .../complete/game/utils/WriteToBigQuery.java| 12 +- .../game/utils/WriteWindowedToBigQuery.java | 8 +- .../examples/complete/game/UserScoreTest.java | 2 +- .../core/GroupAlsoByWindowViaWindowSetDoFn.java | 12 +- .../core/UnboundedReadFromBoundedSource.java| 2 +- .../apache/beam/sdk/util/AssignWindowsDoFn.java | 10 +- .../org/apache/beam/sdk/util/DoFnRunner.java| 21 +- .../apache/beam/sdk/util/DoFnRunnerBase.java| 54 +- .../org/apache/beam/sdk/util/DoFnRunners.java | 24 +- .../beam/sdk/util/GroupAlsoByWindowsDoFn.java | 6 +- .../GroupAlsoByWindowsViaOutputBufferDoFn.java | 4 +- .../sdk/util/GroupByKeyViaGroupByKeyOnly.java | 6 +- .../sdk/util/LateDataDroppingDoFnRunner.java| 4 +- .../apache/beam/sdk/util/PaneInfoTracker.java | 1 - .../apache/beam/sdk/util/ReduceFnRunner.java| 4 +- .../apache/beam/sdk/util/SimpleDoFnRunner.java | 12 +- .../org/apache/beam/sdk/util/WatermarkHold.java | 1 - .../beam/sdk/util/ReduceFnRunnerTest.java | 1 + .../apache/beam/sdk/util/ReduceFnTester.java| 1 + .../beam/sdk/util/SimpleDoFnRunnerTest.java | 6 +- .../GroupAlsoByWindowEvaluatorFactory.java | 4 +- .../ImmutabilityCheckingBundleFactory.java | 4 +- .../beam/runners/direct/ParDoEvaluator.java | 4 +- .../direct/ParDoMultiEvaluatorFactory.java | 11 +- .../direct/ParDoSingleEvaluatorFactory.java | 11 +- .../direct/TransformEvaluatorFactory.java | 6 +- .../direct/WriteWithShardingFactory.java| 4 +- .../ConsumerTrackingPipelineVisitorTest.java| 22 +- .../beam/runners/direct/DirectRunnerTest.java | 24 +- .../ImmutabilityCheckingBundleFactoryTest.java | 6 +- .../ImmutabilityEnforcementFactoryTest.java | 6 +- .../direct/KeyedPValueTrackingVisitorTest.java | 6 +- .../beam/runners/direct/ParDoEvaluatorTest.java | 6 +- .../direct/ParDoMultiEvaluatorFactoryTest.java | 10 +- .../direct/ParDoSingleEvaluatorFactoryTest.java | 10 +- .../runners/direct/WatermarkManagerTest.java| 7 +- .../beam/runners/flink/examples/TFIDF.java | 16 +- .../beam/runners/flink/examples/WordCount.java | 4 +- .../flink/examples/streaming/AutoComplete.java | 16 +- .../flink/examples/streaming/JoinExamples.java | 8 +- .../examples/streaming/KafkaIOExamples.java | 4 +- .../KafkaWindowedWordCountExample.java | 6 +- .../examples/streaming/WindowedWordCount.java | 6 +- .../FlinkBatchTransformTranslators.java | 12 +- .../FlinkStreamingTransformTranslators.java | 9 +- .../functions/FlinkDoFnFunction.java| 10 +- .../FlinkMergingNonShuffleReduceFunction.java | 8 +- .../functions/FlinkMultiOutputDoFnFunction.java | 10 +- .../FlinkMultiOutputProcessContext.java | 6 +- .../functions/FlinkNoElementAssignContext.java | 8 +- .../functions/FlinkPartialReduceFunction.java | 8 +- .../functions/FlinkProcessContext.java | 16 +- .../functions/FlinkReduceFunctio
[14/51] [abbrv] incubator-beam git commit: Port MinimalWordCount example from OldDoFn to DoFn
Port MinimalWordCount example from OldDoFn to DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/4ceec0e8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/4ceec0e8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/4ceec0e8
Branch: refs/heads/python-sdk
Commit: 4ceec0e86f1c4e885168957299dbe81c61fbc7e7
Parents: 64481d0
Author: Kenneth Knowles
Authored: Fri Jul 22 14:28:42 2016 -0700
Committer: Dan Halperin
Committed: Wed Aug 3 18:25:53 2016 -0700
--
.../java/org/apache/beam/examples/MinimalWordCount.java | 9 -
1 file changed, 4 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4ceec0e8/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java
--
diff --git
a/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java
b/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java
index ab0bb6d..df725e3 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java
@@ -22,8 +22,8 @@ import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
-import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;
@@ -89,12 +89,11 @@ public class MinimalWordCount {
// the input text (a set of Shakespeare's texts).
p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
// Concept #2: Apply a ParDo transform to our PCollection of text lines.
This ParDo invokes a
- // OldDoFn (defined in-line) on each element that tokenizes the text line
into individua
- // words.
+ // DoFn (defined in-line) on each element that tokenizes the text line
into individual words.
// The ParDo returns a PCollection, where each element is an
individual word in
// Shakespeare's collected texts.
- .apply("ExtractWords", ParDo.of(new OldDoFn() {
- @Override
+ .apply("ExtractWords", ParDo.of(new DoFn() {
+ @ProcessElement
public void processElement(ProcessContext c) {
for (String word : c.element().split("[^a-zA-Z']+")) {
if (!word.isEmpty()) {
[35/51] [abbrv] incubator-beam git commit: Port join library to new DoFn
Port join library to new DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/620bd994
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/620bd994
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/620bd994
Branch: refs/heads/python-sdk
Commit: 620bd9949a6176ddd1903687fe9b8ba8c5822367
Parents: a1c06d7
Author: Kenneth Knowles
Authored: Wed Aug 3 19:55:21 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../apache/beam/sdk/extensions/joinlibrary/Join.java | 14 +++---
1 file changed, 7 insertions(+), 7 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/620bd994/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
--
diff --git
a/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
b/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
index 88836f9..f4e6ccb 100644
---
a/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
+++
b/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java
@@ -20,7 +20,7 @@ package org.apache.beam.sdk.extensions.joinlibrary;
import static com.google.common.base.Preconditions.checkNotNull;
import org.apache.beam.sdk.coders.KvCoder;
-import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.join.CoGbkResult;
import org.apache.beam.sdk.transforms.join.CoGroupByKey;
@@ -59,8 +59,8 @@ public class Join {
.apply(CoGroupByKey.create());
return coGbkResultCollection.apply(ParDo.of(
- new OldDoFn, KV>>() {
-@Override
+ new DoFn, KV>>() {
+@ProcessElement
public void processElement(ProcessContext c) {
KV e = c.element();
@@ -108,8 +108,8 @@ public class Join {
.apply(CoGroupByKey.create());
return coGbkResultCollection.apply(ParDo.of(
- new OldDoFn, KV>>() {
-@Override
+ new DoFn, KV>>() {
+@ProcessElement
public void processElement(ProcessContext c) {
KV e = c.element();
@@ -161,8 +161,8 @@ public class Join {
.apply(CoGroupByKey.create());
return coGbkResultCollection.apply(ParDo.of(
- new OldDoFn, KV>>() {
-@Override
+ new DoFn, KV>>() {
+@ProcessElement
public void processElement(ProcessContext c) {
KV e = c.element();
[05/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SerializationTest.java
--
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SerializationTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SerializationTest.java
index 5b9eeff..5e96c46 100644
---
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SerializationTest.java
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SerializationTest.java
@@ -30,7 +30,7 @@ import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
@@ -145,9 +145,9 @@ public class SerializationTest {
}
/**
- * A DoFn that tokenizes lines of text into individual words.
+ * A OldDoFn that tokenizes lines of text into individual words.
*/
- static class ExtractWordsFn extends DoFn {
+ static class ExtractWordsFn extends OldDoFn {
private static final Pattern WORD_BOUNDARY =
Pattern.compile("[^a-zA-Z']+");
private final Aggregator emptyLines =
createAggregator("emptyLines", new Sum.SumLongFn());
@@ -173,9 +173,9 @@ public class SerializationTest {
}
/**
- * A DoFn that converts a Word and Count into a printable string.
+ * A OldDoFn that converts a Word and Count into a printable string.
*/
- private static class FormatCountsFn extends DoFn,
StringHolder> {
+ private static class FormatCountsFn extends OldDoFn,
StringHolder> {
@Override
public void processElement(ProcessContext c) {
c.output(new StringHolder(c.element().getKey() + ": " +
c.element().getValue()));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SideEffectsTest.java
--
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SideEffectsTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SideEffectsTest.java
index 60b7f71..5775565 100644
---
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SideEffectsTest.java
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/SideEffectsTest.java
@@ -28,7 +28,7 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringDelegateCoder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.junit.After;
@@ -54,7 +54,7 @@ public class SideEffectsTest implements Serializable {
pipeline.getCoderRegistry().registerCoder(URI.class,
StringDelegateCoder.of(URI.class));
-pipeline.apply(Create.of("a")).apply(ParDo.of(new DoFn() {
+pipeline.apply(Create.of("a")).apply(ParDo.of(new OldDoFn() {
@Override
public void processElement(ProcessContext c) throws Exception {
throw new UserException();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/KafkaStreamingTest.java
--
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/KafkaStreamingTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/KafkaStreamingTest.java
index 904b448..c005f14 100644
---
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/KafkaStreamingTest.java
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/KafkaStreamingTest.java
@@ -27,7 +27,7 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
@@ -122,7 +122,7 @@ public class KafkaStreamingTest {
EMBEDDED_ZOOKEEPER.shutdown();
}
- private static class FormatKVFn extends DoFn, String> {
+ private static class FormatKVFn extends OldDoFn, String> {
@Over
[11/51] [abbrv] incubator-beam git commit: Port AutoComplete example from OldDoFn to DoFn
Port AutoComplete example from OldDoFn to DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/3236eec2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/3236eec2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/3236eec2
Branch: refs/heads/python-sdk
Commit: 3236eec22a8902393e6becefb771b9a4768ccc50
Parents: 49d2f17
Author: Kenneth Knowles
Authored: Fri Jul 22 14:29:37 2016 -0700
Committer: Dan Halperin
Committed: Wed Aug 3 18:25:53 2016 -0700
--
.../beam/examples/complete/AutoComplete.java| 30 ++--
1 file changed, 15 insertions(+), 15 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3236eec2/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
--
diff --git
a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
index 7b44af8..1ab39c9 100644
---
a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
+++
b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
@@ -36,9 +36,9 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.Flatten;
-import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Partition;
@@ -130,8 +130,8 @@ public class AutoComplete {
// Map the KV outputs of Count into our own CompletionCandiate class.
.apply("CreateCompletionCandidates", ParDo.of(
-new OldDoFn, CompletionCandidate>() {
- @Override
+new DoFn, CompletionCandidate>() {
+ @ProcessElement
public void processElement(ProcessContext c) {
c.output(new CompletionCandidate(c.element().getKey(),
c.element().getValue()));
}
@@ -209,8 +209,8 @@ public class AutoComplete {
}
private static class FlattenTops
-extends OldDoFn>,
CompletionCandidate> {
- @Override
+extends DoFn>,
CompletionCandidate> {
+ @ProcessElement
public void processElement(ProcessContext c) {
for (CompletionCandidate cc : c.element().getValue()) {
c.output(cc);
@@ -260,10 +260,10 @@ public class AutoComplete {
}
/**
- * A OldDoFn that keys each candidate by all its prefixes.
+ * A DoFn that keys each candidate by all its prefixes.
*/
private static class AllPrefixes
- extends OldDoFn> {
+ extends DoFn> {
private final int minPrefix;
private final int maxPrefix;
public AllPrefixes(int minPrefix) {
@@ -273,8 +273,8 @@ public class AutoComplete {
this.minPrefix = minPrefix;
this.maxPrefix = maxPrefix;
}
-@Override
- public void processElement(ProcessContext c) {
+@ProcessElement
+public void processElement(ProcessContext c) {
String word = c.element().value;
for (int i = minPrefix; i <= Math.min(word.length(), maxPrefix); i++) {
c.output(KV.of(word.substring(0, i), c.element()));
@@ -341,8 +341,8 @@ public class AutoComplete {
/**
* Takes as input a set of strings, and emits each #hashtag found therein.
*/
- static class ExtractHashtags extends OldDoFn {
-@Override
+ static class ExtractHashtags extends DoFn {
+@ProcessElement
public void processElement(ProcessContext c) {
Matcher m = Pattern.compile("#\\S+").matcher(c.element());
while (m.find()) {
@@ -351,8 +351,8 @@ public class AutoComplete {
}
}
- static class FormatForBigquery extends OldDoFn>, TableRow> {
-@Override
+ static class FormatForBigquery extends DoFn>, TableRow> {
+@ProcessElement
public void processElement(ProcessContext c) {
List completions = new ArrayList<>();
for (CompletionCandidate cc : c.element().getValue()) {
@@ -385,14 +385,14 @@ public class AutoComplete {
* Takes as input a the top candidates per prefix, and emits an entity
* suitable for writing to Datastore.
*/
- static class FormatForDatastore extends OldDoFn>, Entity> {
+ static class FormatForDatastore extends DoFn>, Entity> {
private String kind;
public FormatForDatastore(String kind) {
this.kind = kind;
}
-@Override
+@ProcessElement
public void processEle
[13/51] [abbrv] incubator-beam git commit: Port WindowedWordCount example from OldDoFn to DoFn
Port WindowedWordCount example from OldDoFn to DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ca9e3372
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ca9e3372
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ca9e3372
Branch: refs/heads/python-sdk
Commit: ca9e337203208c7c5876f0710fb3a45430a5b3a8
Parents: 4ceec0e
Author: Kenneth Knowles
Authored: Fri Jul 22 14:29:01 2016 -0700
Committer: Dan Halperin
Committed: Wed Aug 3 18:25:53 2016 -0700
--
.../org/apache/beam/examples/WindowedWordCount.java | 14 +++---
1 file changed, 7 insertions(+), 7 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ca9e3372/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
--
diff --git
a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
index 17f7da3..842cb54 100644
---
a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
+++
b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
@@ -27,7 +27,7 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
@@ -103,14 +103,14 @@ public class WindowedWordCount {
static final int WINDOW_SIZE = 1; // Default window duration in minutes
/**
- * Concept #2: A OldDoFn that sets the data element timestamp. This is a
silly method, just for
+ * Concept #2: A DoFn that sets the data element timestamp. This is a silly
method, just for
* this example, for the bounded data case.
*
* Imagine that many ghosts of Shakespeare are all typing madly at the
same time to recreate
* his masterworks. Each line of the corpus will get a random associated
timestamp somewhere in a
* 2-hour period.
*/
- static class AddTimestampFn extends OldDoFn {
+ static class AddTimestampFn extends DoFn {
private static final Duration RAND_RANGE = Duration.standardHours(2);
private final Instant minTimestamp;
@@ -118,7 +118,7 @@ public class WindowedWordCount {
this.minTimestamp = new Instant(System.currentTimeMillis());
}
-@Override
+@ProcessElement
public void processElement(ProcessContext c) {
// Generate a timestamp that falls somewhere in the past two hours.
long randMillis = (long) (Math.random() * RAND_RANGE.getMillis());
@@ -130,9 +130,9 @@ public class WindowedWordCount {
}
}
- /** A OldDoFn that converts a Word and Count into a BigQuery table row. */
- static class FormatAsTableRowFn extends OldDoFn, TableRow> {
-@Override
+ /** A DoFn that converts a Word and Count into a BigQuery table row. */
+ static class FormatAsTableRowFn extends DoFn, TableRow> {
+@ProcessElement
public void processElement(ProcessContext c) {
TableRow row = new TableRow()
.set("word", c.element().getKey())
[01/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk 65152cab8 -> d72ffb080
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java
index cafe873..517f968 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataTest.java
@@ -24,6 +24,7 @@ import static
org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasName
import static
org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasType;
import static
org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasValue;
import static
org.apache.beam.sdk.transforms.display.DisplayDataMatchers.includesDisplayDataFrom;
+
import static org.hamcrest.Matchers.allOf;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.everyItem;
@@ -40,7 +41,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.display.DisplayData.Builder;
@@ -54,7 +55,6 @@ import com.google.common.testing.EqualsTester;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
-
import org.hamcrest.CustomTypeSafeMatcher;
import org.hamcrest.FeatureMatcher;
import org.hamcrest.Matcher;
@@ -1053,7 +1053,7 @@ public class DisplayDataTest implements Serializable {
private static class IdentityTransform extends PTransform,
PCollection> {
@Override
public PCollection apply(PCollection input) {
- return input.apply(ParDo.of(new DoFn() {
+ return input.apply(ParDo.of(new OldDoFn() {
@Override
public void processElement(ProcessContext c) throws Exception {
c.output(c.element());
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java
index 10a2a7e..97667a3 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java
@@ -29,9 +29,9 @@ import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.RunnableOnService;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.DoFn.RequiresWindowAccess;
import org.apache.beam.sdk.transforms.DoFnTester;
+import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.OldDoFn.RequiresWindowAccess;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -85,8 +85,8 @@ public class CoGroupByKeyTest implements Serializable {
.withCoder(KvCoder.of(BigEndianIntegerCoder.of(),
StringUtf8Coder.of(;
}
return input
-.apply("Identity" + name, ParDo.of(new DoFn,
- KV>() {
+.apply("Identity" + name, ParDo.of(new OldDoFn,
+ KV>() {
@Override
public void processElement(ProcessContext c) {
c.output(c.element());
@@ -313,11 +313,11 @@ public class CoGroupByKeyTest implements Serializable {
}
/**
- * A DoFn used in testCoGroupByKeyWithWindowing(), to test processing the
+ * A OldDoFn used in testCoGroupByKeyWithWindowing(), to test processing the
* results of a CoGroupByKey.
*/
private static class ClickOfPurchaseFn extends
- DoFn, KV> implements
RequiresWindowAccess {
+ OldDoFn, KV> implements
RequiresWindowAccess {
private final TupleTag clicksTag;
private final TupleTag purchasesTag;
@@ -347,11 +347,11 @@ public class CoGroupByKeyTest implements Serializable {
/**
- * A DoFn used in testCoGroupByKeyHandleResults(), to test processing the
+ * A OldDoFn used in testCoGroupByKeyHandleResults(), to test processing the
* r
[32/51] [abbrv] incubator-beam git commit: Port examples to new DoFn
Port examples to new DoFn
Port example tests to new DoFn
Port TfIdf example to new DoFn
Port TopWikipediaSessions example to new DoFn
Port GameState Java 8 example to new DoFn
Port the UserScore example to new DoFn
Port StreamingWordExtract example to new DoFn
Port TrafficMaxLaneFlow to new DoFn
Port TrafficeRoutes example to new DoFn
Port DatastoreWordCount example to new DoFn
Port BigQueryTornadoes example to new DoFn
Port MaxPerKeyExamples to new DoFn
Port CombinePerKeyExamples to new DoFn
Port TriggerExample to new DoFn
Port JoinExamples to new DoFn
Port FilterExamples to new DoFn
Fix mention of DoFn in WordCountTest
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/71e027dc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/71e027dc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/71e027dc
Branch: refs/heads/python-sdk
Commit: 71e027dc1ff7d5de0eea82278427546c07e26e8f
Parents: 734bfb9
Author: Kenneth Knowles
Authored: Wed Aug 3 18:54:22 2016 -0700
Committer: Dan Halperin
Committed: Thu Aug 4 11:54:29 2016 -0700
--
.../examples/complete/StreamingWordExtract.java | 18 ++--
.../apache/beam/examples/complete/TfIdf.java| 28 +-
.../examples/complete/TopWikipediaSessions.java | 31 +---
.../examples/complete/TrafficMaxLaneFlow.java | 16 +-
.../beam/examples/complete/TrafficRoutes.java | 20 ++---
.../examples/cookbook/BigQueryTornadoes.java| 10 +++
.../cookbook/CombinePerKeyExamples.java | 10 +++
.../examples/cookbook/DatastoreWordCount.java | 14 -
.../beam/examples/cookbook/FilterExamples.java | 20 ++---
.../beam/examples/cookbook/JoinExamples.java| 18 ++--
.../examples/cookbook/MaxPerKeyExamples.java| 10 +++
.../beam/examples/cookbook/TriggerExample.java | 25
.../org/apache/beam/examples/WordCountTest.java | 3 +-
.../examples/complete/AutoCompleteTest.java | 6 ++--
.../examples/cookbook/TriggerExampleTest.java | 6 ++--
.../beam/examples/complete/game/GameStats.java | 22 +++---
.../beam/examples/complete/game/UserScore.java | 6 ++--
.../examples/complete/game/UserScoreTest.java | 2 +-
18 files changed, 130 insertions(+), 135 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/71e027dc/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
--
diff --git
a/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
b/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
index b0c9ffd..3f30f21 100644
---
a/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
+++
b/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
@@ -28,7 +28,7 @@ import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
-import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import com.google.api.services.bigquery.model.TableFieldSchema;
@@ -55,9 +55,9 @@ import java.util.ArrayList;
*/
public class StreamingWordExtract {
- /** A OldDoFn that tokenizes lines of text into individual words. */
- static class ExtractWords extends OldDoFn {
-@Override
+ /** A {@link DoFn} that tokenizes lines of text into individual words. */
+ static class ExtractWords extends DoFn {
+@ProcessElement
public void processElement(ProcessContext c) {
String[] words = c.element().split("[^a-zA-Z']+");
for (String word : words) {
@@ -68,9 +68,9 @@ public class StreamingWordExtract {
}
}
- /** A OldDoFn that uppercases a word. */
- static class Uppercase extends OldDoFn {
-@Override
+ /** A {@link DoFn} that uppercases a word. */
+ static class Uppercase extends DoFn {
+@ProcessElement
public void processElement(ProcessContext c) {
c.output(c.element().toUpperCase());
}
@@ -79,11 +79,11 @@ public class StreamingWordExtract {
/**
* Converts strings into BigQuery rows.
*/
- static class StringToRowConverter extends OldDoFn {
+ static class StringToRowConverter extends DoFn {
/**
* In this example, put the whole string into single BigQuery field.
*/
-@Override
+@ProcessElement
public void processElement(ProcessContext c) {
c.output(new TableRow().set("string_field", c.element()));
}
http://git-wip-us.apache
[43/51] [abbrv] incubator-beam git commit: This closes #782
This closes #782 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8daf518b Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8daf518b Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8daf518b Branch: refs/heads/python-sdk Commit: 8daf518bccfe425082c7d0b3f31f3623ff67e000 Parents: fcf6b1d 47341e1 Author: Kenneth Knowles Authored: Thu Aug 4 20:10:55 2016 -0700 Committer: Kenneth Knowles Committed: Thu Aug 4 20:10:55 2016 -0700 -- .../beam/runners/dataflow/DataflowRunner.java | 3 +- .../java/org/apache/beam/sdk/io/PubsubIO.java | 14 +++ .../apache/beam/sdk/io/PubsubUnboundedSink.java | 17 - .../beam/sdk/io/PubsubUnboundedSource.java | 7 ++-- .../beam/sdk/options/PipelineOptions.java | 9 ++--- .../org/apache/beam/sdk/testing/PAssert.java| 39 ++-- .../org/apache/beam/sdk/transforms/Count.java | 4 +- .../org/apache/beam/sdk/transforms/Create.java | 4 +- .../beam/sdk/transforms/DoFnReflector.java | 6 +++ .../beam/sdk/transforms/FlatMapElements.java| 4 +- .../org/apache/beam/sdk/transforms/Flatten.java | 4 +- .../org/apache/beam/sdk/transforms/Keys.java| 4 +- .../org/apache/beam/sdk/transforms/KvSwap.java | 4 +- .../apache/beam/sdk/transforms/MapElements.java | 4 +- .../org/apache/beam/sdk/transforms/ParDo.java | 2 +- .../apache/beam/sdk/transforms/Partition.java | 4 +- .../beam/sdk/transforms/RemoveDuplicates.java | 4 +- .../org/apache/beam/sdk/transforms/Sample.java | 6 +-- .../org/apache/beam/sdk/transforms/Values.java | 4 +- .../org/apache/beam/sdk/transforms/View.java| 8 ++-- .../apache/beam/sdk/transforms/WithKeys.java| 4 +- .../beam/sdk/transforms/WithTimestamps.java | 6 +-- .../beam/sdk/transforms/join/CoGroupByKey.java | 16 .../java/org/apache/beam/sdk/PipelineTest.java | 8 ++-- .../apache/beam/sdk/coders/AvroCoderTest.java | 6 +-- .../beam/sdk/coders/CoderRegistryTest.java | 10 ++--- .../beam/sdk/coders/SerializableCoderTest.java | 10 ++--- .../apache/beam/sdk/io/CountingInputTest.java | 6 +-- .../apache/beam/sdk/io/CountingSourceTest.java | 6 +-- .../beam/sdk/io/PubsubUnboundedSinkTest.java| 6 +-- .../sdk/transforms/ApproximateUniqueTest.java | 6 +-- .../beam/sdk/transforms/CombineFnsTest.java | 4 +- .../apache/beam/sdk/transforms/CombineTest.java | 18 - .../apache/beam/sdk/transforms/CreateTest.java | 4 +- .../apache/beam/sdk/transforms/FlattenTest.java | 8 ++-- .../beam/sdk/transforms/GroupByKeyTest.java | 8 ++-- .../beam/sdk/transforms/WithTimestampsTest.java | 12 +++--- .../display/DisplayDataEvaluatorTest.java | 10 ++--- .../sdk/transforms/display/DisplayDataTest.java | 6 +-- .../sdk/transforms/join/CoGroupByKeyTest.java | 34 - .../sdk/transforms/windowing/WindowTest.java| 10 ++--- .../sdk/transforms/windowing/WindowingTest.java | 23 ++-- .../beam/sdk/values/PCollectionTupleTest.java | 6 +-- .../apache/beam/sdk/values/TypedPValueTest.java | 10 ++--- .../beam/sdk/extensions/joinlibrary/Join.java | 14 +++ .../beam/sdk/io/gcp/bigquery/BigQueryIO.java| 36 +- .../beam/sdk/io/gcp/bigtable/BigtableIO.java| 12 +++--- .../beam/sdk/io/gcp/datastore/V1Beta3.java | 18 - .../sdk/io/gcp/bigquery/BigQueryIOTest.java | 10 ++--- .../sdk/io/gcp/bigtable/BigtableWriteIT.java| 6 +-- .../sdk/io/gcp/datastore/V1Beta3TestUtil.java | 9 ++--- .../java/org/apache/beam/sdk/io/jms/JmsIO.java | 10 ++--- .../org/apache/beam/sdk/io/kafka/KafkaIO.java | 19 +- .../apache/beam/sdk/io/kafka/KafkaIOTest.java | 10 ++--- 54 files changed, 265 insertions(+), 267 deletions(-) --
[16/51] [abbrv] incubator-beam git commit: Run findbugs in the test-compile phase
Run findbugs in the test-compile phase Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/14c6d99e Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/14c6d99e Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/14c6d99e Branch: refs/heads/python-sdk Commit: 14c6d99e087b2e1606422821341136a5d5e8ec23 Parents: 9a329aa Author: Kenneth Knowles Authored: Wed Aug 3 21:31:17 2016 -0700 Committer: Kenneth Knowles Committed: Wed Aug 3 21:31:17 2016 -0700 -- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/14c6d99e/pom.xml -- diff --git a/pom.xml b/pom.xml index 23b7e4d..9e58ffe 100644 --- a/pom.xml +++ b/pom.xml @@ -1029,7 +1029,7 @@ - test + test-compile check
[26/51] [abbrv] incubator-beam git commit: Closes #784
Closes #784 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/be2758cb Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/be2758cb Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/be2758cb Branch: refs/heads/python-sdk Commit: be2758cbf0014dccc34641fd791232ad061549d3 Parents: b8d7559 fb6d2c2 Author: Dan Halperin Authored: Thu Aug 4 11:47:31 2016 -0700 Committer: Dan Halperin Committed: Thu Aug 4 11:47:31 2016 -0700 -- .../core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java| 2 -- .../core/src/main/java/org/apache/beam/sdk/transforms/OldDoFn.java | 2 ++ 2 files changed, 2 insertions(+), 2 deletions(-) --
[39/51] [abbrv] incubator-beam git commit: Correctly determine if DoFn has an anonymous class in ParDo
Correctly determine if DoFn has an anonymous class in ParDo
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/0b186529
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/0b186529
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/0b186529
Branch: refs/heads/python-sdk
Commit: 0b1865295cb89d88878d0a021df103ed45240924
Parents: fcf6b1d
Author: Kenneth Knowles
Authored: Thu Aug 4 14:54:56 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/0b186529/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
index bb1af9c..91f6203 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
@@ -958,7 +958,7 @@ public class ParDo {
@Override
protected String getKindString() {
Class clazz = DoFnReflector.getDoFnClass(fn);
- if (fn.getClass().isAnonymousClass()) {
+ if (clazz.isAnonymousClass()) {
return "AnonymousParMultiDo";
} else {
return String.format("ParMultiDo(%s)",
StringUtils.approximateSimpleName(clazz));
[45/51] [abbrv] incubator-beam git commit: Remove References to Instant#now in the DirectRunner
Remove References to Instant#now in the DirectRunner
The DirectRunner should use exclusively the configured clock to
determine the processing time.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7585cfc3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7585cfc3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7585cfc3
Branch: refs/heads/python-sdk
Commit: 7585cfc3693800b00c4ccc799c27f0311e9b0cc1
Parents: fcf6b1d
Author: Thomas Groh
Authored: Fri Aug 5 09:58:05 2016 -0700
Committer: Thomas Groh
Committed: Fri Aug 5 10:04:21 2016 -0700
--
.../apache/beam/runners/direct/EvaluationContext.java | 14 ++
.../direct/ExecutorServiceParallelExecutor.java | 5 ++---
2 files changed, 12 insertions(+), 7 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7585cfc3/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
--
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
index 23c139d..94f28e2 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
@@ -48,6 +48,8 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.MoreExecutors;
+import org.joda.time.Instant;
+
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
@@ -81,6 +83,7 @@ class EvaluationContext {
/** The options that were used to create this {@link Pipeline}. */
private final DirectOptions options;
+ private final Clock clock;
private final BundleFactory bundleFactory;
/** The current processing time and event time watermarks and timers. */
@@ -116,6 +119,7 @@ class EvaluationContext {
Map, String> stepNames,
Collection> views) {
this.options = checkNotNull(options);
+this.clock = options.getClock();
this.bundleFactory = checkNotNull(bundleFactory);
checkNotNull(rootTransforms);
checkNotNull(valueToConsumers);
@@ -123,9 +127,7 @@ class EvaluationContext {
checkNotNull(views);
this.stepNames = stepNames;
-this.watermarkManager =
-WatermarkManager.create(
-NanosOffsetClock.create(), rootTransforms, valueToConsumers);
+this.watermarkManager = WatermarkManager.create(clock, rootTransforms,
valueToConsumers);
this.sideInputContainer = SideInputContainer.create(this, views);
this.applicationStateInternals = new ConcurrentHashMap<>();
@@ -314,7 +316,7 @@ class EvaluationContext {
AppliedPTransform application, StructuralKey key) {
StepAndKey stepAndKey = StepAndKey.of(application, key);
return new DirectExecutionContext(
-options.getClock(),
+clock,
key,
(CopyOnAccessInMemoryStateInternals)
applicationStateInternals.get(stepAndKey),
watermarkManager.getWatermarks(application));
@@ -427,4 +429,8 @@ class EvaluationContext {
}
return true;
}
+
+ public Instant now() {
+return clock.now();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7585cfc3/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
--
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
index 64836d8..a0a5ec0 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
@@ -40,7 +40,6 @@ import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
-import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -433,9 +432,9 @@ final class ExecutorServiceParallelExecutor implements
PipelineExecutor {
.createKeyedBundle(
null, keyTimers.getKey(), (PCollection)
transform.getInput())
.add(WindowedValue.valueInEmptyWindows(work))
- .commit(Instant.now());
- state.set(ExecutorState.ACTIVE);
+ .commit(evaluationContext.now());
[GitHub] incubator-beam pull request #787: merge master into python-sdk
Github user asfgit closed the pull request at: https://github.com/apache/incubator-beam/pull/787 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[51/51] [abbrv] incubator-beam git commit: This closes #787
This closes #787 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d72ffb08 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d72ffb08 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d72ffb08 Branch: refs/heads/python-sdk Commit: d72ffb0804bfc6f4b4a5e21ed01c6c42424dae7e Parents: 65152ca c398811 Author: Kenneth Knowles Authored: Fri Aug 5 19:52:24 2016 -0700 Committer: Kenneth Knowles Committed: Fri Aug 5 19:52:24 2016 -0700 -- .travis.yml |6 +- KEYS| 141 - NOTICE |1 - README.md | 32 +- examples/java/pom.xml | 73 +- .../beam/examples/DebuggingWordCount.java |4 +- .../apache/beam/examples/MinimalWordCount.java | 30 +- .../apache/beam/examples/WindowedWordCount.java | 91 +- .../org/apache/beam/examples/WordCount.java |6 +- .../examples/common/DataflowExampleOptions.java | 37 - .../examples/common/DataflowExampleUtils.java | 491 -- .../common/ExampleBigQueryTableOptions.java | 10 +- .../beam/examples/common/ExampleOptions.java| 83 + ...xamplePubsubTopicAndSubscriptionOptions.java | 10 +- .../common/ExamplePubsubTopicOptions.java | 12 +- .../beam/examples/common/ExampleUtils.java | 390 ++ .../examples/common/PubsubFileInjector.java | 10 +- .../beam/examples/complete/AutoComplete.java| 91 +- .../examples/complete/StreamingWordExtract.java | 56 +- .../apache/beam/examples/complete/TfIdf.java| 28 +- .../examples/complete/TopWikipediaSessions.java | 38 +- .../examples/complete/TrafficMaxLaneFlow.java | 90 +- .../beam/examples/complete/TrafficRoutes.java | 90 +- .../examples/cookbook/BigQueryTornadoes.java|6 +- .../cookbook/CombinePerKeyExamples.java |6 +- .../examples/cookbook/DatastoreWordCount.java | 67 +- .../beam/examples/cookbook/DeDupExample.java|5 +- .../beam/examples/cookbook/FilterExamples.java | 17 +- .../beam/examples/cookbook/JoinExamples.java| 14 +- .../examples/cookbook/MaxPerKeyExamples.java|6 +- .../beam/examples/cookbook/TriggerExample.java | 97 +- .../org/apache/beam/examples/WordCountIT.java | 118 +- .../org/apache/beam/examples/WordCountTest.java |3 +- .../examples/complete/AutoCompleteTest.java |2 +- .../examples/cookbook/TriggerExampleTest.java |2 +- examples/java8/pom.xml | 81 +- .../beam/examples/MinimalWordCountJava8.java| 30 +- .../beam/examples/complete/game/GameStats.java | 57 +- .../examples/complete/game/HourlyTeamScore.java |9 +- .../examples/complete/game/LeaderBoard.java | 23 +- .../beam/examples/complete/game/UserScore.java |4 +- .../complete/game/injector/InjectorUtils.java |6 +- .../injector/RetryHttpInitializerWrapper.java |5 +- .../complete/game/utils/WriteToBigQuery.java| 20 +- .../game/utils/WriteWindowedToBigQuery.java | 16 +- .../complete/game/HourlyTeamScoreTest.java |2 +- .../examples/complete/game/UserScoreTest.java |2 +- examples/pom.xml|4 +- pom.xml | 265 +- runners/core-java/pom.xml | 55 +- .../core/GroupAlsoByWindowViaWindowSetDoFn.java | 28 +- .../core/UnboundedReadFromBoundedSource.java| 542 ++ .../apache/beam/runners/core/package-info.java | 22 + .../org/apache/beam/sdk/util/AssignWindows.java | 46 + .../apache/beam/sdk/util/AssignWindowsDoFn.java | 80 + .../beam/sdk/util/BatchTimerInternals.java | 141 + .../org/apache/beam/sdk/util/DoFnRunner.java| 63 + .../apache/beam/sdk/util/DoFnRunnerBase.java| 551 ++ .../org/apache/beam/sdk/util/DoFnRunners.java | 222 + .../beam/sdk/util/GroupAlsoByWindowsDoFn.java | 63 + .../GroupAlsoByWindowsViaOutputBufferDoFn.java | 100 + .../sdk/util/GroupByKeyViaGroupByKeyOnly.java | 269 + .../sdk/util/LateDataDroppingDoFnRunner.java| 147 + .../org/apache/beam/sdk/util/NonEmptyPanes.java | 150 + .../apache/beam/sdk/util/PaneInfoTracker.java | 158 + .../sdk/util/PushbackSideInputDoFnRunner.java | 115 + .../java/org/apache/beam/sdk/util/ReduceFn.java | 130 + .../beam/sdk/util/ReduceFnContextFactory.java | 497 ++ .../apache/beam/sdk/util/ReduceFnRunner.java| 988 .../apache/beam/sdk/util/SimpleDoFnRunner.java | 55 + .../apache/beam/sdk/util/SystemReduceFn.java| 139 + .../org/apache/beam/sdk/util/TriggerRunner.java | 245 + .../org/apache/beam/sdk/util/WatermarkHold.java | 540 ++ .../util/common/ElementByteSizeObservable.java | 42 + .../beam/sdk/util/common/Pee
[15/51] [abbrv] incubator-beam git commit: Port microbenchmarks to new vocabulary
Port microbenchmarks to new vocabulary
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e07c3397
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e07c3397
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e07c3397
Branch: refs/heads/python-sdk
Commit: e07c3397d268f50cc879362227a6887cc52f4a3b
Parents: 3236eec
Author: Kenneth Knowles
Authored: Fri Jul 22 14:29:51 2016 -0700
Committer: Dan Halperin
Committed: Wed Aug 3 18:25:53 2016 -0700
--
.../transforms/DoFnReflectorBenchmark.java | 27 ++--
1 file changed, 14 insertions(+), 13 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e07c3397/sdks/java/microbenchmarks/src/main/java/org/apache/beam/sdk/microbenchmarks/transforms/DoFnReflectorBenchmark.java
--
diff --git
a/sdks/java/microbenchmarks/src/main/java/org/apache/beam/sdk/microbenchmarks/transforms/DoFnReflectorBenchmark.java
b/sdks/java/microbenchmarks/src/main/java/org/apache/beam/sdk/microbenchmarks/transforms/DoFnReflectorBenchmark.java
index fd75e95..233b8be 100644
---
a/sdks/java/microbenchmarks/src/main/java/org/apache/beam/sdk/microbenchmarks/transforms/DoFnReflectorBenchmark.java
+++
b/sdks/java/microbenchmarks/src/main/java/org/apache/beam/sdk/microbenchmarks/transforms/DoFnReflectorBenchmark.java
@@ -53,9 +53,10 @@ public class DoFnReflectorBenchmark {
private OldDoFn oldDoFn = new UpperCaseOldDoFn();
private DoFn doFn = new UpperCaseDoFn();
- private StubDoFnProcessContext stubDoFnContext = new
StubDoFnProcessContext(oldDoFn, ELEMENT);
- private StubDoFnWithContextProcessContext stubDoFnWithContextContext =
- new StubDoFnWithContextProcessContext(doFn, ELEMENT);
+ private StubOldDoFnProcessContext stubOldDoFnContext = new
StubOldDoFnProcessContext(oldDoFn,
+ ELEMENT);
+ private StubDoFnProcessContext stubDoFnContext =
+ new StubDoFnProcessContext(doFn, ELEMENT);
private ExtraContextFactory extraContextFactory =
new ExtraContextFactory() {
@@ -83,21 +84,21 @@ public class DoFnReflectorBenchmark {
}
@Benchmark
- public String invokeDoFn() throws Exception {
-oldDoFn.processElement(stubDoFnContext);
+ public String invokeOldDoFn() throws Exception {
+oldDoFn.processElement(stubOldDoFnContext);
return stubDoFnContext.output;
}
@Benchmark
public String invokeDoFnWithContextViaAdaptor() throws Exception {
-adaptedDoFnWithContext.processElement(stubDoFnContext);
-return stubDoFnContext.output;
+adaptedDoFnWithContext.processElement(stubOldDoFnContext);
+return stubOldDoFnContext.output;
}
@Benchmark
public String invokeDoFnWithContext() throws Exception {
-invoker.invokeProcessElement(stubDoFnWithContextContext,
extraContextFactory);
-return stubDoFnWithContextContext.output;
+invoker.invokeProcessElement(stubDoFnContext, extraContextFactory);
+return stubDoFnContext.output;
}
private static class UpperCaseOldDoFn extends OldDoFn {
@@ -116,12 +117,12 @@ public class DoFnReflectorBenchmark {
}
}
- private static class StubDoFnProcessContext extends OldDoFn.ProcessContext {
+ private static class StubOldDoFnProcessContext extends OldDoFn.ProcessContext {
private final String element;
private String output;
-public StubDoFnProcessContext(OldDoFn fn, String element) {
+public StubOldDoFnProcessContext(OldDoFn fn, String
element) {
fn.super();
this.element = element;
}
@@ -186,12 +187,12 @@ public class DoFnReflectorBenchmark {
}
}
- private static class StubDoFnWithContextProcessContext
+ private static class StubDoFnProcessContext
extends DoFn.ProcessContext {
private final String element;
private String output;
-public StubDoFnWithContextProcessContext(DoFn fn, String
element) {
+public StubDoFnProcessContext(DoFn fn, String element) {
fn.super();
this.element = element;
}
[38/51] [abbrv] incubator-beam git commit: Port PAssert to new DoFn
Port PAssert to new DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ef5e31f8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ef5e31f8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ef5e31f8
Branch: refs/heads/python-sdk
Commit: ef5e31f8b79dcedf8feb4bba0e313bfcf330ab1e
Parents: 1959ddb
Author: Kenneth Knowles
Authored: Wed Aug 3 20:15:58 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../org/apache/beam/sdk/testing/PAssert.java| 39 ++--
1 file changed, 19 insertions(+), 20 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ef5e31f8/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
index 80340c2..e07ee3d 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
@@ -33,11 +33,10 @@ import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.runners.PipelineRunner;
import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.MapElements;
-import org.apache.beam.sdk.transforms.OldDoFn;
-import org.apache.beam.sdk.transforms.OldDoFn.RequiresWindowAccess;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
@@ -762,8 +761,8 @@ public class PAssert {
.apply("RewindowActuals", rewindowActuals.windowActuals())
.apply(
ParDo.of(
- new OldDoFn() {
-@Override
+ new DoFn() {
+@ProcessElement
public void processElement(ProcessContext context) throws
CoderException {
context.output(CoderUtils.clone(coder,
context.element()));
}
@@ -884,8 +883,8 @@ public class PAssert {
}
}
- private static final class ConcatFn extends
OldDoFn>, Iterable> {
-@Override
+ private static final class ConcatFn extends DoFn>,
Iterable> {
+@ProcessElement
public void processElement(ProcessContext c) throws Exception {
c.output(Iterables.concat(c.element()));
}
@@ -995,13 +994,13 @@ public class PAssert {
}
/**
- * A {@link OldDoFn} that runs a checking {@link SerializableFunction} on
the contents of a
+ * A {@link DoFn} that runs a checking {@link SerializableFunction} on the
contents of a
* {@link PCollectionView}, and adjusts counters and thrown exceptions for
use in testing.
*
* The input is ignored, but is {@link Integer} to be usable on runners
that do not support
* null values.
*/
- private static class SideInputCheckerDoFn extends OldDoFn {
+ private static class SideInputCheckerDoFn extends DoFn {
private final SerializableFunction checkerFn;
private final Aggregator success =
createAggregator(SUCCESS_COUNTER, new Sum.SumIntegerFn());
@@ -1015,7 +1014,7 @@ public class PAssert {
this.actual = actual;
}
-@Override
+@ProcessElement
public void processElement(ProcessContext c) {
try {
ActualT actualContents = c.sideInput(actual);
@@ -1030,13 +1029,13 @@ public class PAssert {
}
/**
- * A {@link OldDoFn} that runs a checking {@link SerializableFunction} on
the contents of
+ * A {@link DoFn} that runs a checking {@link SerializableFunction} on the
contents of
* the single iterable element of the input {@link PCollection} and adjusts
counters and
* thrown exceptions for use in testing.
*
* The singleton property is presumed, not enforced.
*/
- private static class GroupedValuesCheckerDoFn extends
OldDoFn {
+ private static class GroupedValuesCheckerDoFn extends DoFn {
private final SerializableFunction checkerFn;
private final Aggregator success =
createAggregator(SUCCESS_COUNTER, new Sum.SumIntegerFn());
@@ -1047,7 +1046,7 @@ public class PAssert {
this.checkerFn = checkerFn;
}
-@Override
+@ProcessElement
public void processElement(ProcessContext c) {
try {
doChecks(c.element(), checkerFn, success, failure);
@@ -1061,14 +1060,14 @@ public class PAssert {
}
/**
- * A {@link OldDoFn} that runs a c
[06/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
--
diff --git
a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
index 89243a3..a4af1b0 100644
---
a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
+++
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
@@ -19,7 +19,7 @@ package org.apache.beam.runners.flink.translation.functions;
import
org.apache.beam.runners.flink.translation.utils.SerializedPipelineOptions;
import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.values.PCollectionView;
@@ -30,13 +30,13 @@ import org.apache.flink.util.Collector;
import java.util.Map;
/**
- * Encapsulates a {@link org.apache.beam.sdk.transforms.DoFn}
+ * Encapsulates a {@link OldDoFn}
* inside a Flink {@link
org.apache.flink.api.common.functions.RichMapPartitionFunction}.
*/
public class FlinkDoFnFunction
extends RichMapPartitionFunction,
WindowedValue> {
- private final DoFn doFn;
+ private final OldDoFn doFn;
private final SerializedPipelineOptions serializedOptions;
private final Map, WindowingStrategy> sideInputs;
@@ -47,7 +47,7 @@ public class FlinkDoFnFunction
private final WindowingStrategy windowingStrategy;
public FlinkDoFnFunction(
- DoFn doFn,
+ OldDoFn doFn,
WindowingStrategy windowingStrategy,
Map, WindowingStrategy> sideInputs,
PipelineOptions options) {
@@ -56,7 +56,7 @@ public class FlinkDoFnFunction
this.serializedOptions = new SerializedPipelineOptions(options);
this.windowingStrategy = windowingStrategy;
-this.requiresWindowAccess = doFn instanceof DoFn.RequiresWindowAccess;
+this.requiresWindowAccess = doFn instanceof OldDoFn.RequiresWindowAccess;
this.hasSideInputs = !sideInputs.isEmpty();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMergingNonShuffleReduceFunction.java
--
diff --git
a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMergingNonShuffleReduceFunction.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMergingNonShuffleReduceFunction.java
index 9074d72..2d36043 100644
---
a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMergingNonShuffleReduceFunction.java
+++
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMergingNonShuffleReduceFunction.java
@@ -20,7 +20,7 @@ package org.apache.beam.runners.flink.translation.functions;
import
org.apache.beam.runners.flink.translation.utils.SerializedPipelineOptions;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.CombineFnBase;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.OutputTimeFn;
@@ -60,7 +60,7 @@ public class FlinkMergingNonShuffleReduceFunction<
private final CombineFnBase.PerKeyCombineFn
combineFn;
- private final DoFn, KV> doFn;
+ private final OldDoFn, KV> doFn;
private final WindowingStrategy windowingStrategy;
@@ -81,8 +81,8 @@ public class FlinkMergingNonShuffleReduceFunction<
this.serializedOptions = new SerializedPipelineOptions(pipelineOptions);
-// dummy DoFn because we need one for ProcessContext
-this.doFn = new DoFn, KV>() {
+// dummy OldDoFn because we need one for ProcessContext
+this.doFn = new OldDoFn, KV>() {
@Override
public void processElement(ProcessContext c) throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMultiOutputDoFnFunction.java
--
diff --git
a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkMultiOutputDoFnFunction.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink
[03/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java
index 77c857c..7917aec 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java
@@ -23,8 +23,8 @@ import static
com.google.common.base.Preconditions.checkNotNull;
import org.apache.beam.sdk.coders.AtomicCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
-import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.util.VarInt;
import com.google.common.base.MoreObjects;
@@ -38,8 +38,8 @@ import java.util.Objects;
/**
* Provides information about the pane an element belongs to. Every pane is
implicitly associated
* with a window. Panes are observable only via the
- * {@link org.apache.beam.sdk.transforms.DoFn.ProcessContext#pane} method of
the context
- * passed to a {@link DoFn#processElement} overridden method.
+ * {@link OldDoFn.ProcessContext#pane} method of the context
+ * passed to a {@link OldDoFn#processElement} overridden method.
*
* Note: This does not uniquely identify a pane, and should not be used for
comparisons.
*/
@@ -74,8 +74,8 @@ public final class PaneInfo {
* definitions:
*
* We'll call a pipeline 'simple' if it does not use
- * {@link org.apache.beam.sdk.transforms.DoFn.Context#outputWithTimestamp} in
- * any {@code DoFn}, and it uses the same
+ * {@link OldDoFn.Context#outputWithTimestamp} in
+ * any {@code OldDoFn}, and it uses the same
* {@link
org.apache.beam.sdk.transforms.windowing.Window.Bound#withAllowedLateness}
* argument value on all windows (or uses the default of {@link
org.joda.time.Duration#ZERO}).
* We'll call an element 'locally late', from the point of view of a
computation on a
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
index fe8b66f..03ff481 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
@@ -21,8 +21,8 @@ import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.Coder.NonDeterministicException;
-import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.display.DisplayData;
@@ -645,7 +645,7 @@ public class Window {
// We first apply a (trivial) transform to the input PCollection to
produce a new
// PCollection. This ensures that we don't modify the windowing
strategy of the input
// which may be used elsewhere.
- .apply("Identity", ParDo.of(new DoFn() {
+ .apply("Identity", ParDo.of(new OldDoFn() {
@Override public void processElement(ProcessContext c) {
c.output(c.element());
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BaseExecutionContext.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BaseExecutionContext.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BaseExecutionContext.java
index a62444f..dd36367 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BaseExecutionContext.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BaseExecutionContext.java
@@ -107,7 +107,7 @@ public abstract class BaseExecutionContexthttp://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BucketingFunction.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BucketingFunction.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/BucketingFuncti
[04/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnWithContext.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnWithContext.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnWithContext.java
index 8a83e44..b27163a 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnWithContext.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnWithContext.java
@@ -24,7 +24,7 @@ import static com.google.common.base.Preconditions.checkState;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
-import org.apache.beam.sdk.transforms.DoFn.DelegatingAggregator;
+import org.apache.beam.sdk.transforms.OldDoFn.DelegatingAggregator;
import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.transforms.display.HasDisplayData;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -127,7 +127,7 @@ public abstract class DoFnWithContext
implements Serializable,
*
* If invoked from {@link ProcessElement}), the timestamp
* must not be older than the input element's timestamp minus
- * {@link DoFn#getAllowedTimestampSkew}. The output element will
+ * {@link OldDoFn#getAllowedTimestampSkew}. The output element will
* be in the same windows as the input element.
*
* If invoked from {@link StartBundle} or {@link FinishBundle},
@@ -176,7 +176,7 @@ public abstract class DoFnWithContext
implements Serializable,
*
* If invoked from {@link ProcessElement}), the timestamp
* must not be older than the input element's timestamp minus
- * {@link DoFn#getAllowedTimestampSkew}. The output element will
+ * {@link OldDoFn#getAllowedTimestampSkew}. The output element will
* be in the same windows as the input element.
*
* If invoked from {@link StartBundle} or {@link FinishBundle},
@@ -194,7 +194,7 @@ public abstract class DoFnWithContext
implements Serializable,
}
/**
- * Information accessible when running {@link DoFn#processElement}.
+ * Information accessible when running {@link OldDoFn#processElement}.
*/
public abstract class ProcessContext extends Context {
@@ -358,13 +358,13 @@ public abstract class DoFnWithContext
implements Serializable,
/**
* Returns an {@link Aggregator} with aggregation logic specified by the
* {@link CombineFn} argument. The name provided must be unique across
- * {@link Aggregator}s created within the DoFn. Aggregators can only be
created
+ * {@link Aggregator}s created within the OldDoFn. Aggregators can only be
created
* during pipeline construction.
*
* @param name the name of the aggregator
* @param combiner the {@link CombineFn} to use in the aggregator
* @return an aggregator for the provided name and combiner in the scope of
- * this DoFn
+ * this OldDoFn
* @throws NullPointerException if the name or combiner is null
* @throws IllegalArgumentException if the given name collides with another
* aggregator in this scope
@@ -391,13 +391,13 @@ public abstract class DoFnWithContext
implements Serializable,
/**
* Returns an {@link Aggregator} with the aggregation logic specified by the
* {@link SerializableFunction} argument. The name provided must be unique
- * across {@link Aggregator}s created within the DoFn. Aggregators can only
be
+ * across {@link Aggregator}s created within the OldDoFn. Aggregators can
only be
* created during pipeline construction.
*
* @param name the name of the aggregator
* @param combiner the {@link SerializableFunction} to use in the aggregator
* @return an aggregator for the provided name and combiner in the scope of
- * this DoFn
+ * this OldDoFn
* @throws NullPointerException if the name or combiner is null
* @throws IllegalArgumentException if the given name collides with another
* aggregator in this scope
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java
index a31799e..4466874 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java
@@ -202,7 +202,7 @@ public class Filter extends PTransform,
PCollection> {
@Override
public PCollection apply(PCollection input) {
-PCollection output = i
[44/51] [abbrv] incubator-beam git commit: Use input type in coder inference for MapElements and FlatMapElements
Use input type in coder inference for MapElements and FlatMapElements
Previously, the input TypeDescriptor was unknown, so we would fail
to infer a coder for things like MapElements.of(SimpleFunction)
even if the input PCollection provided a coder for T.
Now, the input type is plumbed appropriately and the coder is inferred.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/4ac5cafe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/4ac5cafe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/4ac5cafe
Branch: refs/heads/python-sdk
Commit: 4ac5cafe90a371cf616f97cb202d5016b68616d1
Parents: 8daf518
Author: Kenneth Knowles
Authored: Fri Jul 29 10:35:01 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 20:18:59 2016 -0700
--
.../beam/sdk/transforms/FlatMapElements.java| 126 +--
.../apache/beam/sdk/transforms/MapElements.java | 60 +
.../beam/sdk/transforms/SimpleFunction.java | 34 +
.../sdk/transforms/FlatMapElementsTest.java | 48 +++
.../beam/sdk/transforms/MapElementsTest.java| 84 +
5 files changed, 288 insertions(+), 64 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4ac5cafe/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
index 694592e..04d993c 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java
@@ -17,8 +17,10 @@
*/
package org.apache.beam.sdk.transforms;
+import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
import java.lang.reflect.ParameterizedType;
@@ -45,8 +47,16 @@ extends PTransform,
PCollection> {
* descriptor need not be provided.
*/
public static MissingOutputTypeDescriptor
- via(SerializableFunction> fn) {
-return new MissingOutputTypeDescriptor<>(fn);
+ via(SerializableFunction> fn) {
+
+// TypeDescriptor interacts poorly with the wildcards needed to correctly
express
+// covariance and contravariance in Java, so instead we cast it to an
invariant
+// function here.
+@SuppressWarnings("unchecked") // safe covariant cast
+SerializableFunction> simplerFn =
+(SerializableFunction>) fn;
+
+return new MissingOutputTypeDescriptor<>(simplerFn);
}
/**
@@ -72,16 +82,15 @@ extends PTransform,
PCollection> {
* To use a Java 8 lambda, see {@link #via(SerializableFunction)}.
*/
public static FlatMapElements
- via(SimpleFunction> fn) {
-
-@SuppressWarnings({"rawtypes", "unchecked"}) // safe by static typing
-TypeDescriptor> iterableType = (TypeDescriptor)
fn.getOutputTypeDescriptor();
-
-@SuppressWarnings("unchecked") // safe by correctness of
getIterableElementType
-TypeDescriptor outputType =
-(TypeDescriptor) getIterableElementType(iterableType);
-
-return new FlatMapElements<>(fn, outputType);
+ via(SimpleFunction> fn) {
+// TypeDescriptor interacts poorly with the wildcards needed to correctly
express
+// covariance and contravariance in Java, so instead we cast it to an
invariant
+// function here.
+@SuppressWarnings("unchecked") // safe covariant cast
+SimpleFunction> simplerFn =
+(SimpleFunction>) fn;
+
+return new FlatMapElements<>(simplerFn, fn.getClass());
}
/**
@@ -91,18 +100,80 @@ extends PTransform,
PCollection> {
*/
public static final class MissingOutputTypeDescriptor {
-private final SerializableFunction> fn;
+private final SerializableFunction> fn;
private MissingOutputTypeDescriptor(
-SerializableFunction> fn) {
+SerializableFunction> fn) {
this.fn = fn;
}
public FlatMapElements
withOutputType(TypeDescriptor outputType) {
- return new FlatMapElements<>(fn, outputType);
+ TypeDescriptor> iterableOutputType =
TypeDescriptors.iterables(outputType);
+
+ return new FlatMapElements<>(
+ SimpleFunction.fromSerializableFunctionWithOutputType(fn,
+ iterableOutputType),
+ fn.getClass());
}
}
+
//
+
+ private final SimpleFunction> fn;
+ private final DisplayData.Item fnClassDisplayData;
+
+ private FlatMapEle
[07/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunnerBase.java
--
diff --git
a/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunnerBase.java
b/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunnerBase.java
index 2696020..ed9ec10 100644
---
a/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunnerBase.java
+++
b/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunnerBase.java
@@ -25,8 +25,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Aggregator.AggregatorFactory;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.DoFn.RequiresWindowAccess;
+import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.OldDoFn.RequiresWindowAccess;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
@@ -58,15 +58,15 @@ import java.util.Set;
*/
public abstract class DoFnRunnerBase implements
DoFnRunner {
- /** The DoFn being run. */
- public final DoFn fn;
+ /** The OldDoFn being run. */
+ public final OldDoFn fn;
- /** The context used for running the DoFn. */
+ /** The context used for running the OldDoFn. */
public final DoFnContext context;
protected DoFnRunnerBase(
PipelineOptions options,
- DoFn fn,
+ OldDoFn fn,
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag mainOutputTag,
@@ -145,7 +145,7 @@ public abstract class DoFnRunnerBase
implements DoFnRunner elem);
@@ -162,17 +162,17 @@ public abstract class DoFnRunnerBase
implements DoFnRunner the type of the DoFn's (main) input elements
- * @param the type of the DoFn's (main) output elements
+ * @param the type of the OldDoFn's (main) input elements
+ * @param the type of the OldDoFn's (main) output elements
*/
private static class DoFnContext
- extends DoFn.Context {
+ extends OldDoFn.Context {
private static final int MAX_SIDE_OUTPUTS = 1000;
final PipelineOptions options;
-final DoFn fn;
+final OldDoFn fn;
final SideInputReader sideInputReader;
final OutputManager outputManager;
final TupleTag mainOutputTag;
@@ -187,7 +187,7 @@ public abstract class DoFnRunnerBase
implements DoFnRunner> outputTags;
public DoFnContext(PipelineOptions options,
- DoFn fn,
+ OldDoFn fn,
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag mainOutputTag,
@@ -317,8 +317,8 @@ public abstract class DoFnRunnerBase
implements DoFnRunner
implements DoFnRunner.ProcessContext
createProcessContext(WindowedValue elem) {
+ protected OldDoFn.ProcessContext createProcessContext(
+ WindowedValue elem) {
return new DoFnProcessContext(fn, context, elem);
}
@@ -365,21 +366,21 @@ public abstract class DoFnRunnerBase
implements DoFnRunner the type of the DoFn's (main) input elements
- * @param the type of the DoFn's (main) output elements
+ * @param the type of the OldDoFn's (main) input elements
+ * @param the type of the OldDoFn's (main) output elements
*/
static class DoFnProcessContext
- extends DoFn.ProcessContext {
+ extends OldDoFn.ProcessContext {
-final DoFn fn;
+final OldDoFn fn;
final DoFnContext context;
final WindowedValue windowedValue;
-public DoFnProcessContext(DoFn fn,
+public DoFnProcessContext(OldDoFn fn,
DoFnContext context,
WindowedValue windowedValue) {
fn.super();
@@ -426,7 +427,8 @@ public abstract class DoFnRunnerBase
implements DoFnRunner
implements DoFnRunnerhttp://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunners.java
--
diff --git
a/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunners.java
b/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunners.java
index cb96da2..a9f3cf4 100644
--- a/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunners.java
+++ b/runners/core-java/src/main/java/org/apache/beam/sdk/util/DoFnRunners.java
@@ -19,7 +19,7 @@ package org.apache.beam.sdk.util;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.Aggregator.AggregatorFactory;
-import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.s
[47/51] [abbrv] incubator-beam git commit: Remove unneccssary Assignment in TransformExecutor
Remove unneccssary Assignment in TransformExecutor
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/acf71d31
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/acf71d31
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/acf71d31
Branch: refs/heads/python-sdk
Commit: acf71d313a43c8f39213ce19277ffebadcc40a77
Parents: 4546fd9
Author: Thomas Groh
Authored: Fri Aug 5 09:59:46 2016 -0700
Committer: Thomas Groh
Committed: Fri Aug 5 10:04:21 2016 -0700
--
.../java/org/apache/beam/runners/direct/TransformExecutor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/acf71d31/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
--
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
index 3db941d..d873bf5 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
@@ -118,7 +118,7 @@ class TransformExecutor implements Runnable {
processElements(evaluator, enforcements);
- TransformResult result = finishBundle(evaluator, enforcements);
+ finishBundle(evaluator, enforcements);
} catch (Throwable t) {
onComplete.handleThrowable(inputBundle, t);
if (t instanceof RuntimeException) {
[36/51] [abbrv] incubator-beam git commit: Port mentions of OldDoFn in PipelineOptions
Port mentions of OldDoFn in PipelineOptions
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/f5011e5c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/f5011e5c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/f5011e5c
Branch: refs/heads/python-sdk
Commit: f5011e5c62cb00fb4d8a91bd7d55d5083789a307
Parents: 620bd99
Author: Kenneth Knowles
Authored: Wed Aug 3 19:56:33 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../java/org/apache/beam/sdk/options/PipelineOptions.java | 9 -
1 file changed, 4 insertions(+), 5 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/f5011e5c/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
--
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
index 365f668..4595fc8 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
@@ -22,8 +22,8 @@ import
org.apache.beam.sdk.options.GoogleApiDebugOptions.GoogleApiTracer;
import org.apache.beam.sdk.options.ProxyInvocationHandler.Deserializer;
import org.apache.beam.sdk.options.ProxyInvocationHandler.Serializer;
import org.apache.beam.sdk.runners.PipelineRunner;
-import org.apache.beam.sdk.transforms.OldDoFn;
-import org.apache.beam.sdk.transforms.OldDoFn.Context;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Context;
import org.apache.beam.sdk.transforms.display.HasDisplayData;
import com.google.auto.service.AutoService;
@@ -35,7 +35,6 @@ import
com.fasterxml.jackson.databind.annotation.JsonSerialize;
import java.lang.reflect.Proxy;
import java.util.ServiceLoader;
-
import javax.annotation.concurrent.ThreadSafe;
/**
@@ -52,7 +51,7 @@ import javax.annotation.concurrent.ThreadSafe;
* and {@link PipelineOptionsFactory#as(Class)}. They can be created
* from command-line arguments with {@link
PipelineOptionsFactory#fromArgs(String[])}.
* They can be converted to another type by invoking {@link
PipelineOptions#as(Class)} and
- * can be accessed from within a {@link OldDoFn} by invoking
+ * can be accessed from within a {@link DoFn} by invoking
* {@link Context#getPipelineOptions()}.
*
* For example:
@@ -151,7 +150,7 @@ import javax.annotation.concurrent.ThreadSafe;
* {@link PipelineOptionsFactory#withValidation()} is invoked.
*
* {@link JsonIgnore @JsonIgnore} is used to prevent a property from being
serialized and
- * available during execution of {@link OldDoFn}. See the Serialization
section below for more
+ * available during execution of {@link DoFn}. See the Serialization section
below for more
* details.
*
* Registration Of PipelineOptions
[23/51] [abbrv] incubator-beam git commit: Closes #707
Closes #707 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/34d50127 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/34d50127 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/34d50127 Branch: refs/heads/python-sdk Commit: 34d501278344e90115c4baea3af6301c37f58972 Parents: 595d2d4 8db6114 Author: Dan Halperin Authored: Wed Aug 3 23:41:22 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 23:41:22 2016 -0700 -- .../beam/sdk/io/gcp/bigquery/BigQueryIO.java| 585 ++- .../sdk/io/gcp/bigquery/BigQueryServices.java | 7 + .../io/gcp/bigquery/BigQueryServicesImpl.java | 51 +- .../sdk/io/gcp/bigquery/BigQueryIOTest.java | 213 ++- 4 files changed, 693 insertions(+), 163 deletions(-) --
[12/51] [abbrv] incubator-beam git commit: Port DebuggingWordCount example from OldDoFn to DoFn
Port DebuggingWordCount example from OldDoFn to DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/49d2f170
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/49d2f170
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/49d2f170
Branch: refs/heads/python-sdk
Commit: 49d2f1706f69c5106a9082ffd2fecaf69b2d868c
Parents: ca9e337
Author: Kenneth Knowles
Authored: Fri Jul 22 14:29:18 2016 -0700
Committer: Dan Halperin
Committed: Wed Aug 3 18:25:53 2016 -0700
--
.../java/org/apache/beam/examples/DebuggingWordCount.java| 8
1 file changed, 4 insertions(+), 4 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/49d2f170/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
--
diff --git
a/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
b/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
index 3c43152..c1b273c 100644
---
a/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
+++
b/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
@@ -24,7 +24,7 @@ import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.transforms.Aggregator;
-import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.KV;
@@ -106,8 +106,8 @@ import java.util.regex.Pattern;
* overridden with {@code --inputFile}.
*/
public class DebuggingWordCount {
- /** A OldDoFn that filters for a specific key based upon a regular
expression. */
- public static class FilterTextFn extends OldDoFn,
KV> {
+ /** A DoFn that filters for a specific key based upon a regular expression.
*/
+ public static class FilterTextFn extends DoFn, KV> {
/**
* Concept #1: The logger below uses the fully qualified class name of
FilterTextFn
* as the logger. All log statements emitted by this logger will be
referenced by this name
@@ -133,7 +133,7 @@ public class DebuggingWordCount {
private final Aggregator unmatchedWords =
createAggregator("umatchedWords", new Sum.SumLongFn());
-@Override
+@ProcessElement
public void processElement(ProcessContext c) {
if (filter.matcher(c.element().getKey()).matches()) {
// Log at the "DEBUG" level each element that we match. When executing
this pipeline
[19/51] [abbrv] incubator-beam git commit: Closes #743
Closes #743 Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/3144363c Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/3144363c Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/3144363c Branch: refs/heads/python-sdk Commit: 3144363c067e8bd5d1b83aca2f4057b6bc33ed04 Parents: c314e67 ac0875d Author: Dan Halperin Authored: Wed Aug 3 23:01:15 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 23:01:15 2016 -0700 -- .../runners/spark/SparkPipelineOptions.java | 6 .../apache/beam/runners/spark/SparkRunner.java | 14 +++-- .../runners/spark/SparkRunnerRegistrar.java | 6 ++-- .../spark/SparkStreamingPipelineOptions.java| 32 .../beam/runners/spark/TestSparkRunner.java | 2 -- .../apache/beam/runners/spark/DeDupTest.java| 2 +- .../beam/runners/spark/EmptyInputTest.java | 2 +- .../beam/runners/spark/SimpleWordCountTest.java | 4 +-- .../runners/spark/SparkRunnerRegistrarTest.java | 2 +- .../apache/beam/runners/spark/TfIdfTest.java| 2 +- .../beam/runners/spark/io/AvroPipelineTest.java | 2 +- .../beam/runners/spark/io/NumShardsTest.java| 2 +- .../io/hadoop/HadoopFileFormatPipelineTest.java | 2 +- .../spark/translation/CombineGloballyTest.java | 2 +- .../spark/translation/CombinePerKeyTest.java| 2 +- .../spark/translation/DoFnOutputTest.java | 6 ++-- .../translation/MultiOutputWordCountTest.java | 2 +- .../spark/translation/SerializationTest.java| 2 +- .../spark/translation/SideEffectsTest.java | 8 ++--- .../streaming/FlattenStreamingTest.java | 8 ++--- .../streaming/KafkaStreamingTest.java | 13 .../streaming/SimpleStreamingWordCountTest.java | 8 ++--- .../streaming/utils/EmbeddedKafkaCluster.java | 4 ++- 23 files changed, 49 insertions(+), 84 deletions(-) --
[02/51] [abbrv] incubator-beam git commit: Rename DoFn to OldDoFn
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnWithContextTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnWithContextTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnWithContextTest.java
index 3b314b2..8b00c03 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnWithContextTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnWithContextTest.java
@@ -142,9 +142,9 @@ public class DoFnWithContextTest implements Serializable {
@Test
public void testDoFnWithContextUsingAggregators() {
NoOpDoFn noOpFn = new NoOpDoFn<>();
-DoFn.Context context = noOpFn.context();
+OldDoFn.Context context = noOpFn.context();
-DoFn fn = spy(noOpFn);
+OldDoFn fn = spy(noOpFn);
context = spy(context);
@SuppressWarnings("unchecked")
@@ -225,7 +225,7 @@ public class DoFnWithContextTest implements Serializable {
}
/**
- * Initialize a test pipeline with the specified {@link DoFn}.
+ * Initialize a test pipeline with the specified {@link OldDoFn}.
*/
private TestPipeline
createTestPipeline(DoFnWithContext fn) {
TestPipeline pipeline = TestPipeline.create();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
index 80825cb..b81eedb 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
@@ -130,7 +130,7 @@ public class FlattenTest implements Serializable {
PCollection output = p
.apply(Create.of((Void) null).withCoder(VoidCoder.of()))
-.apply(ParDo.withSideInputs(view).of(new DoFn() {
+.apply(ParDo.withSideInputs(view).of(new OldDoFn() {
@Override
public void processElement(ProcessContext c) {
for (String side : c.sideInput(view)) {
@@ -339,7 +339,7 @@ public class FlattenTest implements Serializable {
/
- private static class IdentityFn extends DoFn {
+ private static class IdentityFn extends OldDoFn {
@Override
public void processElement(ProcessContext c) {
c.output(c.element());
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
index d6e4589..15c3ba8 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
@@ -19,6 +19,7 @@ package org.apache.beam.sdk.transforms;
import static org.apache.beam.sdk.TestUtils.KvMatcher.isKv;
import static
org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.hasItem;
import static org.hamcrest.Matchers.empty;
@@ -55,7 +56,6 @@ import com.google.common.base.Function;
import com.google.common.collect.Iterables;
import com.fasterxml.jackson.annotation.JsonCreator;
-
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Assert;
@@ -371,7 +371,7 @@ public class GroupByKeyTest {
pipeline.run();
}
- private static class AssertTimestamp extends DoFn, Void> {
+ private static class AssertTimestamp extends OldDoFn, Void> {
private final Instant timestamp;
public AssertTimestamp(Instant timestamp) {
@@ -506,7 +506,7 @@ public class GroupByKeyTest {
* Creates a KV that wraps the original KV together with a random key.
*/
static class AssignRandomKey
- extends DoFn, KV>> {
+ extends OldDoFn, KV>> {
@Override
public void processElement(ProcessContext c) throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a64baf48/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/IntraBundleParallelizationTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/IntraBundleParallelizationTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/Intr
[37/51] [abbrv] incubator-beam git commit: Port easy Java SDK tests to new DoFn
Port easy Java SDK tests to new DoFn
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/1959ddbe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/1959ddbe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/1959ddbe
Branch: refs/heads/python-sdk
Commit: 1959ddbedb2ad61824bf28e1e9139cc677a2aaf5
Parents: f5011e5
Author: Kenneth Knowles
Authored: Wed Aug 3 20:15:12 2016 -0700
Committer: Kenneth Knowles
Committed: Thu Aug 4 14:56:42 2016 -0700
--
.../java/org/apache/beam/sdk/PipelineTest.java | 8 ++---
.../apache/beam/sdk/coders/AvroCoderTest.java | 6 ++--
.../beam/sdk/coders/CoderRegistryTest.java | 10 +++---
.../beam/sdk/coders/SerializableCoderTest.java | 10 +++---
.../apache/beam/sdk/io/CountingInputTest.java | 6 ++--
.../apache/beam/sdk/io/CountingSourceTest.java | 6 ++--
.../beam/sdk/io/PubsubUnboundedSinkTest.java| 6 ++--
.../sdk/transforms/ApproximateUniqueTest.java | 6 ++--
.../beam/sdk/transforms/CombineFnsTest.java | 4 +--
.../apache/beam/sdk/transforms/CombineTest.java | 18 +--
.../apache/beam/sdk/transforms/CreateTest.java | 4 +--
.../apache/beam/sdk/transforms/FlattenTest.java | 8 ++---
.../beam/sdk/transforms/GroupByKeyTest.java | 8 ++---
.../beam/sdk/transforms/WithTimestampsTest.java | 12 +++
.../display/DisplayDataEvaluatorTest.java | 10 +++---
.../sdk/transforms/display/DisplayDataTest.java | 6 ++--
.../sdk/transforms/join/CoGroupByKeyTest.java | 34 ++--
.../sdk/transforms/windowing/WindowTest.java| 10 +++---
.../sdk/transforms/windowing/WindowingTest.java | 23 ++---
.../beam/sdk/values/PCollectionTupleTest.java | 6 ++--
.../apache/beam/sdk/values/TypedPValueTest.java | 10 +++---
21 files changed, 106 insertions(+), 105 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1959ddbe/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
--
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
index 5137031..8b86499 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
@@ -36,8 +36,8 @@ import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.RunnableOnService;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
-import org.apache.beam.sdk.transforms.OldDoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.util.UserCodeException;
@@ -146,9 +146,9 @@ public class PipelineTest {
private static PTransform,
PCollection> addSuffix(
final String suffix) {
-return ParDo.of(new OldDoFn() {
- @Override
- public void processElement(OldDoFn.ProcessContext c) {
+return ParDo.of(new DoFn() {
+ @ProcessElement
+ public void processElement(DoFn.ProcessContext c) {
c.output(c.element() + suffix);
}
});
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1959ddbe/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/AvroCoderTest.java
--
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/AvroCoderTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/AvroCoderTest.java
index 41d0932..3b13e35 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/AvroCoderTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/AvroCoderTest.java
@@ -31,7 +31,7 @@ import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.util.CloudObject;
import org.apache.beam.sdk.util.SerializableUtils;
@@ -134,8 +134,8 @@ public class AvroCoderTest {
}
}
- private static class GetTextFn extends OldDoFn {
-@Override
+ private static class GetTextFn extends DoFn {
+@ProcessElement
public void processElement(ProcessContext c) {
c.output(c.element().text);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1959ddbe/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java
---
[22/51] [abbrv] incubator-beam git commit: [BEAM-383] Modified BigQueryIO to write based on number of files and file sizes
[BEAM-383] Modified BigQueryIO to write based on number of files and file sizes Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8db6114e Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8db6114e Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8db6114e Branch: refs/heads/python-sdk Commit: 8db6114e2087cafc4369f6ec85b04f978dfb1984 Parents: 595d2d4 Author: Ian Zhou Authored: Wed Jul 20 15:56:21 2016 -0700 Committer: Dan Halperin Committed: Wed Aug 3 23:40:27 2016 -0700 -- .../beam/sdk/io/gcp/bigquery/BigQueryIO.java| 585 ++- .../sdk/io/gcp/bigquery/BigQueryServices.java | 7 + .../io/gcp/bigquery/BigQueryServicesImpl.java | 51 +- .../sdk/io/gcp/bigquery/BigQueryIOTest.java | 213 ++- 4 files changed, 693 insertions(+), 163 deletions(-) -- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8db6114e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java -- diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java index 8741c9c..2ba7562 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java @@ -21,6 +21,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; +import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.AtomicCoder; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.Coder.Context; @@ -33,9 +34,6 @@ import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.coders.VoidCoder; import org.apache.beam.sdk.io.AvroSource; import org.apache.beam.sdk.io.BoundedSource; -import org.apache.beam.sdk.io.FileBasedSink; -import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition; -import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryServices.DatasetService; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryServices.JobService; import org.apache.beam.sdk.options.BigQueryOptions; @@ -44,6 +42,7 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.Aggregator; import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.GroupByKey; import org.apache.beam.sdk.transforms.OldDoFn; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.ParDo; @@ -52,7 +51,13 @@ import org.apache.beam.sdk.transforms.Sum; import org.apache.beam.sdk.transforms.View; import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.util.AttemptBoundedExponentialBackOff; +import org.apache.beam.sdk.util.FileIOChannelFactory; +import org.apache.beam.sdk.util.GcsIOChannelFactory; +import org.apache.beam.sdk.util.GcsUtil; import org.apache.beam.sdk.util.GcsUtil.GcsUtilFactory; import org.apache.beam.sdk.util.IOChannelFactory; import org.apache.beam.sdk.util.IOChannelUtils; @@ -80,6 +85,7 @@ import com.google.api.services.bigquery.model.Job; import com.google.api.services.bigquery.model.JobConfigurationExtract; import com.google.api.services.bigquery.model.JobConfigurationLoad; import com.google.api.services.bigquery.model.JobConfigurationQuery; +import com.google.api.services.bigquery.model.JobConfigurationTableCopy; import com.google.api.services.bigquery.model.JobReference; import com.google.api.services.bigquery.model.JobStatistics; import com.google.api.services.bigquery.model.JobStatus; @@ -93,6 +99,7 @@ import com.google.common.base.MoreObjects; import com.google.common.base.Strings; import com.google.common.collect.ImmutableList; import com.google.common.collect.Lists; +import com.google.common.io.CountingOutputStream; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; @@ -110,6 +117,8 @@ import java.io.Serializable; import java.nio.channels.Channels; import java.nio.channels.WritableByteChannel; import java
