You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/01/20 21:05:54 UTC
[1/4] beam git commit: DoFnSignature: deprecate isStateful in favor
of usesState
Repository: beam
Updated Branches:
refs/heads/master eecbcc48b -> 77c7505b8
DoFnSignature: deprecate isStateful in favor of usesState
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/ed434ecf
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/ed434ecf
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/ed434ecf
Branch: refs/heads/master
Commit: ed434ecfdbcfc339815c85aa15852124bb2d3981
Parents: eecbcc4
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri Jan 20 11:28:45 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Fri Jan 20 11:35:10 2017 -0800
----------------------------------------------------------------------
.../beam/runners/dataflow/DataflowPipelineTranslator.java | 2 +-
.../apache/beam/sdk/transforms/reflect/DoFnSignature.java | 10 +++++++---
2 files changed, 8 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/ed434ecf/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 7609745..cfb3d1a 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
@@ -943,7 +943,7 @@ public class DataflowPipelineTranslator {
DoFnInfo.forFn(
fn, windowingStrategy, sideInputs, inputCoder, mainOutput, outputMap))));
- if (signature.isStateful()) {
+ if (signature.usesState()) {
stepContext.addInput(PropertyNames.USES_KEYED_STATE, "true");
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/ed434ecf/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
index ccc9ac3..5255adc 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
@@ -105,13 +105,17 @@ public abstract class DoFnSignature {
@Nullable
public abstract Map<String, OnTimerMethod> onTimerMethods();
- /**
- * Whether the {@link DoFn} described by this signature uses state.
- */
+ /** @deprecated use {@link #usesState()}, it's cleaner */
+ @Deprecated
public boolean isStateful() {
return stateDeclarations().size() > 0;
}
+ /** Whether the {@link DoFn} described by this signature uses state. */
+ public boolean usesState() {
+ return stateDeclarations().size() > 0;
+ }
+
static Builder builder() {
return new AutoValue_DoFnSignature.Builder();
}
[3/4] beam git commit: Set USES_KEYED_STATE in Dataflow when DoFn
uses timers
Posted by ke...@apache.org.
Set USES_KEYED_STATE in Dataflow when DoFn uses timers
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/f535d658
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/f535d658
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/f535d658
Branch: refs/heads/master
Commit: f535d658b551f56ebe4f2c77948fef63be7f44dc
Parents: 978ff55
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri Jan 20 11:36:52 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Fri Jan 20 11:36:52 2017 -0800
----------------------------------------------------------------------
.../apache/beam/runners/dataflow/DataflowPipelineTranslator.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/f535d658/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 cfb3d1a..babbe69 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
@@ -943,7 +943,7 @@ public class DataflowPipelineTranslator {
DoFnInfo.forFn(
fn, windowingStrategy, sideInputs, inputCoder, mainOutput, outputMap))));
- if (signature.usesState()) {
+ if (signature.usesState() || signature.usesTimers()) {
stepContext.addInput(PropertyNames.USES_KEYED_STATE, "true");
}
}
[4/4] beam git commit: This closes #1792: Dataflow USES_KEYED_STATE
when DoFn uses timers
Posted by ke...@apache.org.
This closes #1792: Dataflow USES_KEYED_STATE when DoFn uses timers
Set USES_KEYED_STATE in Dataflow when DoFn uses timers
Add DoFnSignature.usesTimers() shorthand
DoFnSignature: deprecate isStateful in favor of usesState
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/77c7505b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/77c7505b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/77c7505b
Branch: refs/heads/master
Commit: 77c7505b870a8ad7c56053fec84a28eb5b8e7dfa
Parents: eecbcc4 f535d65
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri Jan 20 12:52:07 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Fri Jan 20 12:52:07 2017 -0800
----------------------------------------------------------------------
.../runners/dataflow/DataflowPipelineTranslator.java | 2 +-
.../beam/sdk/transforms/reflect/DoFnSignature.java | 15 ++++++++++++---
2 files changed, 13 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
[2/4] beam git commit: Add DoFnSignature.usesTimers() shorthand
Posted by ke...@apache.org.
Add DoFnSignature.usesTimers() shorthand
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/978ff559
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/978ff559
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/978ff559
Branch: refs/heads/master
Commit: 978ff559f68f1a122346bdbe1a44d2895b598191
Parents: ed434ec
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Jan 18 16:25:51 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Fri Jan 20 11:36:39 2017 -0800
----------------------------------------------------------------------
.../org/apache/beam/sdk/transforms/reflect/DoFnSignature.java | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/978ff559/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
index 5255adc..f470782 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
@@ -116,6 +116,11 @@ public abstract class DoFnSignature {
return stateDeclarations().size() > 0;
}
+ /** Whether the {@link DoFn} described by this signature uses timers. */
+ public boolean usesTimers() {
+ return timerDeclarations().size() > 0;
+ }
+
static Builder builder() {
return new AutoValue_DoFnSignature.Builder();
}