beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [5/9] beam git commit: Add conversion to/from Java SDK trigger to runner API proto
Date Wed, 15 Feb 2017 04:51:47 GMT
Add conversion to/from Java SDK trigger to runner API proto


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

Branch: refs/heads/master
Commit: f4ceaeefe9e8e9d069b760e166c7057a00465360
Parents: 2803864
Author: Kenneth Knowles <klk@google.com>
Authored: Sat Feb 11 17:50:27 2017 -0800
Committer: Kenneth Knowles <klk@google.com>
Committed: Tue Feb 14 14:55:49 2017 -0800

----------------------------------------------------------------------
 .../beam/sdk/transforms/windowing/Triggers.java | 313 +++++++++++++++++++
 .../sdk/transforms/windowing/TriggersTest.java  | 100 ++++++
 2 files changed, 413 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/f4ceaeef/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Triggers.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Triggers.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Triggers.java
new file mode 100644
index 0000000..8ac904c
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Triggers.java
@@ -0,0 +1,313 @@
+/*
+ * 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.transforms.windowing;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+import java.io.Serializable;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi;
+import org.apache.beam.sdk.transforms.windowing.AfterWatermark.AfterWatermarkEarlyAndLate;
+import org.apache.beam.sdk.transforms.windowing.Never.NeverTrigger;
+import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
+import org.apache.beam.sdk.util.TimeDomain;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** Utilities for working with {@link Triggers Triggers}. */
+@Experimental(Experimental.Kind.TRIGGER)
+public class Triggers implements Serializable {
+
+  @VisibleForTesting static final ProtoConverter CONVERTER = new ProtoConverter();
+
+  public static RunnerApi.Trigger toProto(Trigger trigger) {
+    return CONVERTER.convertTrigger(trigger);
+  }
+
+  @VisibleForTesting
+  static class ProtoConverter {
+
+    public RunnerApi.Trigger convertTrigger(Trigger trigger) {
+      Method evaluationMethod = getEvaluationMethod(trigger.getClass());
+      return tryConvert(evaluationMethod, trigger);
+    }
+
+    private RunnerApi.Trigger tryConvert(Method evaluationMethod, Trigger trigger) {
+      try {
+        return (RunnerApi.Trigger) evaluationMethod.invoke(this, trigger);
+      } catch (InvocationTargetException exc) {
+        if (exc.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) exc.getCause();
+        } else {
+          throw new RuntimeException(exc.getCause());
+        }
+      } catch (IllegalAccessException exc) {
+        throw new IllegalStateException(
+            String.format("Internal error: could not invoke %s", evaluationMethod));
+      }
+    }
+
+    private Method getEvaluationMethod(Class<?> clazz) {
+      try {
+        return getClass().getDeclaredMethod("convertSpecific", clazz);
+      } catch (NoSuchMethodException exc) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Cannot translate trigger class %s to a runner-API proto.",
+                clazz.getCanonicalName()),
+            exc);
+      }
+    }
+
+    private RunnerApi.Trigger convertSpecific(DefaultTrigger v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setDefault(RunnerApi.Trigger.Default.getDefaultInstance())
+          .build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterWatermark.FromEndOfWindow v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setAfterEndOfWidow(RunnerApi.Trigger.AfterEndOfWindow.newBuilder())
+          .build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(NeverTrigger v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setNever(RunnerApi.Trigger.Never.getDefaultInstance())
+          .build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterSynchronizedProcessingTime v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setAfterSynchronizedProcessingTime(
+              RunnerApi.Trigger.AfterSynchronizedProcessingTime.getDefaultInstance())
+          .build();
+    }
+
+    private RunnerApi.TimeDomain convertTimeDomain(TimeDomain timeDomain) {
+      switch (timeDomain) {
+        case EVENT_TIME:
+          return RunnerApi.TimeDomain.EVENT_TIME;
+        case PROCESSING_TIME:
+          return RunnerApi.TimeDomain.PROCESSING_TIME;
+        case SYNCHRONIZED_PROCESSING_TIME:
+          return RunnerApi.TimeDomain.SYNCHRONIZED_PROCESSING_TIME;
+        default:
+          throw new IllegalArgumentException(String.format("Unknown time domain: %s", timeDomain));
+      }
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterFirst v) {
+      RunnerApi.Trigger.AfterAny.Builder builder = RunnerApi.Trigger.AfterAny.newBuilder();
+
+      for (Trigger subtrigger : v.subTriggers()) {
+        builder.addSubtriggers(toProto(subtrigger));
+      }
+
+      return RunnerApi.Trigger.newBuilder().setAfterAny(builder).build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterAll v) {
+      RunnerApi.Trigger.AfterAll.Builder builder = RunnerApi.Trigger.AfterAll.newBuilder();
+
+      for (Trigger subtrigger : v.subTriggers()) {
+        builder.addSubtriggers(toProto(subtrigger));
+      }
+
+      return RunnerApi.Trigger.newBuilder().setAfterAll(builder).build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterPane v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setElementCount(
+              RunnerApi.Trigger.ElementCount.newBuilder().setElementCount(v.getElementCount()))
+          .build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterWatermark.AfterWatermarkEarlyAndLate v)
{
+      RunnerApi.Trigger.AfterEndOfWindow.Builder builder =
+          RunnerApi.Trigger.AfterEndOfWindow.newBuilder();
+
+      builder.setEarlyFirings(toProto(v.getEarlyTrigger()));
+      if (v.getLateTrigger() != null) {
+        builder.setLateFirings(toProto(v.getLateTrigger()));
+      }
+
+      return RunnerApi.Trigger.newBuilder().setAfterEndOfWidow(builder).build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterEach v) {
+      RunnerApi.Trigger.AfterEach.Builder builder = RunnerApi.Trigger.AfterEach.newBuilder();
+
+      for (Trigger subtrigger : v.subTriggers()) {
+        builder.addSubtriggers(toProto(subtrigger));
+      }
+
+      return RunnerApi.Trigger.newBuilder().setAfterEach(builder).build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(Repeatedly v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setRepeat(
+              RunnerApi.Trigger.Repeat.newBuilder()
+                  .setSubtrigger(toProto(v.getRepeatedTrigger())))
+          .build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(OrFinallyTrigger v) {
+      return RunnerApi.Trigger.newBuilder()
+          .setOrFinally(
+              RunnerApi.Trigger.OrFinally.newBuilder()
+                  .setMain(toProto(v.getMainTrigger()))
+                  .setFinally(toProto(v.getUntilTrigger())))
+          .build();
+    }
+
+    private RunnerApi.Trigger convertSpecific(AfterProcessingTime v) {
+      RunnerApi.Trigger.AfterProcessingTime.Builder builder =
+          RunnerApi.Trigger.AfterProcessingTime.newBuilder();
+
+      for (TimestampTransform transform : v.getTimestampTransforms()) {
+        builder.addTimestampTransforms(convertTimestampTransform(transform));
+      }
+
+      return RunnerApi.Trigger.newBuilder().setAfterProcessingTime(builder).build();
+    }
+
+    private RunnerApi.TimestampTransform convertTimestampTransform(TimestampTransform transform)
{
+      if (transform instanceof TimestampTransform.Delay) {
+        return RunnerApi.TimestampTransform.newBuilder()
+            .setDelay(
+                RunnerApi.TimestampTransform.Delay.newBuilder()
+                    .setDelayMillis(((TimestampTransform.Delay) transform).getDelay().getMillis()))
+            .build();
+      } else if (transform instanceof TimestampTransform.AlignTo) {
+        TimestampTransform.AlignTo alignTo = (TimestampTransform.AlignTo) transform;
+        return RunnerApi.TimestampTransform.newBuilder()
+            .setAlignTo(
+                RunnerApi.TimestampTransform.AlignTo.newBuilder()
+                    .setPeriod(alignTo.getPeriod().getMillis())
+                    .setOffset(alignTo.getOffset().getMillis()))
+            .build();
+
+      } else {
+        throw new IllegalArgumentException(
+            String.format("Unknown %s: %s", TimestampTransform.class.getSimpleName(), transform));
+      }
+    }
+  }
+
+  public static Trigger fromProto(RunnerApi.Trigger triggerProto) {
+    switch (triggerProto.getTriggerCase()) {
+      case AFTER_ALL:
+        return AfterAll.of(protosToTriggers(triggerProto.getAfterAll().getSubtriggersList()));
+      case AFTER_ANY:
+        return AfterFirst.of(protosToTriggers(triggerProto.getAfterAny().getSubtriggersList()));
+      case AFTER_EACH:
+        return AfterEach.inOrder(
+            protosToTriggers(triggerProto.getAfterEach().getSubtriggersList()));
+      case AFTER_END_OF_WIDOW:
+        RunnerApi.Trigger.AfterEndOfWindow eowProto = triggerProto.getAfterEndOfWidow();
+
+        if (!eowProto.hasEarlyFirings() && !eowProto.hasLateFirings()) {
+          return AfterWatermark.pastEndOfWindow();
+        }
+
+        // It either has early or late firings or both; our typing in Java makes this a smidge
+        // annoying
+        if (triggerProto.getAfterEndOfWidow().hasEarlyFirings()) {
+          AfterWatermarkEarlyAndLate trigger =
+              AfterWatermark.pastEndOfWindow()
+                  .withEarlyFirings(
+                      (OnceTrigger)
+                          fromProto(triggerProto.getAfterEndOfWidow().getEarlyFirings()));
+
+          if (triggerProto.getAfterEndOfWidow().hasLateFirings()) {
+            trigger =
+                trigger.withLateFirings(
+                    (OnceTrigger)
+                        fromProto(triggerProto.getAfterEndOfWidow().getLateFirings()));
+          }
+          return trigger;
+        } else {
+          // only late firings, so return directly
+          return AfterWatermark.pastEndOfWindow()
+              .withLateFirings((OnceTrigger) fromProto(eowProto.getLateFirings()));
+        }
+      case AFTER_PROCESSING_TIME:
+        AfterProcessingTime trigger = AfterProcessingTime.pastFirstElementInPane();
+        for (RunnerApi.TimestampTransform transform :
+            triggerProto.getAfterProcessingTime().getTimestampTransformsList()) {
+          switch (transform.getTimestampTransformCase()) {
+            case ALIGN_TO:
+              trigger =
+                  trigger.alignedTo(
+                      Duration.millis(transform.getAlignTo().getPeriod()),
+                      new Instant(transform.getAlignTo().getOffset()));
+              break;
+            case DELAY:
+              trigger = trigger.plusDelayOf(Duration.millis(transform.getDelay().getDelayMillis()));
+              break;
+            case TIMESTAMPTRANSFORM_NOT_SET:
+              throw new IllegalArgumentException(
+                  String.format(
+                      "Required field 'timestamp_transform' not set in %s", transform));
+            default:
+              throw new IllegalArgumentException(
+                  String.format(
+                      "Unknown timestamp transform case: %s",
+                      transform.getTimestampTransformCase()));
+          }
+        }
+        return trigger;
+      case AFTER_SYNCHRONIZED_PROCESSING_TIME:
+        return AfterSynchronizedProcessingTime.ofFirstElement();
+      case ELEMENT_COUNT:
+        return AfterPane.elementCountAtLeast(triggerProto.getElementCount().getElementCount());
+      case NEVER:
+        return Never.ever();
+      case OR_FINALLY:
+        return fromProto(triggerProto.getOrFinally().getMain())
+            .orFinally((OnceTrigger) fromProto(triggerProto.getOrFinally().getFinally()));
+      case REPEAT:
+        return Repeatedly.forever(fromProto(triggerProto.getRepeat().getSubtrigger()));
+      case DEFAULT:
+        return DefaultTrigger.of();
+      case TRIGGER_NOT_SET:
+        throw new IllegalArgumentException(
+            String.format("Required field 'trigger' not set in %s", triggerProto));
+      default:
+        throw new IllegalArgumentException(
+            String.format("Unknown trigger case: %s", triggerProto.getTriggerCase()));
+    }
+  }
+
+  private static List<Trigger> protosToTriggers(List<RunnerApi.Trigger> triggers)
{
+    List<Trigger> result = Lists.newArrayList();
+    for (RunnerApi.Trigger trigger : triggers) {
+      result.add(fromProto(trigger));
+    }
+    return result;
+  }
+
+  // Do not instantiate
+  private Triggers() {}
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/f4ceaeef/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/TriggersTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/TriggersTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/TriggersTest.java
new file mode 100644
index 0000000..0ac5966
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/TriggersTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.transforms.windowing;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
+
+/** Tests for utilities in {@link Triggers}. */
+@RunWith(Parameterized.class)
+public class TriggersTest {
+
+  @AutoValue
+  abstract static class ToProtoAndBackSpec {
+    abstract Trigger getTrigger();
+  }
+
+  private static ToProtoAndBackSpec toProtoAndBackSpec(Trigger trigger) {
+    return new AutoValue_TriggersTest_ToProtoAndBackSpec(trigger);
+  }
+
+  @Parameters(name = "{index}: {0}")
+  public static Iterable<ToProtoAndBackSpec> data() {
+    return ImmutableList.of(
+        // Atomic triggers
+        toProtoAndBackSpec(AfterWatermark.pastEndOfWindow()),
+        toProtoAndBackSpec(AfterPane.elementCountAtLeast(73)),
+        toProtoAndBackSpec(AfterSynchronizedProcessingTime.ofFirstElement()),
+        toProtoAndBackSpec(Never.ever()),
+        toProtoAndBackSpec(DefaultTrigger.of()),
+        toProtoAndBackSpec(AfterProcessingTime.pastFirstElementInPane()),
+        toProtoAndBackSpec(
+            AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(23))),
+        toProtoAndBackSpec(
+            AfterProcessingTime.pastFirstElementInPane()
+                .alignedTo(Duration.millis(5), new Instant(27))),
+        toProtoAndBackSpec(
+            AfterProcessingTime.pastFirstElementInPane()
+                .plusDelayOf(Duration.standardSeconds(3))
+                .alignedTo(Duration.millis(5), new Instant(27))
+                .plusDelayOf(Duration.millis(13))),
+
+        // Composite triggers
+
+        toProtoAndBackSpec(
+            AfterAll.of(AfterPane.elementCountAtLeast(79), AfterWatermark.pastEndOfWindow())),
+        toProtoAndBackSpec(
+            AfterEach.inOrder(AfterPane.elementCountAtLeast(79), AfterPane.elementCountAtLeast(3))),
+        toProtoAndBackSpec(
+            AfterFirst.of(AfterWatermark.pastEndOfWindow(), AfterPane.elementCountAtLeast(3))),
+        toProtoAndBackSpec(
+            AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(3))),
+        toProtoAndBackSpec(
+            AfterWatermark.pastEndOfWindow().withLateFirings(AfterPane.elementCountAtLeast(3))),
+        toProtoAndBackSpec(
+            AfterWatermark.pastEndOfWindow()
+                .withEarlyFirings(
+                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(42)))
+                .withLateFirings(AfterPane.elementCountAtLeast(3))),
+        toProtoAndBackSpec(Repeatedly.forever(AfterWatermark.pastEndOfWindow())),
+        toProtoAndBackSpec(
+            Repeatedly.forever(AfterPane.elementCountAtLeast(1))
+                .orFinally(AfterWatermark.pastEndOfWindow())));
+  }
+
+  @Parameter(0)
+  public ToProtoAndBackSpec toProtoAndBackSpec;
+
+  @Test
+  public void testToProtoAndBack() throws Exception {
+    Trigger trigger = toProtoAndBackSpec.getTrigger();
+    Trigger toProtoAndBackTrigger = Triggers.fromProto(Triggers.toProto(trigger));
+
+    assertThat(toProtoAndBackTrigger, equalTo(trigger));
+  }
+}


Mime
View raw message