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 2017/02/15 04:51:47 UTC
[5/9] beam git commit: Add conversion to/from Java SDK trigger to
runner API proto
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 <kl...@google.com>
Authored: Sat Feb 11 17:50:27 2017 -0800
Committer: Kenneth Knowles <kl...@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));
+ }
+}