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 2016/12/21 21:59:50 UTC
[1/7] incubator-beam git commit: Add informative Instant formatter to
BoundedWindow
Repository: incubator-beam
Updated Branches:
refs/heads/master ff3951699 -> 57d9bbd79
Add informative Instant formatter to BoundedWindow
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/280a6a8f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/280a6a8f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/280a6a8f
Branch: refs/heads/master
Commit: 280a6a8f729cb382616ad65f71860b61277cbd6f
Parents: ff39516
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon Dec 19 20:40:11 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:45:36 2016 -0800
----------------------------------------------------------------------
.../beam/sdk/transforms/windowing/BoundedWindow.java | 12 ++++++++++++
1 file changed, 12 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/280a6a8f/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
index 3654074..6da2495 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
@@ -39,6 +39,18 @@ public abstract class BoundedWindow {
public static final Instant TIMESTAMP_MAX_VALUE =
new Instant(TimeUnit.MICROSECONDS.toMillis(Long.MAX_VALUE));
+ public static String formatTimestamp(Instant timestamp) {
+ if (timestamp.equals(TIMESTAMP_MIN_VALUE)) {
+ return timestamp.toString() + " (TIMESTAMP_MIN_VALUE)";
+ } else if (timestamp.equals(TIMESTAMP_MAX_VALUE)) {
+ return timestamp.toString() + " (TIMESTAMP_MAX_VALUE)";
+ } else if (timestamp.equals(GlobalWindow.INSTANCE.maxTimestamp())) {
+ return timestamp.toString() + " (end of global window)";
+ } else {
+ return timestamp.toString();
+ }
+ }
+
/**
* Returns the inclusive upper bound of timestamps for values in this window.
*/
[3/7] incubator-beam git commit: Use informative Instant formatter in
WatermarkHold
Posted by ke...@apache.org.
Use informative Instant formatter in WatermarkHold
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/fa4958a6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/fa4958a6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/fa4958a6
Branch: refs/heads/master
Commit: fa4958a6140eb00ceee08b2468f7d88f17538794
Parents: 280a6a8
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon Dec 19 20:40:47 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:45:37 2016 -0800
----------------------------------------------------------------------
.../apache/beam/runners/core/WatermarkHold.java | 4 +++-
.../sdk/transforms/windowing/BoundedWindow.java | 19 +++++++++++++++++++
2 files changed, 22 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/fa4958a6/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
index 7f1afcc..5e5f44d 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
@@ -207,7 +207,9 @@ class WatermarkHold<W extends BoundedWindow> implements Serializable {
Instant shifted = windowingStrategy.getOutputTimeFn().assignOutputTime(timestamp, window);
checkState(!shifted.isBefore(timestamp),
"OutputTimeFn moved element from %s to earlier time %s for window %s",
- timestamp, shifted, window);
+ BoundedWindow.formatTimestamp(timestamp),
+ BoundedWindow.formatTimestamp(shifted),
+ window);
checkState(timestamp.isAfter(window.maxTimestamp())
|| !shifted.isAfter(window.maxTimestamp()),
"OutputTimeFn moved element from %s to %s which is beyond end of "
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/fa4958a6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
index 6da2495..74223b5 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java
@@ -34,11 +34,30 @@ import org.joda.time.Instant;
public abstract class BoundedWindow {
// The min and max timestamps that won't overflow when they are converted to
// usec.
+
+ /**
+ * The minimum value for any Beam timestamp. Often referred to as "-infinity".
+ *
+ * <p>This value and {@link #TIMESTAMP_MAX_VALUE} are chosen so that their
+ * microseconds-since-epoch can be safely represented with a {@code long}.
+ */
public static final Instant TIMESTAMP_MIN_VALUE =
new Instant(TimeUnit.MICROSECONDS.toMillis(Long.MIN_VALUE));
+
+ /**
+ * The maximum value for any Beam timestamp. Often referred to as "+infinity".
+ *
+ * <p>This value and {@link #TIMESTAMP_MIN_VALUE} are chosen so that their
+ * microseconds-since-epoch can be safely represented with a {@code long}.
+ */
public static final Instant TIMESTAMP_MAX_VALUE =
new Instant(TimeUnit.MICROSECONDS.toMillis(Long.MAX_VALUE));
+ /**
+ * Formats a {@link Instant} timestamp with additional Beam-specific metadata, such as indicating
+ * whether the timestamp is the end of the global window or one of the distinguished values {@link
+ * #TIMESTAMP_MIN_VALUE} or {@link #TIMESTAMP_MIN_VALUE}.
+ */
public static String formatTimestamp(Instant timestamp) {
if (timestamp.equals(TIMESTAMP_MIN_VALUE)) {
return timestamp.toString() + " (TIMESTAMP_MIN_VALUE)";
[7/7] incubator-beam git commit: This closes #1669: Preliminaries for
timers in the direct runner
Posted by ke...@apache.org.
This closes #1669: Preliminaries for timers in the direct runner
Hold output watermark according to pending timers
Allow setting timer by ID in DirectTimerInternals
Add UsesTestStream for use with JUnit @Category
Add static Window.withOutputTimeFn to match build method
Use informative Instant formatter in WatermarkHold
Add informative Instant formatter to BoundedWindow
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/57d9bbd7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/57d9bbd7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/57d9bbd7
Branch: refs/heads/master
Commit: 57d9bbd797edfcf32fdd9284b802fc4f9694e8d2
Parents: ff39516 dfe2e62
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Dec 21 13:46:34 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:46:34 2016 -0800
----------------------------------------------------------------------
.../apache/beam/runners/core/WatermarkHold.java | 4 +-
.../runners/direct/DirectTimerInternals.java | 2 +-
.../beam/runners/direct/WatermarkManager.java | 78 ++++++++++++++++++--
.../apache/beam/sdk/testing/UsesTestStream.java | 24 ++++++
.../sdk/transforms/windowing/BoundedWindow.java | 31 ++++++++
.../beam/sdk/transforms/windowing/Window.java | 9 +++
.../apache/beam/sdk/testing/TestStreamTest.java | 12 +--
7 files changed, 144 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
[4/7] incubator-beam git commit: Allow setting timer by ID in
DirectTimerInternals
Posted by ke...@apache.org.
Allow setting timer by ID in DirectTimerInternals
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7f14c463
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7f14c463
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7f14c463
Branch: refs/heads/master
Commit: 7f14c463acd2ae5b86ac81a9528ac4aa7dff765f
Parents: 4d71924
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Dec 7 20:18:44 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:45:37 2016 -0800
----------------------------------------------------------------------
.../runners/direct/DirectTimerInternals.java | 2 +-
.../beam/runners/direct/WatermarkManager.java | 25 ++++++++++++++++++++
2 files changed, 26 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7f14c463/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
index 5ca276d..80e0721 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectTimerInternals.java
@@ -49,7 +49,7 @@ class DirectTimerInternals implements TimerInternals {
@Override
public void setTimer(StateNamespace namespace, String timerId, Instant target,
TimeDomain timeDomain) {
- throw new UnsupportedOperationException("Setting timer by ID not yet supported.");
+ timerUpdateBuilder.setTimer(TimerData.of(timerId, namespace, target, timeDomain));
}
@Deprecated
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7f14c463/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
index 7bed751..f7bafd1 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
@@ -23,11 +23,13 @@ import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.collect.ComparisonChain;
+import com.google.common.collect.HashBasedTable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Ordering;
import com.google.common.collect.SortedMultiset;
+import com.google.common.collect.Table;
import com.google.common.collect.TreeMultiset;
import java.io.Serializable;
import java.util.ArrayList;
@@ -56,6 +58,7 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.TimerInternals;
import org.apache.beam.sdk.util.TimerInternals.TimerData;
+import org.apache.beam.sdk.util.state.StateNamespace;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TaggedPValue;
import org.joda.time.Instant;
@@ -210,6 +213,10 @@ public class WatermarkManager {
private final SortedMultiset<CommittedBundle<?>> pendingElements;
private final Map<StructuralKey<?>, NavigableSet<TimerData>> objectTimers;
+ // Entries in this table represent the authoritative timestamp for which
+ // a per-key-and-StateNamespace timer is set.
+ private final Map<StructuralKey<?>, Table<StateNamespace, String, TimerData>> existingTimers;
+
private AtomicReference<Instant> currentWatermark;
public AppliedPTransformInputWatermark(Collection<? extends Watermark> inputWatermarks) {
@@ -222,6 +229,7 @@ public class WatermarkManager {
this.pendingElements =
TreeMultiset.create(pendingBundleComparator);
this.objectTimers = new HashMap<>();
+ this.existingTimers = new HashMap<>();
currentWatermark = new AtomicReference<>(BoundedWindow.TIMESTAMP_MIN_VALUE);
}
@@ -276,14 +284,31 @@ public class WatermarkManager {
keyTimers = new TreeSet<>();
objectTimers.put(update.key, keyTimers);
}
+ Table<StateNamespace, String, TimerData> existingTimersForKey =
+ existingTimers.get(update.key);
+ if (existingTimersForKey == null) {
+ existingTimersForKey = HashBasedTable.create();
+ existingTimers.put(update.key, existingTimersForKey);
+ }
+
for (TimerData timer : update.setTimers) {
if (TimeDomain.EVENT_TIME.equals(timer.getDomain())) {
+ @Nullable
+ TimerData existingTimer =
+ existingTimersForKey.get(timer.getNamespace(), timer.getTimerId());
+
+ if (existingTimer != null) {
+ keyTimers.remove(existingTimer);
+ }
keyTimers.add(timer);
+ existingTimersForKey.put(timer.getNamespace(), timer.getTimerId(), timer);
}
}
+
for (TimerData timer : update.deletedTimers) {
if (TimeDomain.EVENT_TIME.equals(timer.getDomain())) {
keyTimers.remove(timer);
+ existingTimersForKey.remove(timer.getNamespace(), timer.getTimerId());
}
}
// We don't keep references to timers that have been fired and delivered via #getFiredTimers()
[6/7] incubator-beam git commit: Add UsesTestStream for use with
JUnit @Category
Posted by ke...@apache.org.
Add UsesTestStream for use with JUnit @Category
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/4d71924c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/4d71924c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/4d71924c
Branch: refs/heads/master
Commit: 4d71924ccda9dae97c7cc9535a9780df9457cc3f
Parents: 8188040
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Dec 20 14:20:07 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:45:37 2016 -0800
----------------------------------------------------------------------
.../apache/beam/sdk/testing/UsesTestStream.java | 24 ++++++++++++++++++++
.../apache/beam/sdk/testing/TestStreamTest.java | 12 +++++-----
2 files changed, 30 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4d71924c/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesTestStream.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesTestStream.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesTestStream.java
new file mode 100644
index 0000000..8debb46
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesTestStream.java
@@ -0,0 +1,24 @@
+/*
+ * 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.sdk.testing;
+
+/**
+ * Category tag for tests that use {@link TestStream}, which is not a part of the Beam model
+ * but a special feature currently only implemented by the direct runner.
+ */
+public interface UsesTestStream {}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/4d71924c/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java
index 64aeca3..c12e9f3 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java
@@ -69,7 +69,7 @@ public class TestStreamTest implements Serializable {
@Rule public transient ExpectedException thrown = ExpectedException.none();
@Test
- @Category(NeedsRunner.class)
+ @Category({NeedsRunner.class, UsesTestStream.class})
public void testLateDataAccumulating() {
Instant instant = new Instant(0);
TestStream<Integer> source = TestStream.create(VarIntCoder.of())
@@ -136,7 +136,7 @@ public class TestStreamTest implements Serializable {
}
@Test
- @Category(NeedsRunner.class)
+ @Category({NeedsRunner.class, UsesTestStream.class})
public void testProcessingTimeTrigger() {
TestStream<Long> source = TestStream.create(VarLongCoder.of())
.addElements(TimestampedValue.of(1L, new Instant(1000L)),
@@ -159,7 +159,7 @@ public class TestStreamTest implements Serializable {
}
@Test
- @Category(NeedsRunner.class)
+ @Category({NeedsRunner.class, UsesTestStream.class})
public void testDiscardingMode() {
TestStream<String> stream =
TestStream.create(StringUtf8Coder.of())
@@ -208,7 +208,7 @@ public class TestStreamTest implements Serializable {
}
@Test
- @Category(NeedsRunner.class)
+ @Category({NeedsRunner.class, UsesTestStream.class})
public void testFirstElementLate() {
Instant lateElementTimestamp = new Instant(-1_000_000);
TestStream<String> stream =
@@ -238,7 +238,7 @@ public class TestStreamTest implements Serializable {
}
@Test
- @Category(NeedsRunner.class)
+ @Category({NeedsRunner.class, UsesTestStream.class})
public void testElementsAtAlmostPositiveInfinity() {
Instant endOfGlobalWindow = GlobalWindow.INSTANCE.maxTimestamp();
TestStream<String> stream = TestStream.create(StringUtf8Coder.of())
@@ -261,7 +261,7 @@ public class TestStreamTest implements Serializable {
}
@Test
- @Category(NeedsRunner.class)
+ @Category({NeedsRunner.class, UsesTestStream.class})
public void testMultipleStreams() {
TestStream<String> stream = TestStream.create(StringUtf8Coder.of())
.addElements("foo", "bar")
[2/7] incubator-beam git commit: Hold output watermark according to
pending timers
Posted by ke...@apache.org.
Hold output watermark according to pending timers
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/dfe2e62d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/dfe2e62d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/dfe2e62d
Branch: refs/heads/master
Commit: dfe2e62d103595583e3ca4594cc03885fe1bba16
Parents: 7f14c46
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Dec 20 13:37:40 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:45:37 2016 -0800
----------------------------------------------------------------------
.../beam/runners/direct/WatermarkManager.java | 59 ++++++++++++++++----
1 file changed, 48 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dfe2e62d/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
index f7bafd1..248fafd 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
@@ -211,12 +211,18 @@ public class WatermarkManager {
private static class AppliedPTransformInputWatermark implements Watermark {
private final Collection<? extends Watermark> inputWatermarks;
private final SortedMultiset<CommittedBundle<?>> pendingElements;
- private final Map<StructuralKey<?>, NavigableSet<TimerData>> objectTimers;
+
+ // This tracks only the quantity of timers at each timestamp, for quickly getting the cross-key
+ // minimum
+ private final SortedMultiset<Instant> pendingTimers;
// Entries in this table represent the authoritative timestamp for which
// a per-key-and-StateNamespace timer is set.
private final Map<StructuralKey<?>, Table<StateNamespace, String, TimerData>> existingTimers;
+ // This per-key sorted set allows quick retrieval of timers that should fire for a key
+ private final Map<StructuralKey<?>, NavigableSet<TimerData>> objectTimers;
+
private AtomicReference<Instant> currentWatermark;
public AppliedPTransformInputWatermark(Collection<? extends Watermark> inputWatermarks) {
@@ -224,10 +230,13 @@ public class WatermarkManager {
// The ordering must order elements by timestamp, and must not compare two distinct elements
// as equal. This is built on the assumption that any element added as a pending element will
// be consumed without modifications.
+ //
+ // The same logic is applied for pending timers
Ordering<CommittedBundle<?>> pendingBundleComparator =
new BundleByElementTimestampComparator().compound(Ordering.arbitrary());
this.pendingElements =
TreeMultiset.create(pendingBundleComparator);
+ this.pendingTimers = TreeMultiset.create();
this.objectTimers = new HashMap<>();
this.existingTimers = new HashMap<>();
currentWatermark = new AtomicReference<>(BoundedWindow.TIMESTAMP_MIN_VALUE);
@@ -278,6 +287,14 @@ public class WatermarkManager {
pendingElements.remove(completed);
}
+ private synchronized Instant getEarliestTimerTimestamp() {
+ if (pendingTimers.isEmpty()) {
+ return BoundedWindow.TIMESTAMP_MAX_VALUE;
+ } else {
+ return pendingTimers.firstEntry().getElement();
+ }
+ }
+
private synchronized void updateTimers(TimerUpdate update) {
NavigableSet<TimerData> keyTimers = objectTimers.get(update.key);
if (keyTimers == null) {
@@ -291,27 +308,43 @@ public class WatermarkManager {
existingTimers.put(update.key, existingTimersForKey);
}
- for (TimerData timer : update.setTimers) {
+ for (TimerData timer : update.getSetTimers()) {
+ if (TimeDomain.EVENT_TIME.equals(timer.getDomain())) {
+ @Nullable
+ TimerData existingTimer =
+ existingTimersForKey.get(timer.getNamespace(), timer.getTimerId());
+
+ if (existingTimer == null) {
+ pendingTimers.add(timer.getTimestamp());
+ keyTimers.add(timer);
+ } else if (!existingTimer.equals(timer)) {
+ keyTimers.remove(existingTimer);
+ keyTimers.add(timer);
+ } // else the timer is already set identically, so noop
+
+ existingTimersForKey.put(timer.getNamespace(), timer.getTimerId(), timer);
+ }
+ }
+
+ for (TimerData timer : update.getDeletedTimers()) {
if (TimeDomain.EVENT_TIME.equals(timer.getDomain())) {
@Nullable
TimerData existingTimer =
existingTimersForKey.get(timer.getNamespace(), timer.getTimerId());
if (existingTimer != null) {
+ pendingTimers.remove(existingTimer.getTimestamp());
keyTimers.remove(existingTimer);
+ existingTimersForKey.remove(existingTimer.getNamespace(), existingTimer.getTimerId());
}
- keyTimers.add(timer);
- existingTimersForKey.put(timer.getNamespace(), timer.getTimerId(), timer);
}
}
- for (TimerData timer : update.deletedTimers) {
+ for (TimerData timer : update.getCompletedTimers()) {
if (TimeDomain.EVENT_TIME.equals(timer.getDomain())) {
- keyTimers.remove(timer);
- existingTimersForKey.remove(timer.getNamespace(), timer.getTimerId());
+ pendingTimers.remove(timer.getTimestamp());
}
}
- // We don't keep references to timers that have been fired and delivered via #getFiredTimers()
}
private synchronized Map<StructuralKey<?>, List<TimerData>> extractFiredEventTimeTimers() {
@@ -336,11 +369,12 @@ public class WatermarkManager {
* {@link #refresh()} for more information.
*/
private static class AppliedPTransformOutputWatermark implements Watermark {
- private final Watermark inputWatermark;
+ private final AppliedPTransformInputWatermark inputWatermark;
private final PerKeyHolds holds;
private AtomicReference<Instant> currentWatermark;
- public AppliedPTransformOutputWatermark(AppliedPTransformInputWatermark inputWatermark) {
+ public AppliedPTransformOutputWatermark(
+ AppliedPTransformInputWatermark inputWatermark) {
this.inputWatermark = inputWatermark;
holds = new PerKeyHolds();
currentWatermark = new AtomicReference<>(BoundedWindow.TIMESTAMP_MIN_VALUE);
@@ -377,7 +411,10 @@ public class WatermarkManager {
@Override
public synchronized WatermarkUpdate refresh() {
Instant oldWatermark = currentWatermark.get();
- Instant newWatermark = INSTANT_ORDERING.min(inputWatermark.get(), holds.getMinHold());
+ Instant newWatermark = INSTANT_ORDERING.min(
+ inputWatermark.get(),
+ inputWatermark.getEarliestTimerTimestamp(),
+ holds.getMinHold());
newWatermark = INSTANT_ORDERING.max(oldWatermark, newWatermark);
currentWatermark.set(newWatermark);
return WatermarkUpdate.fromTimestamps(oldWatermark, newWatermark);
[5/7] incubator-beam git commit: Add static Window.withOutputTimeFn
to match build method
Posted by ke...@apache.org.
Add static Window.withOutputTimeFn to match build method
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8188040d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8188040d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8188040d
Branch: refs/heads/master
Commit: 8188040d930b1fa49efd4ed7d5f821d05d6f28ef
Parents: fa4958a
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Dec 20 13:57:55 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Dec 21 13:45:37 2016 -0800
----------------------------------------------------------------------
.../org/apache/beam/sdk/transforms/windowing/Window.java | 9 +++++++++
1 file changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8188040d/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
index 0c430d0..1241abe 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java
@@ -223,6 +223,15 @@ public class Window {
}
/**
+ * <b><i>(Experimental)</i></b> Override the default {@link OutputTimeFn}, to control
+ * the output timestamp of values output from a {@link GroupByKey} operation.
+ */
+ @Experimental(Kind.OUTPUT_TIME)
+ public static <T> Bound<T> withOutputTimeFn(OutputTimeFn<?> outputTimeFn) {
+ return new Bound(null).withOutputTimeFn(outputTimeFn);
+ }
+
+ /**
* A {@code PTransform} that windows the elements of a {@code PCollection<T>},
* into finite windows according to a user-specified {@code WindowFn}.
*