beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tg...@apache.org
Subject [1/2] beam git commit: Improve check around accumulation mode
Date Sat, 18 Feb 2017 03:50:57 GMT
Repository: beam
Updated Branches:
  refs/heads/master 918c7fa36 -> 01de25575


Improve check around accumulation mode

If multiple panes can ever be produced, accumulation mode must be
explicitly set, as there is currently no default accumulation mode. A
pipeline with allowed lateness and the default trigger can produce
multiple panes; as a result, accumulation mode must be set even if there
is no custom trigger if allowed lateness is specified.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/89cbf3bf
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/89cbf3bf
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/89cbf3bf

Branch: refs/heads/master
Commit: 89cbf3bfbbf6677cb294b3d85645f2f90f85414a
Parents: 918c7fa
Author: Thomas Groh <tgroh@google.com>
Authored: Fri Feb 17 09:36:41 2017 -0800
Committer: Thomas Groh <tgroh@google.com>
Committed: Fri Feb 17 19:50:28 2017 -0800

----------------------------------------------------------------------
 .../beam/sdk/transforms/windowing/Window.java   | 42 +++++++++++------
 .../beam/sdk/transforms/SplittableDoFnTest.java |  3 +-
 .../sdk/transforms/windowing/WindowTest.java    | 48 ++++++++++++++++++--
 3 files changed, 72 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/89cbf3bf/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 faa3704..eac1c97 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
