date:20161221


[ 
https://issues.apache.org/jira/browse/BEAM-469?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15768525#comment-15768525
 ] 

ASF GitHub Bot commented on BEAM-469:
-

GitHub user dhalperi opened a pull request:

https://github.com/apache/incubator-beam/pull/1680

[BEAM-XXX] Make KVCoder more efficient by removing unnecessary nesting

See [BEAM-469] for more information about why this is
correct.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/dhalperi/incubator-beam 
efficient-nested-coders

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-beam/pull/1680.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 #1680


commit 621e8250c9535d773c4f4440a34ea0833912b51f
Author: Dan Halperin 
Date:   2016-12-21T23:37:49Z

[BEAM-XXX] Make KVCoder more efficient by removing unnecessary nesting

See [BEAM-469] for more information about why this is
correct.




> NullableCoder optimized encoding via passthrough context
> 
>
> Key: BEAM-469
> URL: https://issues.apache.org/jira/browse/BEAM-469
> Project: Beam
>  Issue Type: Improvement
>  Components: sdk-java-core
>Reporter: Luke Cwik
>Assignee: Thomas Groh
>Priority: Trivial
>  Labels: backward-incompatible
> Fix For: 0.3.0-incubating
>
>
> NullableCoder should encode using the context given and not always use the 
> nested context. For coders which can efficiently encode in the outer context 
> such as StringUtf8Coder or ByteArrayCoder, we are forcing them to prefix 
> themselves with their length.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-beam pull request #1680: [BEAM-XXX] Make KVCoder more efficient by...

2016-12-21 Thread dhalperi

GitHub user dhalperi opened a pull request:

https://github.com/apache/incubator-beam/pull/1680

[BEAM-XXX] Make KVCoder more efficient by removing unnecessary nesting

See [BEAM-469] for more information about why this is
correct.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/dhalperi/incubator-beam 
efficient-nested-coders

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-beam/pull/1680.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 #1680


commit 621e8250c9535d773c4f4440a34ea0833912b51f
Author: Dan Halperin 
Date:   2016-12-21T23:37:49Z

[BEAM-XXX] Make KVCoder more efficient by removing unnecessary nesting

See [BEAM-469] for more information about why this is
correct.




---
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-1201) Remove producesSortedKeys from Source


[ 
https://issues.apache.org/jira/browse/BEAM-1201?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15768483#comment-15768483
 ] 

ASF GitHub Bot commented on BEAM-1201:
--

GitHub user dhalperi opened a pull request:

https://github.com/apache/incubator-beam/pull/1679

[BEAM-1201] Remove BoundedSource.producesSortedKeys

R: @jkff

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/dhalperi/incubator-beam 
remove-produces-sorted-keys

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-beam/pull/1679.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 #1679


commit ee15138543f8b9926466cf4e4dc6857b3173345e
Author: Dan Halperin 
Date:   2016-12-21T23:32:38Z

[BEAM-1201] Remove BoundedSource.producesSortedKeys

Unused and unclear; for more information see the linked JIRA.




