You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2017/03/08 22:25:16 UTC
[1/9] beam git commit: Revert "[maven-release-plugin] prepare release
v0.6.0-RC1"
Repository: beam
Updated Branches:
refs/heads/release-0.6.0 1c91d0e59 -> b25a0369d
Revert "[maven-release-plugin] prepare release v0.6.0-RC1"
This reverts commit 1c91d0e598c5dd62a9c34cc0bf4bfcc9a6f2faae.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/8ab36fa2
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/8ab36fa2
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/8ab36fa2
Branch: refs/heads/release-0.6.0
Commit: 8ab36fa2d5dfc237c7619c278848e31f3412a0e1
Parents: 1c91d0e
Author: Ahmet Altay <al...@google.com>
Authored: Wed Mar 8 13:37:57 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:37:57 2017 -0800
----------------------------------------------------------------------
examples/java/pom.xml | 2 +-
examples/java8/pom.xml | 2 +-
examples/pom.xml | 2 +-
pom.xml | 4 ++--
runners/apex/pom.xml | 2 +-
runners/core-construction-java/pom.xml | 2 +-
runners/core-java/pom.xml | 2 +-
runners/direct-java/pom.xml | 2 +-
runners/flink/examples/pom.xml | 2 +-
runners/flink/pom.xml | 2 +-
runners/flink/runner/pom.xml | 2 +-
runners/google-cloud-dataflow-java/pom.xml | 2 +-
runners/pom.xml | 2 +-
runners/spark/pom.xml | 2 +-
sdks/common/fn-api/pom.xml | 2 +-
sdks/common/pom.xml | 2 +-
sdks/common/runner-api/pom.xml | 2 +-
sdks/java/build-tools/pom.xml | 2 +-
sdks/java/core/pom.xml | 2 +-
sdks/java/extensions/jackson/pom.xml | 2 +-
sdks/java/extensions/join-library/pom.xml | 2 +-
sdks/java/extensions/pom.xml | 2 +-
sdks/java/extensions/sorter/pom.xml | 2 +-
sdks/java/harness/pom.xml | 2 +-
sdks/java/io/elasticsearch/pom.xml | 2 +-
sdks/java/io/google-cloud-platform/pom.xml | 2 +-
sdks/java/io/hadoop-common/pom.xml | 2 +-
sdks/java/io/hbase/pom.xml | 2 +-
sdks/java/io/hdfs/pom.xml | 2 +-
sdks/java/io/jdbc/pom.xml | 2 +-
sdks/java/io/jms/pom.xml | 2 +-
sdks/java/io/kafka/pom.xml | 2 +-
sdks/java/io/kinesis/pom.xml | 2 +-
sdks/java/io/mongodb/pom.xml | 2 +-
sdks/java/io/mqtt/pom.xml | 2 +-
sdks/java/io/pom.xml | 2 +-
sdks/java/java8tests/pom.xml | 2 +-
sdks/java/javadoc/pom.xml | 2 +-
sdks/java/maven-archetypes/examples-java8/pom.xml | 2 +-
sdks/java/maven-archetypes/examples/pom.xml | 2 +-
sdks/java/maven-archetypes/pom.xml | 2 +-
sdks/java/maven-archetypes/starter/pom.xml | 2 +-
sdks/java/pom.xml | 2 +-
sdks/pom.xml | 2 +-
sdks/python/pom.xml | 2 +-
45 files changed, 46 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/examples/java/pom.xml
----------------------------------------------------------------------
diff --git a/examples/java/pom.xml b/examples/java/pom.xml
index 3fc0118..9da814b 100644
--- a/examples/java/pom.xml
+++ b/examples/java/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-examples-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/examples/java8/pom.xml
----------------------------------------------------------------------
diff --git a/examples/java8/pom.xml b/examples/java8/pom.xml
index d00dbc9..580a154 100644
--- a/examples/java8/pom.xml
+++ b/examples/java8/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-examples-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/examples/pom.xml
----------------------------------------------------------------------
diff --git a/examples/pom.xml b/examples/pom.xml
index f5f7175..550578b 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index fbaafb7..eded684 100644
--- a/pom.xml
+++ b/pom.xml
@@ -34,7 +34,7 @@
<url>http://beam.apache.org/</url>
<inceptionYear>2016</inceptionYear>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<licenses>
<license>
@@ -48,7 +48,7 @@
<connection>scm:git:https://git-wip-us.apache.org/repos/asf/beam.git</connection>
<developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/beam.git</developerConnection>
<url>https://git-wip-us.apache.org/repos/asf?p=beam.git;a=summary</url>
- <tag>v0.6.0-RC1</tag>
+ <tag>release-0.6.0</tag>
</scm>
<issueManagement>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/apex/pom.xml
----------------------------------------------------------------------
diff --git a/runners/apex/pom.xml b/runners/apex/pom.xml
index 1401e3d..f5fe4bc 100644
--- a/runners/apex/pom.xml
+++ b/runners/apex/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/core-construction-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/pom.xml b/runners/core-construction-java/pom.xml
index ce96be3..b602f5d 100644
--- a/runners/core-construction-java/pom.xml
+++ b/runners/core-construction-java/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>beam-runners-parent</artifactId>
<groupId>org.apache.beam</groupId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/core-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/core-java/pom.xml b/runners/core-java/pom.xml
index 5db59ff..28ab0ee 100644
--- a/runners/core-java/pom.xml
+++ b/runners/core-java/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/direct-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
index d172b2b..ced9cd6 100644
--- a/runners/direct-java/pom.xml
+++ b/runners/direct-java/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/flink/examples/pom.xml
----------------------------------------------------------------------
diff --git a/runners/flink/examples/pom.xml b/runners/flink/examples/pom.xml
index b4d77dd..1d426bd 100644
--- a/runners/flink/examples/pom.xml
+++ b/runners/flink/examples/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-flink-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/flink/pom.xml
----------------------------------------------------------------------
diff --git a/runners/flink/pom.xml b/runners/flink/pom.xml
index 7af4842..0030f61 100644
--- a/runners/flink/pom.xml
+++ b/runners/flink/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/flink/runner/pom.xml
----------------------------------------------------------------------
diff --git a/runners/flink/runner/pom.xml b/runners/flink/runner/pom.xml
index 852a173..13d5b10 100644
--- a/runners/flink/runner/pom.xml
+++ b/runners/flink/runner/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-flink-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/google-cloud-dataflow-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/pom.xml b/runners/google-cloud-dataflow-java/pom.xml
index 73bee16..14f27b2 100644
--- a/runners/google-cloud-dataflow-java/pom.xml
+++ b/runners/google-cloud-dataflow-java/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/pom.xml
----------------------------------------------------------------------
diff --git a/runners/pom.xml b/runners/pom.xml
index f6379d0..3f74f7b 100644
--- a/runners/pom.xml
+++ b/runners/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/runners/spark/pom.xml
----------------------------------------------------------------------
diff --git a/runners/spark/pom.xml b/runners/spark/pom.xml
index f58f439..0d4c413 100644
--- a/runners/spark/pom.xml
+++ b/runners/spark/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/common/fn-api/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/common/fn-api/pom.xml b/sdks/common/fn-api/pom.xml
index 6ada6af..5a41d9e 100644
--- a/sdks/common/fn-api/pom.xml
+++ b/sdks/common/fn-api/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-common-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/common/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/common/pom.xml b/sdks/common/pom.xml
index 0c5e4d2..55db181 100644
--- a/sdks/common/pom.xml
+++ b/sdks/common/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/common/runner-api/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/common/runner-api/pom.xml b/sdks/common/runner-api/pom.xml
index aadc560..9c6de1e 100644
--- a/sdks/common/runner-api/pom.xml
+++ b/sdks/common/runner-api/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-common-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/build-tools/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/build-tools/pom.xml b/sdks/java/build-tools/pom.xml
index 9a55048..545f394 100644
--- a/sdks/java/build-tools/pom.xml
+++ b/sdks/java/build-tools/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/core/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/core/pom.xml b/sdks/java/core/pom.xml
index 93c26ad..4f89550 100644
--- a/sdks/java/core/pom.xml
+++ b/sdks/java/core/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/extensions/jackson/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/jackson/pom.xml b/sdks/java/extensions/jackson/pom.xml
index db2779a..1dfbd72 100644
--- a/sdks/java/extensions/jackson/pom.xml
+++ b/sdks/java/extensions/jackson/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-extensions-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/extensions/join-library/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/join-library/pom.xml b/sdks/java/extensions/join-library/pom.xml
index c5b6324..728c6f7 100644
--- a/sdks/java/extensions/join-library/pom.xml
+++ b/sdks/java/extensions/join-library/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-extensions-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/extensions/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/pom.xml b/sdks/java/extensions/pom.xml
index 721495f..26d92de 100644
--- a/sdks/java/extensions/pom.xml
+++ b/sdks/java/extensions/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/extensions/sorter/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sorter/pom.xml b/sdks/java/extensions/sorter/pom.xml
index a6a4cdd..9d03ba2 100644
--- a/sdks/java/extensions/sorter/pom.xml
+++ b/sdks/java/extensions/sorter/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-extensions-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/harness/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/harness/pom.xml b/sdks/java/harness/pom.xml
index 4a7fa0a..80b01ca 100644
--- a/sdks/java/harness/pom.xml
+++ b/sdks/java/harness/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/elasticsearch/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/elasticsearch/pom.xml b/sdks/java/io/elasticsearch/pom.xml
index 9b27cb9..3279dfd 100644
--- a/sdks/java/io/elasticsearch/pom.xml
+++ b/sdks/java/io/elasticsearch/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/google-cloud-platform/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/pom.xml b/sdks/java/io/google-cloud-platform/pom.xml
index 83d551a..66a4207 100644
--- a/sdks/java/io/google-cloud-platform/pom.xml
+++ b/sdks/java/io/google-cloud-platform/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/hadoop-common/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/hadoop-common/pom.xml b/sdks/java/io/hadoop-common/pom.xml
index 996d4ac..fcd984f 100644
--- a/sdks/java/io/hadoop-common/pom.xml
+++ b/sdks/java/io/hadoop-common/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/hbase/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/hbase/pom.xml b/sdks/java/io/hbase/pom.xml
index 4064c42..dfcca7a 100644
--- a/sdks/java/io/hbase/pom.xml
+++ b/sdks/java/io/hbase/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/hdfs/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/hdfs/pom.xml b/sdks/java/io/hdfs/pom.xml
index 9dc68e5..f3a1a27 100644
--- a/sdks/java/io/hdfs/pom.xml
+++ b/sdks/java/io/hdfs/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/jdbc/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/jdbc/pom.xml b/sdks/java/io/jdbc/pom.xml
index 87b0b27..fd5c52b 100644
--- a/sdks/java/io/jdbc/pom.xml
+++ b/sdks/java/io/jdbc/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/jms/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/jms/pom.xml b/sdks/java/io/jms/pom.xml
index d510583..cc7fb27 100644
--- a/sdks/java/io/jms/pom.xml
+++ b/sdks/java/io/jms/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/kafka/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/kafka/pom.xml b/sdks/java/io/kafka/pom.xml
index 8eaf4bf..d5ffe63 100644
--- a/sdks/java/io/kafka/pom.xml
+++ b/sdks/java/io/kafka/pom.xml
@@ -21,7 +21,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/kinesis/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/kinesis/pom.xml b/sdks/java/io/kinesis/pom.xml
index bef65a4..5138368 100644
--- a/sdks/java/io/kinesis/pom.xml
+++ b/sdks/java/io/kinesis/pom.xml
@@ -21,7 +21,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/mongodb/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/mongodb/pom.xml b/sdks/java/io/mongodb/pom.xml
index a469501..7deae14 100644
--- a/sdks/java/io/mongodb/pom.xml
+++ b/sdks/java/io/mongodb/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/mqtt/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/mqtt/pom.xml b/sdks/java/io/mqtt/pom.xml
index 04f5416..8b1d848 100644
--- a/sdks/java/io/mqtt/pom.xml
+++ b/sdks/java/io/mqtt/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/io/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/pom.xml b/sdks/java/io/pom.xml
index 22432b9..8d5b69b 100644
--- a/sdks/java/io/pom.xml
+++ b/sdks/java/io/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/java8tests/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/java8tests/pom.xml b/sdks/java/java8tests/pom.xml
index aee81ef..fda1d63 100644
--- a/sdks/java/java8tests/pom.xml
+++ b/sdks/java/java8tests/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/javadoc/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/javadoc/pom.xml b/sdks/java/javadoc/pom.xml
index 91c76fc..b785c98 100644
--- a/sdks/java/javadoc/pom.xml
+++ b/sdks/java/javadoc/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/maven-archetypes/examples-java8/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/examples-java8/pom.xml b/sdks/java/maven-archetypes/examples-java8/pom.xml
index 4603e11..2632d6d 100644
--- a/sdks/java/maven-archetypes/examples-java8/pom.xml
+++ b/sdks/java/maven-archetypes/examples-java8/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-maven-archetypes-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/maven-archetypes/examples/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/examples/pom.xml b/sdks/java/maven-archetypes/examples/pom.xml
index 674ff60..09e5428 100644
--- a/sdks/java/maven-archetypes/examples/pom.xml
+++ b/sdks/java/maven-archetypes/examples/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-maven-archetypes-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/maven-archetypes/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/pom.xml b/sdks/java/maven-archetypes/pom.xml
index 3d1b539..194e5bd 100644
--- a/sdks/java/maven-archetypes/pom.xml
+++ b/sdks/java/maven-archetypes/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/maven-archetypes/starter/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/starter/pom.xml b/sdks/java/maven-archetypes/starter/pom.xml
index 29df45e..092995a 100644
--- a/sdks/java/maven-archetypes/starter/pom.xml
+++ b/sdks/java/maven-archetypes/starter/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-maven-archetypes-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/java/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/pom.xml b/sdks/java/pom.xml
index 7d7153a..a09a6be 100644
--- a/sdks/java/pom.xml
+++ b/sdks/java/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/pom.xml b/sdks/pom.xml
index cf21743..f130816 100644
--- a/sdks/pom.xml
+++ b/sdks/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/beam/blob/8ab36fa2/sdks/python/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/python/pom.xml b/sdks/python/pom.xml
index 22d9d39..98b7fa3 100644
--- a/sdks/python/pom.xml
+++ b/sdks/python/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-parent</artifactId>
- <version>0.6.0</version>
+ <version>0.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
[9/9] beam git commit: This closes #2202
Posted by al...@apache.org.
This closes #2202
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/b25a0369
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/b25a0369
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/b25a0369
Branch: refs/heads/release-0.6.0
Commit: b25a0369dc5e5e4eacdc6e40b1e81452f9685579
Parents: 1c91d0e edaf3ac
Author: Ahmet Altay <al...@google.com>
Authored: Wed Mar 8 14:07:45 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 14:07:45 2017 -0800
----------------------------------------------------------------------
examples/java/pom.xml | 5 +-
examples/java8/pom.xml | 2 +-
examples/pom.xml | 2 +-
pom.xml | 4 +-
runners/apex/pom.xml | 2 +-
runners/core-construction-java/pom.xml | 5 +-
runners/core-java/pom.xml | 5 +-
runners/direct-java/pom.xml | 5 +-
runners/flink/examples/pom.xml | 2 +-
runners/flink/pom.xml | 2 +-
runners/flink/runner/pom.xml | 2 +-
runners/google-cloud-dataflow-java/pom.xml | 7 +-
.../dataflow/BatchStatefulParDoOverrides.java | 283 +++++++++++++++++++
.../dataflow/DataflowPipelineTranslator.java | 5 +-
.../beam/runners/dataflow/DataflowRunner.java | 6 +
.../beam/runners/dataflow/util/DoFnInfo.java | 9 +
.../BatchStatefulParDoOverridesTest.java | 169 +++++++++++
.../DataflowPipelineTranslatorTest.java | 72 +++++
runners/pom.xml | 2 +-
runners/spark/pom.xml | 2 +-
sdks/common/fn-api/pom.xml | 2 +-
sdks/common/pom.xml | 2 +-
sdks/common/runner-api/pom.xml | 2 +-
sdks/java/build-tools/pom.xml | 2 +-
sdks/java/core/pom.xml | 5 +-
.../apache/beam/sdk/transforms/ParDoTest.java | 117 ++++++++
sdks/java/extensions/jackson/pom.xml | 7 +-
sdks/java/extensions/join-library/pom.xml | 2 +-
sdks/java/extensions/pom.xml | 2 +-
sdks/java/extensions/sorter/pom.xml | 5 +-
sdks/java/harness/pom.xml | 2 +-
sdks/java/io/elasticsearch/pom.xml | 9 +-
.../sdk/io/elasticsearch/ElasticsearchIO.java | 102 ++++---
sdks/java/io/google-cloud-platform/pom.xml | 2 +-
sdks/java/io/hadoop-common/pom.xml | 2 +-
sdks/java/io/hbase/pom.xml | 5 +-
sdks/java/io/hdfs/pom.xml | 5 +-
sdks/java/io/jdbc/pom.xml | 2 +-
sdks/java/io/jms/pom.xml | 2 +-
sdks/java/io/kafka/pom.xml | 5 +-
sdks/java/io/kinesis/pom.xml | 2 +-
sdks/java/io/mongodb/pom.xml | 2 +-
sdks/java/io/mqtt/pom.xml | 2 +-
sdks/java/io/pom.xml | 2 +-
sdks/java/java8tests/pom.xml | 2 +-
sdks/java/javadoc/pom.xml | 2 +-
.../maven-archetypes/examples-java8/pom.xml | 2 +-
.../main/resources/archetype-resources/pom.xml | 11 +-
sdks/java/maven-archetypes/examples/pom.xml | 2 +-
.../main/resources/archetype-resources/pom.xml | 11 +-
sdks/java/maven-archetypes/pom.xml | 2 +-
sdks/java/maven-archetypes/starter/pom.xml | 2 +-
sdks/java/pom.xml | 2 +-
sdks/pom.xml | 2 +-
sdks/python/apache_beam/io/range_trackers.py | 2 +-
sdks/python/apache_beam/transforms/trigger.py | 2 +-
sdks/python/apache_beam/typehints/decorators.py | 5 +-
sdks/python/apache_beam/typehints/typehints.py | 3 +-
.../apache_beam/utils/pipeline_options.py | 2 +-
sdks/python/apache_beam/version.py | 2 +-
sdks/python/pom.xml | 2 +-
sdks/python/setup.py | 1 -
62 files changed, 807 insertions(+), 131 deletions(-)
----------------------------------------------------------------------
[2/9] beam git commit: Change Json parsing from gson to jackson for
ElasticsearchIO
Posted by al...@apache.org.
Change Json parsing from gson to jackson for ElasticsearchIO
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/d5257658
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/d5257658
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/d5257658
Branch: refs/heads/release-0.6.0
Commit: d5257658f094fe8c2a8668027bbdd4a26396ba0b
Parents: 8ab36fa
Author: Isma�l Mej�a <ie...@gmail.com>
Authored: Mon Mar 6 09:13:31 2017 +0100
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:38:39 2017 -0800
----------------------------------------------------------------------
sdks/java/io/elasticsearch/pom.xml | 7 +-
.../sdk/io/elasticsearch/ElasticsearchIO.java | 102 +++++++++----------
2 files changed, 52 insertions(+), 57 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/d5257658/sdks/java/io/elasticsearch/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/elasticsearch/pom.xml b/sdks/java/io/elasticsearch/pom.xml
index 3279dfd..5ea4452 100644
--- a/sdks/java/io/elasticsearch/pom.xml
+++ b/sdks/java/io/elasticsearch/pom.xml
@@ -47,9 +47,8 @@
</dependency>
<dependency>
- <groupId>com.google.code.gson</groupId>
- <artifactId>gson</artifactId>
- <version>2.6.2</version>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
@@ -116,7 +115,7 @@
</dependency>
<dependency>
- <groupId>org.apache.commons</groupId>
+ <groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>1.3.2</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/beam/blob/d5257658/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java b/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
index 5073834..b08cb24 100644
--- a/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
+++ b/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
@@ -20,17 +20,13 @@ package org.apache.beam.sdk.io.elasticsearch;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
-import com.google.gson.Gson;
-import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
import java.io.Serializable;
import java.net.MalformedURLException;
import java.net.URL;
@@ -38,11 +34,11 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.NoSuchElementException;
-import java.util.Set;
import javax.annotation.Nullable;
import org.apache.beam.sdk.coders.Coder;
@@ -140,11 +136,10 @@ public class ElasticsearchIO {
private ElasticsearchIO() {}
- private static JsonObject parseResponse(Response response) throws IOException {
- InputStream content = response.getEntity().getContent();
- InputStreamReader inputStreamReader = new InputStreamReader(content, "UTF-8");
- JsonObject jsonObject = new Gson().fromJson(inputStreamReader, JsonObject.class);
- return jsonObject;
+ private static final ObjectMapper mapper = new ObjectMapper();
+
+ private static JsonNode parseResponse(Response response) throws IOException {
+ return mapper.readValue(response.getEntity().getContent(), JsonNode.class);
}
/** A POJO describing a connection configuration to Elasticsearch. */
@@ -428,23 +423,24 @@ public class ElasticsearchIO {
// But, as each shard (replica or primary) is responsible for only one part of the data,
// there will be no duplicate.
- JsonObject statsJson = getStats(true);
- JsonObject shardsJson =
+ JsonNode statsJson = getStats(true);
+ JsonNode shardsJson =
statsJson
- .getAsJsonObject("indices")
- .getAsJsonObject(spec.getConnectionConfiguration().getIndex())
- .getAsJsonObject("shards");
- Set<Map.Entry<String, JsonElement>> shards = shardsJson.entrySet();
- for (Map.Entry<String, JsonElement> shardJson : shards) {
+ .path("indices")
+ .path(spec.getConnectionConfiguration().getIndex())
+ .path("shards");
+
+ Iterator<Map.Entry<String, JsonNode>> shards = shardsJson.fields();
+ while (shards.hasNext()) {
+ Map.Entry<String, JsonNode> shardJson = shards.next();
String shardId = shardJson.getKey();
- JsonArray value = (JsonArray) shardJson.getValue();
+ JsonNode value = (JsonNode) shardJson.getValue();
boolean isPrimaryShard =
value
- .get(0)
- .getAsJsonObject()
- .getAsJsonObject("routing")
- .getAsJsonPrimitive("primary")
- .getAsBoolean();
+ .path(0)
+ .path("routing")
+ .path("primary")
+ .asBoolean();
if (isPrimaryShard) {
sources.add(new BoundedElasticsearchSource(spec, shardId));
}
@@ -463,14 +459,14 @@ public class ElasticsearchIO {
// NB: Elasticsearch 5.x now provides the slice API.
// (https://www.elastic.co/guide/en/elasticsearch/reference/5.0/search-request-scroll.html
// #sliced-scroll)
- JsonObject statsJson = getStats(false);
- JsonObject indexStats =
+ JsonNode statsJson = getStats(false);
+ JsonNode indexStats =
statsJson
- .getAsJsonObject("indices")
- .getAsJsonObject(spec.getConnectionConfiguration().getIndex())
- .getAsJsonObject("primaries");
- JsonObject store = indexStats.getAsJsonObject("store");
- return store.getAsJsonPrimitive("size_in_bytes").getAsLong();
+ .path("indices")
+ .path(spec.getConnectionConfiguration().getIndex())
+ .path("primaries");
+ JsonNode store = indexStats.path("store");
+ return store.path("size_in_bytes").asLong();
}
@Override
@@ -494,7 +490,7 @@ public class ElasticsearchIO {
return StringUtf8Coder.of();
}
- private JsonObject getStats(boolean shardLevel) throws IOException {
+ private JsonNode getStats(boolean shardLevel) throws IOException {
HashMap<String, String> params = new HashMap<>();
if (shardLevel) {
params.put("level", "shards");
@@ -544,13 +540,13 @@ public class ElasticsearchIO {
HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
response =
restClient.performRequest("GET", endPoint, params, queryEntity, new BasicHeader("", ""));
- JsonObject searchResult = parseResponse(response);
+ JsonNode searchResult = parseResponse(response);
updateScrollId(searchResult);
return readNextBatchAndReturnFirstDocument(searchResult);
}
- private void updateScrollId(JsonObject searchResult) {
- scrollId = searchResult.getAsJsonPrimitive("_scroll_id").getAsString();
+ private void updateScrollId(JsonNode searchResult) {
+ scrollId = searchResult.path("_scroll_id").asText();
}
@Override
@@ -571,15 +567,15 @@ public class ElasticsearchIO {
Collections.<String, String>emptyMap(),
scrollEntity,
new BasicHeader("", ""));
- JsonObject searchResult = parseResponse(response);
+ JsonNode searchResult = parseResponse(response);
updateScrollId(searchResult);
return readNextBatchAndReturnFirstDocument(searchResult);
}
}
- private boolean readNextBatchAndReturnFirstDocument(JsonObject searchResult) {
+ private boolean readNextBatchAndReturnFirstDocument(JsonNode searchResult) {
//stop if no more data
- JsonArray hits = searchResult.getAsJsonObject("hits").getAsJsonArray("hits");
+ JsonNode hits = searchResult.path("hits").path("hits");
if (hits.size() == 0) {
current = null;
batchIterator = null;
@@ -587,8 +583,8 @@ public class ElasticsearchIO {
}
// list behind iterator is empty
List<String> batch = new ArrayList<>();
- for (JsonElement hit : hits) {
- String document = hit.getAsJsonObject().getAsJsonObject("_source").toString();
+ for (JsonNode hit : hits) {
+ String document = hit.path("_source").toString();
batch.add(document);
}
batchIterator = batch.listIterator();
@@ -780,26 +776,26 @@ public class ElasticsearchIO {
Collections.<String, String>emptyMap(),
requestBody,
new BasicHeader("", ""));
- JsonObject searchResult = parseResponse(response);
- boolean errors = searchResult.getAsJsonPrimitive("errors").getAsBoolean();
+ JsonNode searchResult = parseResponse(response);
+ boolean errors = searchResult.path("errors").asBoolean();
if (errors) {
StringBuilder errorMessages =
new StringBuilder(
"Error writing to Elasticsearch, some elements could not be inserted:");
- JsonArray items = searchResult.getAsJsonArray("items");
+ JsonNode items = searchResult.path("items");
//some items present in bulk might have errors, concatenate error messages
- for (JsonElement item : items) {
- JsonObject creationObject = item.getAsJsonObject().getAsJsonObject("create");
- JsonObject error = creationObject.getAsJsonObject("error");
+ for (JsonNode item : items) {
+ JsonNode creationObject = item.path("create");
+ JsonNode error = creationObject.get("error");
if (error != null) {
- String type = error.getAsJsonPrimitive("type").getAsString();
- String reason = error.getAsJsonPrimitive("reason").getAsString();
- String docId = creationObject.getAsJsonPrimitive("_id").getAsString();
+ String type = error.path("type").asText();
+ String reason = error.path("reason").asText();
+ String docId = creationObject.path("_id").asText();
errorMessages.append(String.format("%nDocument id %s: %s (%s)", docId, reason, type));
- JsonObject causedBy = error.getAsJsonObject("caused_by");
+ JsonNode causedBy = error.get("caused_by");
if (causedBy != null) {
- String cbReason = causedBy.getAsJsonPrimitive("reason").getAsString();
- String cbType = causedBy.getAsJsonPrimitive("type").getAsString();
+ String cbReason = causedBy.path("reason").asText();
+ String cbType = causedBy.path("type").asText();
errorMessages.append(String.format("%nCaused by: %s (%s)", cbReason, cbType));
}
}
[4/9] beam git commit: Add ServicesResourceTransformer to all shading
configuration
Posted by al...@apache.org.
Add ServicesResourceTransformer to all shading configuration
This ensures that files in META-INF/services aren't overwritten. Instead, they are concatenated.
This is critical to ensure PipelineOptionsRegistrar, RunnerRegistrar, IOChannelFactoryRegistrar
and FileSystemRegistrar work well for users.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/92c5b5bd
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/92c5b5bd
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/92c5b5bd
Branch: refs/heads/release-0.6.0
Commit: 92c5b5bd732d9fc019fa6820afcc31a92a026bbf
Parents: 730def7
Author: Davor Bonaci <da...@google.com>
Authored: Tue Mar 7 11:57:38 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:40:17 2017 -0800
----------------------------------------------------------------------
examples/java/pom.xml | 3 +++
runners/core-construction-java/pom.xml | 3 +++
runners/core-java/pom.xml | 3 +++
runners/direct-java/pom.xml | 3 +++
runners/google-cloud-dataflow-java/pom.xml | 3 +++
sdks/java/core/pom.xml | 3 +++
sdks/java/extensions/jackson/pom.xml | 5 ++++-
sdks/java/extensions/sorter/pom.xml | 3 +++
sdks/java/io/hbase/pom.xml | 3 +++
sdks/java/io/hdfs/pom.xml | 3 +++
sdks/java/io/kafka/pom.xml | 3 +++
.../src/main/resources/archetype-resources/pom.xml | 11 +++--------
.../src/main/resources/archetype-resources/pom.xml | 11 +++--------
13 files changed, 40 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/examples/java/pom.xml
----------------------------------------------------------------------
diff --git a/examples/java/pom.xml b/examples/java/pom.xml
index 9da814b..f8cd208 100644
--- a/examples/java/pom.xml
+++ b/examples/java/pom.xml
@@ -383,6 +383,9 @@
</excludes>
</filter>
</filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/runners/core-construction-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/pom.xml b/runners/core-construction-java/pom.xml
index b602f5d..bcbd750 100644
--- a/runners/core-construction-java/pom.xml
+++ b/runners/core-construction-java/pom.xml
@@ -95,6 +95,9 @@
</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/runners/core-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/core-java/pom.xml b/runners/core-java/pom.xml
index 28ab0ee..d9853c7 100644
--- a/runners/core-java/pom.xml
+++ b/runners/core-java/pom.xml
@@ -97,6 +97,9 @@
<shadedPattern>org.apache.beam.runners.core.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/runners/direct-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
index ced9cd6..4830e73 100644
--- a/runners/direct-java/pom.xml
+++ b/runners/direct-java/pom.xml
@@ -130,6 +130,9 @@
<shadedPattern>org.apache.beam.runners.direct.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/runners/google-cloud-dataflow-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/pom.xml b/runners/google-cloud-dataflow-java/pom.xml
index 14f27b2..33ebcfb 100644
--- a/runners/google-cloud-dataflow-java/pom.xml
+++ b/runners/google-cloud-dataflow-java/pom.xml
@@ -155,6 +155,9 @@
</excludes>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/core/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/core/pom.xml b/sdks/java/core/pom.xml
index 4f89550..553cafb 100644
--- a/sdks/java/core/pom.xml
+++ b/sdks/java/core/pom.xml
@@ -127,6 +127,9 @@
<shadedPattern>org.apache.beam.sdk.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/extensions/jackson/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/jackson/pom.xml b/sdks/java/extensions/jackson/pom.xml
index 1dfbd72..d9d4d48 100644
--- a/sdks/java/extensions/jackson/pom.xml
+++ b/sdks/java/extensions/jackson/pom.xml
@@ -60,6 +60,9 @@
<shadedPattern>org.apache.beam.sdk.extensions.jackson.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
@@ -119,4 +122,4 @@
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/extensions/sorter/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/sorter/pom.xml b/sdks/java/extensions/sorter/pom.xml
index 9d03ba2..a6edf53 100644
--- a/sdks/java/extensions/sorter/pom.xml
+++ b/sdks/java/extensions/sorter/pom.xml
@@ -92,6 +92,9 @@
<shadedPattern>org.apache.beam.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/io/hbase/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/hbase/pom.xml b/sdks/java/io/hbase/pom.xml
index dfcca7a..779c8de 100644
--- a/sdks/java/io/hbase/pom.xml
+++ b/sdks/java/io/hbase/pom.xml
@@ -72,6 +72,9 @@
<shadedPattern>org.apache.beam.sdk.io.hbase.repackaged.com.google.protobuf</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/io/hdfs/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/hdfs/pom.xml b/sdks/java/io/hdfs/pom.xml
index f3a1a27..254bfd2 100644
--- a/sdks/java/io/hdfs/pom.xml
+++ b/sdks/java/io/hdfs/pom.xml
@@ -58,6 +58,9 @@
<shadedPattern>org.apache.beam.sdk.io.hdfs.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/io/kafka/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/kafka/pom.xml b/sdks/java/io/kafka/pom.xml
index d5ffe63..2250cb4 100644
--- a/sdks/java/io/kafka/pom.xml
+++ b/sdks/java/io/kafka/pom.xml
@@ -69,6 +69,9 @@
<shadedPattern>org.apache.beam.sdk.io.kafka.repackaged.com.google.thirdparty</shadedPattern>
</relocation>
</relocations>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/maven-archetypes/examples-java8/src/main/resources/archetype-resources/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/examples-java8/src/main/resources/archetype-resources/pom.xml b/sdks/java/maven-archetypes/examples-java8/src/main/resources/archetype-resources/pom.xml
index 18e3d9d..3845ba0 100644
--- a/sdks/java/maven-archetypes/examples-java8/src/main/resources/archetype-resources/pom.xml
+++ b/sdks/java/maven-archetypes/examples-java8/src/main/resources/archetype-resources/pom.xml
@@ -101,6 +101,9 @@
</excludes>
</filter>
</filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
@@ -192,14 +195,6 @@
<scope>runtime</scope>
</dependency>
</dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
</profile>
<profile>
http://git-wip-us.apache.org/repos/asf/beam/blob/92c5b5bd/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/pom.xml b/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/pom.xml
index 3a30333..3e51f97 100644
--- a/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/pom.xml
+++ b/sdks/java/maven-archetypes/examples/src/main/resources/archetype-resources/pom.xml
@@ -101,6 +101,9 @@
</excludes>
</filter>
</filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
</configuration>
</execution>
</executions>
@@ -192,14 +195,6 @@
<scope>runtime</scope>
</dependency>
</dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
</profile>
<profile>
[5/9] beam git commit: Explicitly GBK before stateful ParDo in
Dataflow batch
Posted by al...@apache.org.
Explicitly GBK before stateful ParDo in Dataflow batch
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/5e2afa29
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/5e2afa29
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/5e2afa29
Branch: refs/heads/release-0.6.0
Commit: 5e2afa29a3a0fe93e662b2fe7173c1641c253cd5
Parents: 92c5b5b
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Mar 2 14:29:56 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:41:01 2017 -0800
----------------------------------------------------------------------
.../dataflow/BatchStatefulParDoOverrides.java | 283 +++++++++++++++++++
.../dataflow/DataflowPipelineTranslator.java | 5 +-
.../beam/runners/dataflow/DataflowRunner.java | 6 +
.../beam/runners/dataflow/util/DoFnInfo.java | 9 +
.../BatchStatefulParDoOverridesTest.java | 169 +++++++++++
.../DataflowPipelineTranslatorTest.java | 72 +++++
.../apache/beam/sdk/transforms/ParDoTest.java | 117 ++++++++
7 files changed, 660 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
new file mode 100644
index 0000000..91f84ab
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
@@ -0,0 +1,283 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.common.collect.Iterables;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.runners.core.construction.ReplacementOutputs;
+import org.apache.beam.runners.dataflow.BatchViewOverrides.GroupByKeyAndSortValuesOnly;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.ParDo.BoundMulti;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.WindowingStrategy;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TaggedPValue;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.joda.time.Instant;
+
+/**
+ * {@link PTransformOverrideFactory PTransformOverrideFactories} that expands to correctly implement
+ * stateful {@link ParDo} using window-unaware {@link GroupByKeyAndSortValuesOnly} to linearize
+ * processing per key.
+ *
+ * <p>This implementation relies on implementation details of the Dataflow runner, specifically
+ * standard fusion behavior of {@link ParDo} tranforms following a {@link GroupByKey}.
+ */
+public class BatchStatefulParDoOverrides {
+
+ /**
+ * Returns a {@link PTransformOverrideFactory} that replaces a single-output
+ * {@link ParDo} with a composite transform specialized for the {@link DataflowRunner}.
+ */
+ public static <K, InputT, OutputT>
+ PTransformOverrideFactory<
+ PCollection<KV<K, InputT>>, PCollection<OutputT>, ParDo.Bound<KV<K, InputT>, OutputT>>
+ singleOutputOverrideFactory() {
+ return new SingleOutputOverrideFactory<>();
+ }
+
+ /**
+ * Returns a {@link PTransformOverrideFactory} that replaces a multi-output
+ * {@link ParDo} with a composite transform specialized for the {@link DataflowRunner}.
+ */
+ public static <K, InputT, OutputT>
+ PTransformOverrideFactory<
+ PCollection<KV<K, InputT>>, PCollectionTuple,
+ ParDo.BoundMulti<KV<K, InputT>, OutputT>>
+ multiOutputOverrideFactory() {
+ return new MultiOutputOverrideFactory<>();
+ }
+
+ private static class SingleOutputOverrideFactory<K, InputT, OutputT>
+ implements PTransformOverrideFactory<
+ PCollection<KV<K, InputT>>, PCollection<OutputT>, ParDo.Bound<KV<K, InputT>, OutputT>> {
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public PTransform<PCollection<KV<K, InputT>>, PCollection<OutputT>> getReplacementTransform(
+ ParDo.Bound<KV<K, InputT>, OutputT> originalParDo) {
+ return new StatefulSingleOutputParDo<>(originalParDo);
+ }
+
+ @Override
+ public PCollection<KV<K, InputT>> getInput(List<TaggedPValue> inputs, Pipeline p) {
+ return (PCollection<KV<K, InputT>>) Iterables.getOnlyElement(inputs).getValue();
+ }
+
+ @Override
+ public Map<PValue, ReplacementOutput> mapOutputs(
+ List<TaggedPValue> outputs, PCollection<OutputT> newOutput) {
+ return ReplacementOutputs.singleton(outputs, newOutput);
+ }
+ }
+
+ private static class MultiOutputOverrideFactory<K, InputT, OutputT>
+ implements PTransformOverrideFactory<
+ PCollection<KV<K, InputT>>, PCollectionTuple, ParDo.BoundMulti<KV<K, InputT>, OutputT>> {
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public PTransform<PCollection<KV<K, InputT>>, PCollectionTuple> getReplacementTransform(
+ BoundMulti<KV<K, InputT>, OutputT> originalParDo) {
+ return new StatefulMultiOutputParDo<>(originalParDo);
+ }
+
+ @Override
+ public PCollection<KV<K, InputT>> getInput(List<TaggedPValue> inputs, Pipeline p) {
+ return (PCollection<KV<K, InputT>>) Iterables.getOnlyElement(inputs).getValue();
+ }
+
+ @Override
+ public Map<PValue, ReplacementOutput> mapOutputs(
+ List<TaggedPValue> outputs, PCollectionTuple newOutput) {
+ return ReplacementOutputs.tagged(outputs, newOutput);
+ }
+ }
+
+ static class StatefulSingleOutputParDo<K, InputT, OutputT>
+ extends PTransform<PCollection<KV<K, InputT>>, PCollection<OutputT>> {
+
+ private final ParDo.Bound<KV<K, InputT>, OutputT> originalParDo;
+
+ StatefulSingleOutputParDo(ParDo.Bound<KV<K, InputT>, OutputT> originalParDo) {
+ this.originalParDo = originalParDo;
+ }
+
+ ParDo.Bound<KV<K, InputT>, OutputT> getOriginalParDo() {
+ return originalParDo;
+ }
+
+ @Override
+ public PCollection<OutputT> expand(PCollection<KV<K, InputT>> input) {
+ DoFn<KV<K, InputT>, OutputT> fn = originalParDo.getFn();
+ verifyFnIsStateful(fn);
+
+ PTransform<
+ PCollection<? extends KV<K, Iterable<KV<Instant, WindowedValue<KV<K, InputT>>>>>>,
+ PCollection<OutputT>>
+ statefulParDo =
+ ParDo.of(new BatchStatefulDoFn<>(fn)).withSideInputs(originalParDo.getSideInputs());
+
+ return input.apply(new GbkBeforeStatefulParDo<K, InputT>()).apply(statefulParDo);
+ }
+ }
+
+ static class StatefulMultiOutputParDo<K, InputT, OutputT>
+ extends PTransform<PCollection<KV<K, InputT>>, PCollectionTuple> {
+
+ private final BoundMulti<KV<K, InputT>, OutputT> originalParDo;
+
+ StatefulMultiOutputParDo(ParDo.BoundMulti<KV<K, InputT>, OutputT> originalParDo) {
+ this.originalParDo = originalParDo;
+ }
+
+ @Override
+ public PCollectionTuple expand(PCollection<KV<K, InputT>> input) {
+ DoFn<KV<K, InputT>, OutputT> fn = originalParDo.getFn();
+ verifyFnIsStateful(fn);
+
+ PTransform<
+ PCollection<? extends KV<K, Iterable<KV<Instant, WindowedValue<KV<K, InputT>>>>>>,
+ PCollectionTuple>
+ statefulParDo =
+ ParDo.of(new BatchStatefulDoFn<K, InputT, OutputT>(fn))
+ .withSideInputs(originalParDo.getSideInputs())
+ .withOutputTags(
+ originalParDo.getMainOutputTag(), originalParDo.getSideOutputTags());
+
+ return input.apply(new GbkBeforeStatefulParDo<K, InputT>()).apply(statefulParDo);
+ }
+
+ public BoundMulti<KV<K, InputT>, OutputT> getOriginalParDo() {
+ return originalParDo;
+ }
+ }
+
+ static class GbkBeforeStatefulParDo<K, V>
+ extends PTransform<
+ PCollection<KV<K, V>>,
+ PCollection<KV<K, Iterable<KV<Instant, WindowedValue<KV<K, V>>>>>>> {
+
+ @Override
+ public PCollection<KV<K, Iterable<KV<Instant, WindowedValue<KV<K, V>>>>>> expand(
+ PCollection<KV<K, V>> input) {
+
+ WindowingStrategy<?, ?> inputWindowingStrategy = input.getWindowingStrategy();
+
+ // A KvCoder is required since this goes through GBK. Further, WindowedValueCoder
+ // is not registered by default, so we explicitly set the relevant coders.
+ checkState(
+ input.getCoder() instanceof KvCoder,
+ "Input to a %s using state requires a %s, but the coder was %s",
+ ParDo.class.getSimpleName(),
+ KvCoder.class.getSimpleName(),
+ input.getCoder());
+ KvCoder<K, V> kvCoder = (KvCoder<K, V>) input.getCoder();
+ Coder<K> keyCoder = kvCoder.getKeyCoder();
+ Coder<? extends BoundedWindow> windowCoder =
+ inputWindowingStrategy.getWindowFn().windowCoder();
+
+ return input
+ // Stash the original timestamps, etc, for when it is fed to the user's DoFn
+ .apply("ReifyWindows", ParDo.of(new ReifyWindowedValueFn<K, V>()))
+ .setCoder(
+ KvCoder.of(
+ keyCoder,
+ KvCoder.of(InstantCoder.of(), WindowedValue.getFullCoder(kvCoder, windowCoder))))
+
+ // Group by key and sort by timestamp, dropping windows as they are reified
+ .apply(
+ "PartitionKeys",
+ new GroupByKeyAndSortValuesOnly<K, Instant, WindowedValue<KV<K, V>>>())
+
+ // The GBKO sets the windowing strategy to the global default
+ .setWindowingStrategyInternal(inputWindowingStrategy);
+ }
+ }
+
+ /** A key-preserving {@link DoFn} that reifies a windowed value. */
+ static class ReifyWindowedValueFn<K, V>
+ extends DoFn<KV<K, V>, KV<K, KV<Instant, WindowedValue<KV<K, V>>>>> {
+ @ProcessElement
+ public void processElement(final ProcessContext c, final BoundedWindow window) {
+ c.output(
+ KV.of(
+ c.element().getKey(),
+ KV.of(
+ c.timestamp(), WindowedValue.of(c.element(), c.timestamp(), window, c.pane()))));
+ }
+ }
+
+ /**
+ * A key-preserving {@link DoFn} that explodes an iterable that has been grouped by key and
+ * window.
+ */
+ public static class BatchStatefulDoFn<K, V, OutputT>
+ extends DoFn<KV<K, Iterable<KV<Instant, WindowedValue<KV<K, V>>>>>, OutputT> {
+
+ private final DoFn<KV<K, V>, OutputT> underlyingDoFn;
+
+ BatchStatefulDoFn(DoFn<KV<K, V>, OutputT> underlyingDoFn) {
+ this.underlyingDoFn = underlyingDoFn;
+ }
+
+ public DoFn<KV<K, V>, OutputT> getUnderlyingDoFn() {
+ return underlyingDoFn;
+ }
+
+ @ProcessElement
+ public void processElement(final ProcessContext c, final BoundedWindow window) {
+ throw new UnsupportedOperationException(
+ "BatchStatefulDoFn.ProcessElement should never be invoked");
+ }
+
+ @Override
+ public TypeDescriptor<OutputT> getOutputTypeDescriptor() {
+ return underlyingDoFn.getOutputTypeDescriptor();
+ }
+ }
+
+ private static <InputT, OutputT> void verifyFnIsStateful(DoFn<InputT, OutputT> fn) {
+ DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());
+
+ // It is still correct to use this without state or timers, but a bad idea.
+ // Since it is internal it should never be used wrong, so it is OK to crash.
+ checkState(
+ signature.usesState() || signature.usesTimers(),
+ "%s used for %s that does not use state or timers.",
+ BatchStatefulParDoOverrides.class.getSimpleName(),
+ ParDo.class.getSimpleName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
index 7e559e9..fc47593 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
@@ -956,7 +956,10 @@ public class DataflowPipelineTranslator {
DoFnInfo.forFn(
fn, windowingStrategy, sideInputs, inputCoder, mainOutput, outputMap))));
- if (signature.usesState() || signature.usesTimers()) {
+ // Setting USES_KEYED_STATE will cause an ungrouped shuffle, which works
+ // in streaming but does not work in batch
+ if (context.getPipelineOptions().isStreaming()
+ && (signature.usesState() || signature.usesTimers())) {
stepContext.addInput(PropertyNames.USES_KEYED_STATE, "true");
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/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 15147f1..b782786 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
@@ -341,6 +341,12 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
UnsupportedOverrideFactory.withMessage(
"The DataflowRunner in batch mode does not support Read.Unbounded"));
ptoverrides
+ // State and timer pardos are implemented by expansion to GBK-then-ParDo
+ .put(PTransformMatchers.stateOrTimerParDoMulti(),
+ BatchStatefulParDoOverrides.multiOutputOverrideFactory())
+ .put(PTransformMatchers.stateOrTimerParDoSingle(),
+ BatchStatefulParDoOverrides.singleOutputOverrideFactory())
+
// Write uses views internally
.put(PTransformMatchers.classEqualTo(Write.class), new BatchWriteFactory(this))
.put(
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DoFnInfo.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DoFnInfo.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DoFnInfo.java
index 4d80a39..55c62ae 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DoFnInfo.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DoFnInfo.java
@@ -72,6 +72,15 @@ public class DoFnInfo<InputT, OutputT> implements Serializable {
outputMap);
}
+ public DoFnInfo<InputT, OutputT> withFn(DoFn<InputT, OutputT> newFn) {
+ return DoFnInfo.forFn(newFn,
+ windowingStrategy,
+ sideInputViews,
+ inputCoder,
+ mainOutput,
+ outputMap);
+ }
+
private DoFnInfo(
DoFn<InputT, OutputT> doFn,
WindowingStrategy<?, ?> windowingStrategy,
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java
new file mode 100644
index 0000000..ef3e414
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverridesTest.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.beam.runners.dataflow;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.LinkedList;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.dataflow.BatchStatefulParDoOverrides.StatefulMultiOutputParDo;
+import org.apache.beam.runners.dataflow.BatchStatefulParDoOverrides.StatefulSingleOutputParDo;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.Pipeline.PipelineVisitor;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.runners.TransformHierarchy.Node;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.util.GcsUtil;
+import org.apache.beam.sdk.util.TestCredential;
+import org.apache.beam.sdk.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.util.state.StateSpec;
+import org.apache.beam.sdk.util.state.StateSpecs;
+import org.apache.beam.sdk.util.state.ValueState;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/** Tests for {@link BatchStatefulParDoOverrides}. */
+@RunWith(JUnit4.class)
+public class BatchStatefulParDoOverridesTest implements Serializable {
+
+ @Test
+ public void testSingleOutputOverrideNonCrashing() throws Exception {
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline pipeline = Pipeline.create(options);
+
+ DummyStatefulDoFn fn = new DummyStatefulDoFn();
+ pipeline.apply(Create.of(KV.of(1, 2))).apply(ParDo.of(fn));
+
+ DataflowRunner runner = (DataflowRunner) pipeline.getRunner();
+ runner.replaceTransforms(pipeline);
+ assertThat(findBatchStatefulDoFn(pipeline), equalTo((DoFn) fn));
+ }
+
+ @Test
+ public void testMultiOutputOverrideNonCrashing() throws Exception {
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline pipeline = Pipeline.create(options);
+
+ TupleTag<Integer> mainOutputTag = new TupleTag<Integer>() {};
+
+ DummyStatefulDoFn fn = new DummyStatefulDoFn();
+ pipeline
+ .apply(Create.of(KV.of(1, 2)))
+ .apply(ParDo.withOutputTags(mainOutputTag, TupleTagList.empty()).of(fn));
+
+ DataflowRunner runner = (DataflowRunner) pipeline.getRunner();
+ runner.replaceTransforms(pipeline);
+ assertThat(findBatchStatefulDoFn(pipeline), equalTo((DoFn) fn));
+ }
+
+ private static DummyStatefulDoFn findBatchStatefulDoFn(Pipeline p) {
+ FindBatchStatefulDoFnVisitor findBatchStatefulDoFnVisitor = new FindBatchStatefulDoFnVisitor();
+ p.traverseTopologically(findBatchStatefulDoFnVisitor);
+ return (DummyStatefulDoFn) findBatchStatefulDoFnVisitor.getStatefulDoFn();
+ }
+
+ private static class DummyStatefulDoFn extends DoFn<KV<Integer, Integer>, Integer> {
+
+ @StateId("foo")
+ private final StateSpec<Object, ValueState<Integer>> spec = StateSpecs.value(VarIntCoder.of());
+
+ @ProcessElement
+ public void processElem(ProcessContext c) {
+ // noop
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ return other instanceof DummyStatefulDoFn;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+ }
+
+ private static class FindBatchStatefulDoFnVisitor extends PipelineVisitor.Defaults {
+
+ @Nullable private DoFn<?, ?> batchStatefulDoFn;
+
+ public DoFn<?, ?> getStatefulDoFn() {
+ assertThat(batchStatefulDoFn, not(nullValue()));
+ return batchStatefulDoFn;
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(Node node) {
+ if (node.getTransform() instanceof StatefulSingleOutputParDo) {
+ batchStatefulDoFn =
+ ((StatefulSingleOutputParDo) node.getTransform()).getOriginalParDo().getFn();
+ return CompositeBehavior.DO_NOT_ENTER_TRANSFORM;
+ } else if (node.getTransform() instanceof StatefulMultiOutputParDo) {
+ batchStatefulDoFn =
+ ((StatefulMultiOutputParDo) node.getTransform()).getOriginalParDo().getFn();
+ return CompositeBehavior.DO_NOT_ENTER_TRANSFORM;
+ } else {
+ return CompositeBehavior.ENTER_TRANSFORM;
+ }
+ }
+ }
+
+ private static DataflowPipelineOptions buildPipelineOptions() throws IOException {
+ GcsUtil mockGcsUtil = mock(GcsUtil.class);
+ when(mockGcsUtil.expand(any(GcsPath.class))).then(new Answer<List<GcsPath>>() {
+ @Override
+ public List<GcsPath> answer(InvocationOnMock invocation) throws Throwable {
+ return ImmutableList.of((GcsPath) invocation.getArguments()[0]);
+ }
+ });
+ when(mockGcsUtil.bucketAccessible(any(GcsPath.class))).thenReturn(true);
+
+ DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ options.setRunner(DataflowRunner.class);
+ options.setGcpCredential(new TestCredential());
+ options.setJobName("some-job-name");
+ options.setProject("some-project");
+ options.setTempLocation(GcsPath.fromComponents("somebucket", "some/path").toString());
+ options.setFilesToStage(new LinkedList<String>());
+ options.setGcsUtil(mockGcsUtil);
+ return options;
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java
index d4271e5..660e92e 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java
@@ -22,8 +22,10 @@ import static org.apache.beam.sdk.util.Structs.getDictionary;
import static org.apache.beam.sdk.util.Structs.getString;
import static org.hamcrest.Matchers.allOf;
import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasEntry;
import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.not;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
@@ -81,10 +83,15 @@ import org.apache.beam.sdk.util.Structs;
import org.apache.beam.sdk.util.TestCredential;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.util.state.StateSpec;
+import org.apache.beam.sdk.util.state.StateSpecs;
+import org.apache.beam.sdk.util.state.ValueState;
+import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PDone;
import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -825,6 +832,71 @@ public class DataflowPipelineTranslatorTest implements Serializable {
assertEquals("CollectionToSingleton", collectionToSingletonStep.getKind());
}
+ /**
+ * Smoke test to fail fast if translation of a stateful ParDo
+ * in batch breaks.
+ */
+ @Test
+ public void testBatchStatefulParDoTranslation() throws Exception {
+ DataflowPipelineOptions options = buildPipelineOptions();
+ DataflowRunner runner = DataflowRunner.fromOptions(options);
+ options.setStreaming(false);
+ DataflowPipelineTranslator translator = DataflowPipelineTranslator.fromOptions(options);
+
+ Pipeline pipeline = Pipeline.create(options);
+
+ TupleTag<Integer> mainOutputTag = new TupleTag<Integer>() {};
+
+ pipeline
+ .apply(Create.of(KV.of(1, 1)))
+ .apply(
+ ParDo.withOutputTags(mainOutputTag, TupleTagList.empty()).of(
+ new DoFn<KV<Integer, Integer>, Integer>() {
+ @StateId("unused")
+ final StateSpec<Object, ValueState<Integer>> stateSpec =
+ StateSpecs.value(VarIntCoder.of());
+
+ @ProcessElement
+ public void process(ProcessContext c) {
+ // noop
+ }
+ }));
+
+ runner.replaceTransforms(pipeline);
+
+ Job job =
+ translator
+ .translate(
+ pipeline,
+ runner,
+ Collections.<DataflowPackage>emptyList())
+ .getJob();
+
+ // The job should look like:
+ // 0. ParallelRead (Create)
+ // 1. ParDo(ReifyWVs)
+ // 2. GroupByKeyAndSortValuesONly
+ // 3. A ParDo over grouped and sorted KVs that is executed via ungrouping service-side
+
+ List<Step> steps = job.getSteps();
+ assertEquals(4, steps.size());
+
+ Step createStep = steps.get(0);
+ assertEquals("ParallelRead", createStep.getKind());
+
+ Step reifyWindowedValueStep = steps.get(1);
+ assertEquals("ParallelDo", reifyWindowedValueStep.getKind());
+
+ Step gbkStep = steps.get(2);
+ assertEquals("GroupByKey", gbkStep.getKind());
+
+ Step statefulParDoStep = steps.get(3);
+ assertEquals("ParallelDo", statefulParDoStep.getKind());
+ assertThat(
+ (String) statefulParDoStep.getProperties().get(PropertyNames.USES_KEYED_STATE),
+ not(equalTo("true")));
+ }
+
@Test
public void testToSingletonTranslationWithIsmSideInput() throws Exception {
// A "change detector" test that makes sure the translation
http://git-wip-us.apache.org/repos/asf/beam/blob/5e2afa29/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
index 6db0af4..e58f78e 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
@@ -48,6 +48,7 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -1506,6 +1507,54 @@ public class ParDoTest implements Serializable {
@Test
@Category({RunnableOnService.class, UsesStatefulParDo.class})
+ public void testValueStateDedup() {
+ final String stateId = "foo";
+
+ DoFn<KV<Integer, Integer>, Integer> onePerKey =
+ new DoFn<KV<Integer, Integer>, Integer>() {
+
+ @StateId(stateId)
+ private final StateSpec<Object, ValueState<Integer>> seenSpec =
+ StateSpecs.value(VarIntCoder.of());
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext c, @StateId(stateId) ValueState<Integer> seenState) {
+ Integer seen = MoreObjects.firstNonNull(seenState.read(), 0);
+
+ if (seen == 0) {
+ seenState.write(seen + 1);
+ c.output(c.element().getValue());
+ }
+ }
+ };
+
+ int numKeys = 50;
+ // A big enough list that we can see some deduping
+ List<KV<Integer, Integer>> input = new ArrayList<>();
+
+ // The output should have no dupes
+ Set<Integer> expectedOutput = new HashSet<>();
+
+ for (int key = 0; key < numKeys; ++key) {
+ int output = 1000 + key;
+ expectedOutput.add(output);
+
+ for (int i = 0; i < 15; ++i) {
+ input.add(KV.of(key, output));
+ }
+ }
+
+ Collections.shuffle(input);
+
+ PCollection<Integer> output = pipeline.apply(Create.of(input)).apply(ParDo.of(onePerKey));
+
+ PAssert.that(output).containsInAnyOrder(expectedOutput);
+ pipeline.run();
+ }
+
+ @Test
+ @Category({RunnableOnService.class, UsesStatefulParDo.class})
public void testValueStateFixedWindows() {
final String stateId = "foo";
@@ -1936,6 +1985,74 @@ public class ParDoTest implements Serializable {
pipeline.run();
}
+ /**
+ * Tests that event time timers for multiple keys both fire. This particularly exercises
+ * implementations that may GC in ways not simply governed by the watermark.
+ */
+ @Test
+ @Category({RunnableOnService.class, UsesTimersInParDo.class})
+ public void testEventTimeTimerMultipleKeys() throws Exception {
+ final String timerId = "foo";
+ final String stateId = "sizzle";
+
+ final int offset = 5000;
+ final int timerOutput = 4093;
+
+ DoFn<KV<String, Integer>, KV<String, Integer>> fn =
+ new DoFn<KV<String, Integer>, KV<String, Integer>>() {
+
+ @TimerId(timerId)
+ private final TimerSpec spec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @StateId(stateId)
+ private final StateSpec<Object, ValueState<String>> stateSpec =
+ StateSpecs.value(StringUtf8Coder.of());
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext context,
+ @TimerId(timerId) Timer timer,
+ @StateId(stateId) ValueState<String> state,
+ BoundedWindow window) {
+ timer.set(window.maxTimestamp());
+ state.write(context.element().getKey());
+ context.output(
+ KV.of(context.element().getKey(), context.element().getValue() + offset));
+ }
+
+ @OnTimer(timerId)
+ public void onTimer(OnTimerContext context, @StateId(stateId) ValueState<String> state) {
+ context.output(KV.of(state.read(), timerOutput));
+ }
+ };
+
+ // Enough keys that we exercise interesting code paths
+ int numKeys = 50;
+ List<KV<String, Integer>> input = new ArrayList<>();
+ List<KV<String, Integer>> expectedOutput = new ArrayList<>();
+
+ for (Integer key = 0; key < numKeys; ++key) {
+ // Each key should have just one final output at GC time
+ expectedOutput.add(KV.of(key.toString(), timerOutput));
+
+ for (int i = 0; i < 15; ++i) {
+ // Each input should be output with the offset added
+ input.add(KV.of(key.toString(), i));
+ expectedOutput.add(KV.of(key.toString(), i + offset));
+ }
+ }
+
+ Collections.shuffle(input);
+
+ PCollection<KV<String, Integer>> output =
+ pipeline
+ .apply(
+ Create.of(input))
+ .apply(ParDo.of(fn));
+ PAssert.that(output).containsInAnyOrder(expectedOutput);
+ pipeline.run();
+ }
+
@Test
@Category({RunnableOnService.class, UsesTimersInParDo.class})
public void testAbsoluteProcessingTimeTimerRejected() throws Exception {
[3/9] beam git commit: [BEAM-1646] Remove duplicated bigquery
dependency
Posted by al...@apache.org.
[BEAM-1646] Remove duplicated bigquery dependency
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/730def77
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/730def77
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/730def77
Branch: refs/heads/release-0.6.0
Commit: 730def77ebf216295d953b13722f21185f674ccc
Parents: d525765
Author: Mark Liu <ma...@google.com>
Authored: Tue Mar 7 14:26:37 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:39:21 2017 -0800
----------------------------------------------------------------------
sdks/python/setup.py | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/730def77/sdks/python/setup.py
----------------------------------------------------------------------
diff --git a/sdks/python/setup.py b/sdks/python/setup.py
index cf210d9..8b3e468 100644
--- a/sdks/python/setup.py
+++ b/sdks/python/setup.py
@@ -87,7 +87,6 @@ REQUIRED_PACKAGES = [
'avro>=1.7.7,<2.0.0',
'crcmod>=1.7,<2.0',
'dill==0.2.6',
- 'google-cloud-bigquery>=0.22.1,<1.0.0',
'httplib2>=0.8,<0.10',
'mock>=1.0.1,<3.0.0',
'oauth2client>=2.0.1,<4.0.0',
[7/9] beam git commit: Update dataflow container version to beam-0.6.0
Posted by al...@apache.org.
Update dataflow container version to beam-0.6.0
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/f572328c
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/f572328c
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/f572328c
Branch: refs/heads/release-0.6.0
Commit: f572328ce23e70adee8001e3d10f1479bd9a380d
Parents: 3518b12
Author: Ahmet Altay <al...@google.com>
Authored: Wed Mar 8 13:42:36 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:42:36 2017 -0800
----------------------------------------------------------------------
runners/google-cloud-dataflow-java/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/f572328c/runners/google-cloud-dataflow-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/pom.xml b/runners/google-cloud-dataflow-java/pom.xml
index 33ebcfb..dbd0780 100644
--- a/runners/google-cloud-dataflow-java/pom.xml
+++ b/runners/google-cloud-dataflow-java/pom.xml
@@ -33,7 +33,7 @@
<packaging>jar</packaging>
<properties>
- <dataflow.container_version>0.6.0</dataflow.container_version>
+ <dataflow.container_version>beam-0.6.0</dataflow.container_version>
<dataflow.environment_major_version>6</dataflow.environment_major_version>
</properties>
[6/9] beam git commit: [BEAM-1649] Fix unresolved references in
Python SDK
Posted by al...@apache.org.
[BEAM-1649] Fix unresolved references in Python SDK
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/3518b12f
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/3518b12f
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/3518b12f
Branch: refs/heads/release-0.6.0
Commit: 3518b12fbfc984fcfe12e12ba06809e57744f006
Parents: 5e2afa2
Author: Tibor Kiss <ti...@gmail.com>
Authored: Wed Mar 8 12:18:39 2017 +0100
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:41:15 2017 -0800
----------------------------------------------------------------------
sdks/python/apache_beam/io/range_trackers.py | 2 +-
sdks/python/apache_beam/transforms/trigger.py | 2 +-
sdks/python/apache_beam/typehints/decorators.py | 5 ++---
sdks/python/apache_beam/typehints/typehints.py | 3 ++-
sdks/python/apache_beam/utils/pipeline_options.py | 2 +-
5 files changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/3518b12f/sdks/python/apache_beam/io/range_trackers.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/io/range_trackers.py b/sdks/python/apache_beam/io/range_trackers.py
index 059b6ca..8627f76 100644
--- a/sdks/python/apache_beam/io/range_trackers.py
+++ b/sdks/python/apache_beam/io/range_trackers.py
@@ -339,7 +339,7 @@ class OrderedPositionRangeTracker(iobase.RangeTracker):
def stop_position(self):
with self._lock:
- return self._end_position
+ return self._stop_position
def try_claim(self, position):
with self._lock:
http://git-wip-us.apache.org/repos/asf/beam/blob/3518b12f/sdks/python/apache_beam/transforms/trigger.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py
index 69a35ae..04198ba 100644
--- a/sdks/python/apache_beam/transforms/trigger.py
+++ b/sdks/python/apache_beam/transforms/trigger.py
@@ -63,7 +63,7 @@ class ValueStateTag(StateTag):
"""StateTag pointing to an element."""
def __repr__(self):
- return 'ValueStateTag(%s, %s)' % (self.tag, self.combine_fn)
+ return 'ValueStateTag(%s)' % (self.tag)
def with_prefix(self, prefix):
return ValueStateTag(prefix + self.tag)
http://git-wip-us.apache.org/repos/asf/beam/blob/3518b12f/sdks/python/apache_beam/typehints/decorators.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/typehints/decorators.py b/sdks/python/apache_beam/typehints/decorators.py
index df15f1b..9ac9168 100644
--- a/sdks/python/apache_beam/typehints/decorators.py
+++ b/sdks/python/apache_beam/typehints/decorators.py
@@ -234,9 +234,8 @@ def _unpack_positional_arg_hints(arg, hint):
if isinstance(arg, list):
tuple_constraint = typehints.Tuple[[typehints.Any] * len(arg)]
if not typehints.is_consistent_with(hint, tuple_constraint):
- raise typehints.TypeCheckError(
- 'Bad tuple arguments for %s: expected %s, got %s' % (
- arg, tuple_constraint, hint))
+ raise TypeCheckError('Bad tuple arguments for %s: expected %s, got %s' %
+ (arg, tuple_constraint, hint))
if isinstance(hint, typehints.TupleConstraint):
return tuple(_unpack_positional_arg_hints(a, t)
for a, t in zip(arg, hint.tuple_types))
http://git-wip-us.apache.org/repos/asf/beam/blob/3518b12f/sdks/python/apache_beam/typehints/typehints.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/typehints/typehints.py b/sdks/python/apache_beam/typehints/typehints.py
index d0ef60e..20f842a 100644
--- a/sdks/python/apache_beam/typehints/typehints.py
+++ b/sdks/python/apache_beam/typehints/typehints.py
@@ -994,7 +994,8 @@ class WindowedTypeConstraint(TypeConstraint):
'%s hint type-constraint violated. The type of element in '
'is incorrect. Expected an instance of type %s, '
'instead received an instance of type %s.' %
- (repr(self), _unified_repr(self.inner_type), elem.__class__.__name__))
+ (repr(self), _unified_repr(self.inner_type),
+ instance.value.__class__.__name__))
class GeneratorHint(IteratorHint):
http://git-wip-us.apache.org/repos/asf/beam/blob/3518b12f/sdks/python/apache_beam/utils/pipeline_options.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/pipeline_options.py b/sdks/python/apache_beam/utils/pipeline_options.py
index 709808f..d07c328 100644
--- a/sdks/python/apache_beam/utils/pipeline_options.py
+++ b/sdks/python/apache_beam/utils/pipeline_options.py
@@ -158,7 +158,7 @@ class PipelineOptions(HasDisplayData):
# Special methods which may be accessed before the object is
# fully constructed (e.g. in unpickling).
if name[:2] == name[-2:] == '__':
- return object.__getattr__(self, name)
+ return object.__getattribute__(self, name)
elif name in self._visible_option_list():
return self._all_options.get(name, getattr(self._visible_options, name))
else:
[8/9] beam git commit: Update py-sdk version to release version.
Posted by al...@apache.org.
Update py-sdk version to release version.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/edaf3ac9
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/edaf3ac9
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/edaf3ac9
Branch: refs/heads/release-0.6.0
Commit: edaf3ac9f57c208bb7ce444d409b0909ef2d1a67
Parents: f572328
Author: Ahmet Altay <al...@google.com>
Authored: Wed Mar 8 13:43:28 2017 -0800
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Mar 8 13:43:28 2017 -0800
----------------------------------------------------------------------
sdks/python/apache_beam/version.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/edaf3ac9/sdks/python/apache_beam/version.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/version.py b/sdks/python/apache_beam/version.py
index 12509fb..e0f9b2f 100644
--- a/sdks/python/apache_beam/version.py
+++ b/sdks/python/apache_beam/version.py
@@ -21,7 +21,7 @@
import re
-__version__ = '0.6.0.dev'
+__version__ = '0.6.0'
# The following utilities are legacy code from the Maven integration;