You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by bc...@apache.org on 2016/06/13 18:39:23 UTC
[1/2] incubator-beam git commit: Make example AddTimestampFn range
deterministic
Repository: incubator-beam
Updated Branches:
refs/heads/master 9e0d7d650 -> e1b305ea5
Make example AddTimestampFn range deterministic
The timestamps added in the WindowedWordCount example are currently
based on when the bundles are executed, which makes the min/max bounds
non-deterministic. This change makes the range based on the
construction time.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/335202a0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/335202a0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/335202a0
Branch: refs/heads/master
Commit: 335202a033ced6f30f1b0e5df9da047241abc750
Parents: 9e0d7d6
Author: Scott Wegner <sw...@google.com>
Authored: Thu Jun 9 11:31:23 2016 -0700
Committer: bchambers <bc...@google.com>
Committed: Mon Jun 13 11:21:25 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/beam/examples/WindowedWordCount.java | 11 ++++++++---
1 file changed, 8 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/335202a0/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
index 4d019bb..9ca26bf 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java
@@ -124,13 +124,18 @@ public class WindowedWordCount {
* 2-hour period.
*/
static class AddTimestampFn extends DoFn<String, String> {
- private static final long RAND_RANGE = 7200000; // 2 hours in ms
+ private static final Duration RAND_RANGE = Duration.standardHours(2);
+ private final Instant minTimestamp;
+
+ AddTimestampFn() {
+ this.minTimestamp = new Instant(System.currentTimeMillis());
+ }
@Override
public void processElement(ProcessContext c) {
// Generate a timestamp that falls somewhere in the past two hours.
- long randomTimestamp = System.currentTimeMillis()
- - (int) (Math.random() * RAND_RANGE);
+ long randMillis = (long) (Math.random() * RAND_RANGE.getMillis());
+ Instant randomTimestamp = minTimestamp.plus(randMillis);
/**
* Concept #2: Set the data element with that timestamp.
*/
[2/2] incubator-beam git commit: This closes #441
Posted by bc...@apache.org.
This closes #441
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e1b305ea
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e1b305ea
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e1b305ea
Branch: refs/heads/master
Commit: e1b305ea5d48e0f5f71ad0030c20c492b0f95c30
Parents: 9e0d7d6 335202a
Author: bchambers <bc...@google.com>
Authored: Mon Jun 13 11:21:31 2016 -0700
Committer: bchambers <bc...@google.com>
Committed: Mon Jun 13 11:21:31 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/beam/examples/WindowedWordCount.java | 11 ++++++++---
1 file changed, 8 insertions(+), 3 deletions(-)
----------------------------------------------------------------------