beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bchamb...@apache.org
Subject [1/2] incubator-beam git commit: Make example AddTimestampFn range deterministic
Date Mon, 13 Jun 2016 18:39:23 GMT
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 <swegner@google.com>
Authored: Thu Jun 9 11:31:23 2016 -0700
Committer: bchambers <bchambers@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.
        */


Mime
View raw message