> Remove producesSortedKeys from Source
> -
>
> Key: BEAM-1201
> URL: https://issues.apache.org/jira/browse/BEAM-1201
> Project: Beam
>  Issue Type: Improvement
>  Components: sdk-java-core
>Reporter: Daniel Halperin
>Assignee: Daniel Halperin
>Priority: Minor
>  Labels: backward-incompatible
>
> This is a holdover from a precursor of the old Dataflow SDK that we just 
> failed to delete before releasing Dataflow 1.0, but we can delete before the 
> first stable release of Beam.
> This function has never been used by any runner. It does not mean anything 
> obvious to implementors, as many sources produce {{T}}, not {{KV}} -- 
> what does it mean in the former case? (And how do you get the latter case 
> correct?)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (BEAM-1201) Remove producesSortedKeys from Source


 [ 
https://issues.apache.org/jira/browse/BEAM-1201?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Daniel Halperin updated BEAM-1201:
--
Labels: backward-incompatible  (was: )

> Remove producesSortedKeys from Source
> -
>
> Key: BEAM-1201
> URL: https://issues.apache.org/jira/browse/BEAM-1201
> Project: Beam
>  Issue Type: Improvement
>  Components: sdk-java-core
>Reporter: Daniel Halperin
>Assignee: Daniel Halperin
>Priority: Minor
>  Labels: backward-incompatible
>
> This is a holdover from a precursor of the old Dataflow SDK that we just 
> failed to delete before releasing Dataflow 1.0, but we can delete before the 
> first stable release of Beam.
> This function has never been used by any runner. It does not mean anything 
> obvious to implementors, as many sources produce {{T}}, not {{KV}} -- 
> what does it mean in the former case? (And how do you get the latter case 
> correct?)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (BEAM-1201) Remove producesSortedKeys from BoundedSource


 [ 
https://issues.apache.org/jira/browse/BEAM-1201?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Daniel Halperin updated BEAM-1201:
--
Summary: Remove producesSortedKeys from BoundedSource  (was: Remove 
producesSortedKeys from Source)

> Remove producesSortedKeys from BoundedSource
> 
>
> Key: BEAM-1201
> URL: https://issues.apache.org/jira/browse/BEAM-1201
> Project: Beam
>  Issue Type: Improvement
>  Components: sdk-java-core
>Reporter: Daniel Halperin
>Assignee: Daniel Halperin
>Priority: Minor
>  Labels: backward-incompatible
>
> This is a holdover from a precursor of the old Dataflow SDK that we just 
> failed to delete before releasing Dataflow 1.0, but we can delete before the 
> first stable release of Beam.
> This function has never been used by any runner. It does not mean anything 
> obvious to implementors, as many sources produce {{T}}, not {{KV}} -- 
> what does it mean in the former case? (And how do you get the latter case 
> correct?)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-beam pull request #1679: [BEAM-1201] Remove BoundedSource.produces...

2016-12-21 Thread dhalperi

GitHub user dhalperi opened a pull request:

https://github.com/apache/incubator-beam/pull/1679

[BEAM-1201] Remove BoundedSource.producesSortedKeys

R: @jkff

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/dhalperi/incubator-beam 
remove-produces-sorted-keys

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-beam/pull/1679.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 #1679


commit ee15138543f8b9926466cf4e4dc6857b3173345e
Author: Dan Halperin 
Date:   2016-12-21T23:32:38Z

[BEAM-1201] Remove BoundedSource.producesSortedKeys

Unused and unclear; for more information see the linked JIRA.




---
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-469) NullableCoder optimized encoding via passthrough context


[ 
https://issues.apache.org/jira/browse/BEAM-469?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15768476#comment-15768476
 ] 

Daniel Halperin commented on BEAM-469:
--

Sorry I missed this JIRA comment, [~mariusz89016]! A bit late, but...

Say a coder C does not have the nested context. Then we actually have the 
guarantee that no one will put later elements.

So if {{NullableCoder}} does not have the nested context, then no one will put 
more elements after whatever the {{NullableCoder}} puts. If the NC puts {{0}} 
then that's it -- the element is null. But if the NC puts {{1}}, then we know 
that all remaining bytes in the encoded string belong to the inner coder. That 
is effectively saying that the inner coder also does not need to have the 
nested context, so it does not need to write its own length.

In your example, the {{NullableCoder}} is used in an inner context. So the 
inner coder needs to also use the inner context, because there may be more 
encoded elements later.

In either case: the context of the nullable coder can be the same as the 
context of the inner coder. This is why in the patch here, we simply pass the 
NC's context down into the inner coder. All we have removed is the _additional_ 
nesting that was used.

https://patch-diff.githubusercontent.com/raw/apache/incubator-beam/pull/992.patch
 

> NullableCoder optimized encoding via passthrough context
> 
>
> Key: BEAM-469
> URL: https://issues.apache.org/jira/browse/BEAM-469
> Project: Beam
>  Issue Type: Improvement
>  Components: sdk-java-core
>Reporter: Luke Cwik
>Assignee: Thomas Groh
>Priority: Trivial
>  Labels: backward-incompatible
> Fix For: 0.3.0-incubating
>
>
> NullableCoder should encode using the context given and not always use the 
> nested context. For coders which can efficiently encode in the outer context 
> such as StringUtf8Coder or ByteArrayCoder, we are forcing them to prefix 
> themselves with their length.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (BEAM-646) Get runners out of the apply()


[ 
https://issues.apache.org/jira/browse/BEAM-646?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15768470#comment-15768470
 ] 

ASF GitHub Bot commented on BEAM-646:
-

Github user asfgit closed the pull request at:

https://github.com/apache/incubator-beam/pull/1582


> Get runners out of the apply()
> --
>
> Key: BEAM-646
> URL: https://issues.apache.org/jira/browse/BEAM-646
> Project: Beam
>  Issue Type: Improvement
>  Components: beam-model-runner-api, sdk-java-core
>Reporter: Kenneth Knowles
>Assignee: Thomas Groh
>
> Right now, the runner intercepts calls to apply() and replaces transforms as 
> we go. This means that there is no "original" user graph. For portability and 
> misc architectural benefits, we would like to build the original graph first, 
> and have the runner override later.
> Some runners already work in this manner, but we could integrate it more 
> smoothly, with more validation, via some handy APIs on e.g. the Pipeline 
> object.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-beam pull request #1582: [BEAM-646] Add Parameters to finishSpecif...

2016-12-21 Thread asfgit

Github user asfgit closed the pull request at:

https://github.com/apache/incubator-beam/pull/1582


---
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.
---

[1/3] incubator-beam git commit: Add Parameters to finishSpecifying

2016-12-21 Thread tgroh

Repository: incubator-beam
Updated Branches:
  refs/heads/master 57d9bbd79 -> 7984fe3fc


Add Parameters to finishSpecifying

Remove the need to use getProducingTransformInternal in TypedPValue.

Ensure that all nodes are finished specifying before a call to
TransformHierarchy#visit. This ensures that all nodes are fully
specified without requiring the Pipeline or Runner to do so explicitly.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/038950df
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/038950df
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/038950df

Branch: refs/heads/master
Commit: 038950df02fa553cbb91f57978efe125a9ebc80f
Parents: b053be4
Author: Thomas Groh 
Authored: Thu Dec 8 14:33:36 2016 -0800
Committer: Thomas Groh 
Committed: Wed Dec 21 15:26:17 2016 -0800

--
 .../translation/ParDoBoundTranslatorTest.java   | 32 
 .../beam/runners/direct/DirectGraphVisitor.java | 21 --
 .../beam/runners/direct/DirectRunner.java   |  1 -
 .../runners/direct/DirectGraphVisitorTest.java  | 32 +---
 .../direct/FlattenEvaluatorFactoryTest.java |  2 +
 .../direct/KeyedPValueTrackingVisitorTest.java  | 17 -
 .../beam/runners/spark/ForceStreamingTest.java  |  2 -
 .../main/java/org/apache/beam/sdk/Pipeline.java |  3 +
 .../beam/sdk/runners/TransformHierarchy.java| 45 ++-
 .../transforms/join/KeyedPCollectionTuple.java  | 32 
 .../java/org/apache/beam/sdk/values/PBegin.java |  5 --
 .../apache/beam/sdk/values/PCollectionList.java | 13 +---
 .../beam/sdk/values/PCollectionTuple.java   | 13 +---
 .../java/org/apache/beam/sdk/values/PInput.java |  9 ---
 .../org/apache/beam/sdk/values/POutput.java | 20 ++---
 .../beam/sdk/values/POutputValueBase.java   |  4 +-
 .../java/org/apache/beam/sdk/values/PValue.java | 15 
 .../org/apache/beam/sdk/values/PValueBase.java  |  3 +-
 .../org/apache/beam/sdk/values/TypedPValue.java | 78 +++-
 .../sdk/runners/TransformHierarchyTest.java | 34 +
 .../apache/beam/sdk/transforms/ParDoTest.java   |  7 +-
 .../apache/beam/sdk/values/TypedPValueTest.java |  7 +-
 22 files changed, 185 insertions(+), 210 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/038950df/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslatorTest.java
--
diff --git 
a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslatorTest.java
 
b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslatorTest.java
index fa94b2a..f88a94d 100644
--- 
a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslatorTest.java
+++ 
b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslatorTest.java
@@ -45,6 +45,7 @@ import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VoidCoder;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.transforms.Create;
@@ -286,21 +287,22 @@ public class ParDoBoundTranslatorTest {
 Arrays.asList(sideInput1, sideInput2),
 Arrays.asList(;
 
- outputs.get(mainOutputTag).apply(ParDo.of(new EmbeddedCollector()));
- ApexRunnerResult result = (ApexRunnerResult) pipeline.run();
-
- HashSet expected = Sets.newHashSet("processing: 3: [11, 222]",
- "processing: -42: [11, 222]", "processing: 666: [11, 222]");
- long timeout = System.currentTimeMillis() + TIMEOUT_MILLIS;
- while (System.currentTimeMillis() < timeout) {
-   if (EmbeddedCollector.RESULTS.containsAll(expected)) {
- break;
-   }
-   LOG.info("Waiting for expected results.");
-   Thread.sleep(SLEEP_MILLIS);
- }
- result.cancel();
- Assert.assertEquals(Sets.newHashSet(expected), EmbeddedCollector.RESULTS);
+outputs.get(mainOutputTag).apply(ParDo.of(new EmbeddedCollector()));
+outputs.get(sideOutputTag).setCoder(VoidCoder.of());
+ApexRunnerResult result = (ApexRunnerResult) pipeline.run();
+
+HashSet expected = Sets.newHashSet("processing: 3: [11, 222]",
+"processing: -42: [11, 222]", "processing: 666: [11, 222]");
+long timeout = System.currentTimeMillis() + TIMEOUT_MILLIS;
+while (System.currentTimeMillis() < timeout) {
+  if (EmbeddedCollector.RESULTS.containsAll(expected)) {
+break;
+  }
+  LOG.info("Waiting for expected

[2/3] incubator-beam git commit: Use CountingSource in ForceStreamingTest

2016-12-21 Thread tgroh

Use CountingSource in ForceStreamingTest

Removes the requirement to have a FakeUnboundedSource, plus the read is
fully specified.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/b053be46
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/b053be46
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/b053be46

Branch: refs/heads/master
Commit: b053be460c2e6ff486faed1b1a0996af63f93db2
Parents: 57d9bbd
Author: Thomas Groh 
Authored: Tue Dec 20 14:23:21 2016 -0800
Committer: Thomas Groh 
Committed: Wed Dec 21 15:26:17 2016 -0800

--
 .../beam/runners/spark/ForceStreamingTest.java  | 39 +---
 1 file changed, 2 insertions(+), 37 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/b053be46/runners/spark/src/test/java/org/apache/beam/runners/spark/ForceStreamingTest.java
--
diff --git 
a/runners/spark/src/test/java/org/apache/beam/runners/spark/ForceStreamingTest.java
 
b/runners/spark/src/test/java/org/apache/beam/runners/spark/ForceStreamingTest.java
index eb17eea..1b2ff08 100644
--- 
a/runners/spark/src/test/java/org/apache/beam/runners/spark/ForceStreamingTest.java
+++ 
b/runners/spark/src/test/java/org/apache/beam/runners/spark/ForceStreamingTest.java
@@ -23,10 +23,9 @@ import static org.hamcrest.MatcherAssert.assertThat;
 import java.io.IOException;
 import java.util.List;
 import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.io.BoundedReadFromUnboundedSource;
+import org.apache.beam.sdk.io.CountingSource;
 import org.apache.beam.sdk.io.Read;
-import org.apache.beam.sdk.io.UnboundedSource;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.runners.TransformHierarchy;
@@ -59,7 +58,7 @@ public class ForceStreamingTest {
 // apply the BoundedReadFromUnboundedSource.
 @SuppressWarnings("unchecked")
 BoundedReadFromUnboundedSource boundedRead =
-Read.from(new FakeUnboundedSource()).withMaxNumRecords(-1);
+Read.from(CountingSource.unbounded()).withMaxNumRecords(-1);
 //noinspection unchecked
 pipeline.apply(boundedRead);
 
@@ -86,38 +85,4 @@ public class ForceStreamingTest {
 }
 
   }
-
-  /**
-   * A fake {@link UnboundedSource} to satisfy the compiler.
-   */
-  private static class FakeUnboundedSource extends UnboundedSource {
-
-@Override
-public List generateInitialSplits(
-int desiredNumSplits,
-PipelineOptions options) throws Exception {
-  return null;
-}
-
-@Override
-public UnboundedReader createReader(
-PipelineOptions options,
-CheckpointMark checkpointMark) throws IOException {
-  return null;
-}
-
-@Override
-public Coder getCheckpointMarkCoder() {
-  return null;
-}
-
-@Override
-public void validate() { }
-
-@Override
-public Coder getDefaultOutputCoder() {
-  return null;
-}
-  }
-
 }

[3/3] incubator-beam git commit: This closes #1582

2016-12-21 Thread tgroh

This closes #1582


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7984fe3f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7984fe3f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7984fe3f

Branch: refs/heads/master
Commit: 7984fe3fc20160d2286433434190f35658aef158
Parents: 57d9bbd 038950d
Author: Thomas Groh 
Authored: Wed Dec 21 15:26:18 2016 -0800
Committer: Thomas Groh 
Committed: Wed Dec 21 15:26:18 2016 -0800

--
 .../translation/ParDoBoundTranslatorTest.java   | 32 
 .../beam/runners/direct/DirectGraphVisitor.java | 21 --
 .../beam/runners/direct/DirectRunner.java   |  1 -
 .../runners/direct/DirectGraphVisitorTest.java  | 32 +---
 .../direct/FlattenEvaluatorFactoryTest.java |  2 +
 .../direct/KeyedPValueTrackingVisitorTest.java  | 17 -
 .../beam/runners/spark/ForceStreamingTest.java  | 41 +-
 .../main/java/org/apache/beam/sdk/Pipeline.java |  3 +
 .../beam/sdk/runners/TransformHierarchy.java| 45 ++-
 .../transforms/join/KeyedPCollectionTuple.java  | 32 
 .../java/org/apache/beam/sdk/values/PBegin.java |  5 --
 .../apache/beam/sdk/values/PCollectionList.java | 13 +---
 .../beam/sdk/values/PCollectionTuple.java   | 13 +---
 .../java/org/apache/beam/sdk/values/PInput.java |  9 ---
 .../org/apache/beam/sdk/values/POutput.java | 20 ++---
 .../beam/sdk/values/POutputValueBase.java   |  4 +-
 .../java/org/apache/beam/sdk/values/PValue.java | 15 
 .../org/apache/beam/sdk/values/PValueBase.java  |  3 +-
 .../org/apache/beam/sdk/values/TypedPValue.java | 78 +++-
 .../sdk/runners/TransformHierarchyTest.java | 34 +
 .../apache/beam/sdk/transforms/ParDoTest.java   |  7 +-
 .../apache/beam/sdk/values/TypedPValueTest.java |  7 +-
 22 files changed, 187 insertions(+), 247 deletions(-)
--

[jira] [Commented] (BEAM-1112) Python E2E Integration Test Framework - Batch Only


[ 
https://issues.apache.org/jira/browse/BEAM-1112?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15768461#comment-15768461
 ] 

ASF GitHub Bot commented on BEAM-1112:
--

Github user markflyhigh closed the pull request at:

https://github.com/apache/incubator-beam/pull/1639


> Python E2E Integration Test Framework - Batch Only
> --
>
> Key: BEAM-1112
> URL: https://issues.apache.org/jira/browse/BEAM-1112
> Project: Beam
>  Issue Type: Task
>  Components: sdk-py, testing
>Reporter: Mark Liu
>Assignee: Mark Liu
>
> Parity with Java. 
> Build e2e integration test framework that can configure and run batch 
> pipeline with specified test runner, wait for pipeline execution and verify 
> results with given verifiers in the end.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (BEAM-1112) Python E2E Integration Test Framework - Batch Only


[ 
https://issues.apache.org/jira/browse/BEAM-1112?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15768462#comment-15768462
 ] 

ASF GitHub Bot commented on BEAM-1112:
--

GitHub user markflyhigh reopened a pull request:

https://github.com/apache/incubator-beam/pull/1639

[BEAM-1112] Python E2E Test Framework And Wordcount E2E Test

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).
 - [ ] 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).

---

 - E2e test framework that supports TestRunner start and verify pipeline 
job.
   - add `TestOptions` which defined `on_success_matcher` that is used to 
verify state/output of pipeline job.
   - validate `on_success_matcher` before pipeline execution to make sure 
it's unpicklable to a subclass of BaseMatcher.
   - create a `TestDataflowRunner` which provide functionalities of 
`DataflowRunner` plus result verification.
   - provide a test verifier `PipelineStateMatcher` that can verify 
pipeline job finished in DONE or not.
 - Add wordcount_it (it = integration test) that build e2e test based on 
existing wordcount pipeline.
   - include wordcount_it to nose collector, so that wordcount_it can be 
collected and run by nose.
   - skip ITs when running unit tests from tox in precommit and postcommit.

Current changes will not change behavior of existing pre/postcommit.
Test is done by running `tox -e py27 -c sdks/python/tox.ini` for unit test 
and running wordcount_it with `TestDataflowRunner` on service 
([link](https://pantheon.corp.google.com/dataflow/job/2016-12-15_17_36_16-3857167705491723621?project=google.com:clouddfe)).

TODO:
 - Output data verifier that verify pipeline output that stores in 
filesystem.
 - Add wordcount_it to precommit and replace existing wordcount execution 
command in postcommit with a better structured nose command.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/markflyhigh/incubator-beam e2e-testrunner

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-beam/pull/1639.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 #1639






> Python E2E Integration Test Framework - Batch Only
> --
>
> Key: BEAM-1112
> URL: https://issues.apache.org/jira/browse/BEAM-1112
> Project: Beam
>  Issue Type: Task
>  Components: sdk-py, testing
>Reporter: Mark Liu
>Assignee: Mark Liu
>
> Parity with Java. 
> Build e2e integration test framework that can configure and run batch 
> pipeline with specified test runner, wait for pipeline execution and verify 
> results with given verifiers in the end.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] incubator-beam pull request #1639: [BEAM-1112] Python E2E Test Framework And...

2016-12-21 Thread markflyhigh

GitHub user markflyhigh reopened a pull request:

https://github.com/apache/incubator-beam/pull/1639

[BEAM-1112] Python E2E Test Framework And Wordcount E2E Test

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).
- [ ] 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).

---

- E2e test framework that supports TestRunner start and verify pipeline
job.
- add `TestOptions` which defined `on_success_matcher` that is used to
verify state/output of pipeline job.
- validate `on_success_matcher` before pipeline execution to make sure
it's unpicklable to a subclass of BaseMatcher.
- create a `TestDataflowRunner` which provide functionalities of
`DataflowRunner` plus result verification.
- provide a test verifier `PipelineStateMatcher` that can verify
pipeline job finished in DONE or not.
- Add wordcount_it (it = integration test) that build e2e test based on
existing wordcount pipeline.
- include wordcount_it to nose collector, so that wordcount_it can be
collected and run by nose.
- skip ITs when running unit tests from tox in precommit and postcommit.

Current changes will not change behavior of existing pre/postcommit.
Test is done by running `tox -e py27 -c sdks/python/tox.ini` for unit test
and running wordcount_it with `TestDataflowRunner` on service
([link](https://pantheon.corp.google.com/dataflow/job/2016-12-15_17_36_16-3857167705491723621?project=google.com:clouddfe)).

TODO:
- Output data verifier that verify pipeline output that stores in
filesystem.
- Add wordcount_it to precommit and replace existing wordcount execution
command in postcommit with a better structured nose command.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/markflyhigh/incubator-beam e2e-testrunner

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-beam/pull/1639.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 #1639

---
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 #1639: [BEAM-1112] Python E2E Test Framework And...

2016-12-21 Thread markflyhigh

Github user markflyhigh closed the pull request at:

https://github.com/apache/incubator-beam/pull/1639


---
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.
---

[1/2] incubator-beam git commit: Add support for date partitioned table names

Repository: incubator-beam
Updated Branches:
  refs/heads/python-sdk bb09c07b6 -> 409d067b3


Add support for date partitioned table names

These names have the format "tablename$mmdd".  Previously the dollar
sign caused this to be deemed invalid.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a1af871a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a1af871a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a1af871a

Branch: refs/heads/python-sdk
Commit: a1af871a0c8c92a6d84f2e9950615f7737118d7e
Parents: bb09c07
Author: Kevin Graney 
Authored: Tue Dec 6 15:09:42 2016 -0500
Committer: Robert Bradshaw 
Committed: Wed Dec 21 15:16:45 2016 -0800

--
 sdks/python/apache_beam/io/bigquery.py  | 6 --
 sdks/python/apache_beam/io/bigquery_test.py | 8 
 2 files changed, 12 insertions(+), 2 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a1af871a/sdks/python/apache_beam/io/bigquery.py
--
diff --git a/sdks/python/apache_beam/io/bigquery.py 
b/sdks/python/apache_beam/io/bigquery.py
index ce75e10..2059de4 100644
--- a/sdks/python/apache_beam/io/bigquery.py
+++ b/sdks/python/apache_beam/io/bigquery.py
@@ -275,7 +275,9 @@ def _parse_table_reference(table, dataset=None, 
project=None):
   then the table argument must contain the entire table reference:
   'DATASET.TABLE' or 'PROJECT:DATASET.TABLE'. This argument can be a
   bigquery.TableReference instance in which case dataset and project are
-  ignored and the reference is returned as a result.
+  ignored and the reference is returned as a result.  Additionally, for 
date
+  partitioned tables, appending '$mmdd' to the table name is supported,
+  e.g. 'DATASET.TABLE$mmdd'.
 dataset: The ID of the dataset containing this table or null if the table
   reference is specified entirely by the table argument.
 project: The ID of the project containing this table or null if the table
@@ -300,7 +302,7 @@ def _parse_table_reference(table, dataset=None, 
project=None):
   # table name.
   if dataset is None:
 match = re.match(
-r'^((?P.+):)?(?P\w+)\.(?P\w+)$', table)
+r'^((?P.+):)?(?P\w+)\.(?P[\w\$]+)$', table)
 if not match:
   raise ValueError(
   'Expected a table reference (PROJECT:DATASET.TABLE or '

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a1af871a/sdks/python/apache_beam/io/bigquery_test.py
--
diff --git a/sdks/python/apache_beam/io/bigquery_test.py 
b/sdks/python/apache_beam/io/bigquery_test.py
index a2cf947..f6f9363 100644
--- a/sdks/python/apache_beam/io/bigquery_test.py
+++ b/sdks/python/apache_beam/io/bigquery_test.py
@@ -208,6 +208,14 @@ class TestBigQuerySource(unittest.TestCase):
 self.assertEqual(source.query, 'my_query')
 self.assertIsNone(source.table_reference)
 
+  def test_date_partitioned_table_name(self):
+source = beam.io.BigQuerySource('dataset.table$20030102', validate=True)
+dd = DisplayData.create_from(source)
+expected_items = [
+DisplayDataItemMatcher('validation', True),
+DisplayDataItemMatcher('table', 'dataset.table$20030102')]
+hc.assert_that(dd.items, hc.contains_inanyorder(*expected_items))
+
 
 class TestBigQuerySink(unittest.TestCase):

[2/2] incubator-beam git commit: Closes #1534

Closes #1534


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/409d067b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/409d067b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/409d067b

Branch: refs/heads/python-sdk
Commit: 409d067b36036981e330a055b652bb74a93f4ca2
Parents: bb09c07 a1af871
Author: Robert Bradshaw 
Authored: Wed Dec 21 15:16:46 2016 -0800
Committer: Robert Bradshaw 
Committed: Wed Dec 21 15:16:46 2016 -0800

--
 sdks/python/apache_beam/io/bigquery.py  | 6 --
 sdks/python/apache_beam/io/bigquery_test.py | 8 
 2 files changed, 12 insertions(+), 2 deletions(-)
--

[1/2] incubator-beam git commit: Fixing inconsistencies in PipelineOptions

Repository: incubator-beam
Updated Branches:
  refs/heads/python-sdk 3454d691f -> bb09c07b6


Fixing inconsistencies in PipelineOptions

The following options have changed:

* job_name - Default is 'beamapp-username-date-microseconds'. Test was added.
* staging_location and temp_location - staging_location was the default of
  temp_location. Now it's the other way around, and the tests reflect that.
* machine_type alias of worker_machine_type has been removed.
* disk_type alias of worker_disk_type has been removed.
* disk_source_image option has been removed.
* no_save_main_session option has been removed.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/35e2fdc7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/35e2fdc7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/35e2fdc7

Branch: refs/heads/python-sdk
Commit: 35e2fdc7f22f20d74a569e86ced931209661dec1
Parents: 3454d69
Author: Pablo 
Authored: Tue Dec 6 18:01:54 2016 -0800
Committer: Robert Bradshaw 
Committed: Wed Dec 21 15:14:52 2016 -0800

--
 sdks/python/apache_beam/internal/apiclient.py   | 45 
 .../apache_beam/internal/apiclient_test.py  |  6 +++
 sdks/python/apache_beam/utils/options.py| 33 ++
 .../utils/pipeline_options_validator.py | 11 ++---
 .../utils/pipeline_options_validator_test.py|  8 ++--
 5 files changed, 54 insertions(+), 49 deletions(-)
--


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/35e2fdc7/sdks/python/apache_beam/internal/apiclient.py
--
diff --git a/sdks/python/apache_beam/internal/apiclient.py 
b/sdks/python/apache_beam/internal/apiclient.py
index f1341a7..3a9ba46 100644
--- a/sdks/python/apache_beam/internal/apiclient.py
+++ b/sdks/python/apache_beam/internal/apiclient.py
@@ -18,6 +18,8 @@
 """Dataflow client utility functions."""
 
 import codecs
+from datetime import datetime
+import getpass
 import json
 import logging
 import os
@@ -46,10 +48,6 @@ from apache_beam.utils.options import WorkerOptions
 from apache_beam.internal.clients import storage
 import apache_beam.internal.clients.dataflow as dataflow
 
-BIGQUERY_API_SERVICE = 'bigquery.googleapis.com'
-COMPUTE_API_SERVICE = 'compute.googleapis.com'
-STORAGE_API_SERVICE = 'storage.googleapis.com'
-
 
 class Step(object):
   """Wrapper for a dataflow Step protobuf."""
@@ -121,11 +119,13 @@ class Environment(object):
 self.worker_options = options.view_as(WorkerOptions)
 self.debug_options = options.view_as(DebugOptions)
 self.proto = dataflow.Environment()
-self.proto.clusterManagerApiService = COMPUTE_API_SERVICE
-self.proto.dataset = '%s/cloud_dataflow' % BIGQUERY_API_SERVICE
+self.proto.clusterManagerApiService = 
GoogleCloudOptions.COMPUTE_API_SERVICE
+self.proto.dataset = '{}/cloud_dataflow'.format(
+GoogleCloudOptions.BIGQUERY_API_SERVICE)
 self.proto.tempStoragePrefix = (
-self.google_cloud_options.temp_location.replace('gs:/',
-STORAGE_API_SERVICE))
+self.google_cloud_options.temp_location.replace(
+'gs:/',
+GoogleCloudOptions.STORAGE_API_SERVICE))
 # User agent information.
 self.proto.userAgent = dataflow.Environment.UserAgentValue()
 self.local = 'localhost' in self.google_cloud_options.dataflow_endpoint
@@ -165,7 +165,7 @@ class Environment(object):
   dataflow.Package(
   location='%s/%s' % (
   self.google_cloud_options.staging_location.replace(
-  'gs:/', STORAGE_API_SERVICE),
+  'gs:/', GoogleCloudOptions.STORAGE_API_SERVICE),
   package),
   name=package))
 