@@ -434,22 +434,34 @@ public class Window {
 
       // Make sure that the windowing strategy is complete & valid.
       if (outputStrategy.isTriggerSpecified()
-          && !(outputStrategy.getTrigger() instanceof DefaultTrigger)) {
-        if (!(outputStrategy.getWindowFn() instanceof GlobalWindows)
-            && !outputStrategy.isAllowedLatenessSpecified()) {
-          throw new IllegalArgumentException("Except when using GlobalWindows,"
-              + " calling .triggering() to specify a trigger requires that the allowed lateness
be"
-              + " specified using .withAllowedLateness() to set the upper bound on how late
data"
-              + " can arrive before being dropped. See Javadoc for more details.");
-        }
-
-        if (!outputStrategy.isModeSpecified()) {
-          throw new IllegalArgumentException(
-              "Calling .triggering() to specify a trigger requires that the accumulation
mode be"
-              + " specified using .discardingFiredPanes() or .accumulatingFiredPanes()."
-              + " See Javadoc for more details.");
-        }
+          && !(outputStrategy.getTrigger() instanceof DefaultTrigger)
+          && !(outputStrategy.getWindowFn() instanceof GlobalWindows)
+          && !outputStrategy.isAllowedLatenessSpecified()) {
+        throw new IllegalArgumentException(
+            "Except when using GlobalWindows,"
+                + " calling .triggering() to specify a trigger requires that the allowed
lateness"
+                + " be specified using .withAllowedLateness() to set the upper bound on how
late"
+                + " data can arrive before being dropped. See Javadoc for more details.");
       }
+
+      if (!outputStrategy.isModeSpecified() && canProduceMultiplePanes(outputStrategy))
{
+        throw new IllegalArgumentException(
+            "Calling .triggering() to specify a trigger or calling .withAllowedLateness()
to"
+                + " specify an allowed lateness greater than zero requires that the accumulation"
+                + " mode be specified using .discardingFiredPanes() or .accumulatingFiredPanes()."
+                + " See Javadoc for more details.");
+      }
+    }
+
+    private boolean canProduceMultiplePanes(WindowingStrategy<?, ?> strategy) {
+      // The default trigger is Repeatedly.forever(AfterWatermark.pastEndOfWindow()); This
fires
+      // for every late-arriving element if allowed lateness is nonzero, and thus we must
have
+      // an accumulating mode specified
+      boolean dataCanArriveLate =
+          !(strategy.getWindowFn() instanceof GlobalWindows)
+              && strategy.getAllowedLateness().getMillis() > 0;
+      boolean hasCustomTrigger = !(strategy.getTrigger() instanceof DefaultTrigger);
+      return dataCanArriveLate || hasCustomTrigger;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/89cbf3bf/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
index e5c0ef0..fefccc4 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
@@ -310,7 +310,8 @@ public class SplittableDoFnTest {
         p.apply(stream)
             .apply(
                 Window.<String>into(FixedWindows.of(Duration.standardMinutes(1)))
-                    .withAllowedLateness(Duration.standardMinutes(1)));
+                    .withAllowedLateness(Duration.standardMinutes(1))
+                    .discardingFiredPanes());
 
     PCollection<KV<String, Integer>> afterSDF =
         input

http://git-wip-us.apache.org/repos/asf/beam/blob/89cbf3bf/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
index e21668e..55c7297 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
@@ -21,6 +21,7 @@ import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisp
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.includesDisplayDataFor;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.isOneOf;
 import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
@@ -42,6 +43,7 @@ import org.apache.beam.sdk.transforms.display.DisplayData;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.util.WindowingStrategy.AccumulationMode;
 import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TimestampedValue;
 import org.hamcrest.Matchers;
 import org.joda.time.Duration;
@@ -96,6 +98,26 @@ public class WindowTest implements Serializable {
   }
 
   @Test
+  public void testWindowIntoAccumulatingLatenessNoTrigger() {
+    FixedWindows fixed = FixedWindows.of(Duration.standardMinutes(10));
+    WindowingStrategy<?, ?> strategy =
+        pipeline
+            .apply(Create.of("hello", "world").withCoder(StringUtf8Coder.of()))
+            .apply(
+                "Lateness",
+                Window.<String>into(fixed)
+                    .withAllowedLateness(Duration.standardDays(1))
+                    .accumulatingFiredPanes())
+            .getWindowingStrategy();
+
+    assertThat(strategy.isTriggerSpecified(), is(false));
+    assertThat(strategy.isModeSpecified(), is(true));
+    assertThat(strategy.isAllowedLatenessSpecified(), is(true));
+    assertThat(strategy.getMode(), equalTo(AccumulationMode.ACCUMULATING_FIRED_PANES));
+    assertThat(strategy.getAllowedLateness(), equalTo(Duration.standardDays(1)));
+  }
+
+  @Test
   public void testWindowPropagatesEachPart() {
     FixedWindows fixed10 = FixedWindows.of(Duration.standardMinutes(10));
     Repeatedly trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
@@ -159,13 +181,29 @@ public class WindowTest implements Serializable {
     FixedWindows fixed10 = FixedWindows.of(Duration.standardMinutes(10));
     Repeatedly trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
 
+    PCollection<String> input =
+        pipeline
+            .apply(Create.of("hello", "world").withCoder(StringUtf8Coder.of()))
+            .apply("Window", Window.<String>into(fixed10));
     thrown.expect(IllegalArgumentException.class);
     thrown.expectMessage("requires that the accumulation mode");
-    pipeline
-      .apply(Create.of("hello", "world").withCoder(StringUtf8Coder.of()))
-      .apply("Window", Window.<String>into(fixed10))
-      .apply("Lateness", Window.<String>withAllowedLateness(Duration.standardDays(1)))
-      .apply("Trigger", Window.<String>triggering(trigger));
+    input.apply(
+        "Triggering",
+        Window.<String>withAllowedLateness(Duration.standardDays(1)).triggering(trigger));
+  }
+
+  @Test
+  public void testMissingModeViaLateness() {
+    FixedWindows fixed = FixedWindows.of(Duration.standardMinutes(10));
+    PCollection<String> input =
+        pipeline
+            .apply(Create.of("hello", "world").withCoder(StringUtf8Coder.of()))
+            .apply("Window", Window.<String>into(fixed));
+    thrown.expect(IllegalArgumentException.class);
+    thrown.expectMessage("allowed lateness");
+    thrown.expectMessage("accumulation mode be specified");
+    input
+        .apply("Lateness", Window.<String>withAllowedLateness(Duration.standardDays(1)));
   }
 
   @Test


Mime
View raw message