[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