@@ -174,7 +174,7 @@ class Environment(object):
 packages=package_descriptors,
 taskrunnerSettings=dataflow.TaskRunnerSettings(
 parallelWorkerSettings=dataflow.WorkerSettings(
-baseUrl='https://dataflow.googleapis.com',
+baseUrl=GoogleCloudOptions.DATAFLOW_ENDPOINT,
 servicePath=self.google_cloud_options.dataflow_endpoint)))
 pool.autoscalingSettings = dataflow.AutoscalingSettings()
 # Set worker pool options received through command line.
@@ -195,8 +195,6 @@ class Environment(object):
   pool.diskSizeGb = self.worker_options.disk_size_gb
 if self.worker_options.disk_type:
   pool.diskType = self.worker_options.disk_type
-if self.worker_options.disk_source_image:
-  pool.diskSourceImage = self.worker_options.disk_source_image
 if self.worker_options.zone:
   pool.zone = self.worker_options.zone
 if

[2/2] incubator-beam git commit: Closes #1526

Closes #1526


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/bb09c07b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/bb09c07b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/bb09c07b

Branch: refs/heads/python-sdk
Commit: bb09c07b6351dcc53c0bdc8bf1259261ad2edfba
Parents: 3454d69 35e2fdc
Author: Robert Bradshaw 
Authored: Wed Dec 21 15:15:20 2016 -0800
Committer: Robert Bradshaw 
Committed: Wed Dec 21 15:15:20 2016 -0800

--
 sdks/python/apache_beam/internal/apiclient.py   | 45 
 .../apache_beam/internal/apiclient_test.py  |  6 +++
 sdks/python/apache_beam/utils/options.py| 33 ++
 .../utils/pipeline_options_validator.py | 11 ++---
 .../utils/pipeline_options_validator_test.py|  8 ++--
 5 files changed, 54 insertions(+), 49 deletions(-)
--

[jira] [Created] (BEAM-1202) Coders should have meaningful equals methods

2016-12-21 Thread Thomas Groh (JIRA)

Thomas Groh created BEAM-1202:
-

 Summary: Coders should have meaningful equals methods
 Key: BEAM-1202
 URL: https://issues.apache.org/jira/browse/BEAM-1202
 Project: Beam
  Issue Type: Bug
  Components: sdk-java-core
Reporter: Thomas Groh
Assignee: Davor Bonaci


{{StandardCoder}} implements an equality check based on the component coders 
and equal classes. Any coder that is configured, or that does not extend 
{{StandardCoder}}, should have meaningful implementations of {{equals}} and 
{{hashCode}}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (BEAM-1202) Coders should have meaningful equals methods

2016-12-21 Thread Thomas Groh (JIRA)


 [ 
https://issues.apache.org/jira/browse/BEAM-1202?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Thomas Groh updated BEAM-1202:
--
Assignee: (was: Davor Bonaci)

> Coders should have meaningful equals methods
> 
>
> Key: BEAM-1202
> URL: https://issues.apache.org/jira/browse/BEAM-1202
> Project: Beam
>  Issue Type: Bug
>  Components: sdk-java-core
>Reporter: Thomas Groh
>
> {{StandardCoder}} implements an equality check based on the component coders 
> and equal classes. Any coder that is configured, or that does not extend 
> {{StandardCoder}}, should have meaningful implementations of {{equals}} and 
> {{hashCode}}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (BEAM-1201) Remove producesSortedKeys from Source