Add "Early Triggers" Java kata

learning/katas/java/Triggers/Early Triggers/Early Triggers/src/GenerateEvent.java

@@ -0,0 +1,40 @@
+/*
+ * 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:https://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.
+ */
+
+import static org.apache.beam.sdk.values.TypeDescriptors.strings;
+
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+
+public class GenerateEvent extends PTransform<PBegin, PCollection<String>> {
+
+ static GenerateEvent everySecond() {
+ return new GenerateEvent();
+ }
+
+ public PCollection<String> expand(PBegin input) {
+ return input
+ .apply(GenerateSequence.from(1).withRate(1, Duration.standardSeconds(1)))
+ .apply(MapElements.into(strings()).via(num -> "event"));
+ }
+
+}

learning/katas/java/Triggers/Early Triggers/Early Triggers/src/Task.java

@@ -0,0 +1,62 @@
+/*
+ * 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:https://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.
+ */
+
+import org.apache.beam.learning.katas.util.Log;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+
+public class Task {
+
+ public static void main(String[] args) {
+ PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
+ Pipeline pipeline = Pipeline.create(options);
+
+ PCollection<String> events =
+ pipeline.apply(GenerateEvent.everySecond());
+
+ PCollection<Long> output = applyTransform(events);
+
+ output.apply(Log.ofElements());
+
+ pipeline.run();
+ }
+
+ static PCollection<Long> applyTransform(PCollection<String> events) {
+ return events
+ .apply(
+ Window.<String>into(FixedWindows.of(Duration.standardDays(1)))
+ .triggering(
+ AfterWatermark.pastEndOfWindow()
+ .withEarlyFirings(
+ AfterProcessingTime.pastFirstElementInPane()))
+ .withAllowedLateness(Duration.ZERO)
+ .discardingFiredPanes())
+
+ .apply(Combine.globally(Count.<String>combineFn()).withoutDefaults());
+ }
+
+}

learning/katas/java/Triggers/Early Triggers/Early Triggers/task.html

@@ -0,0 +1,56 @@
+<!--
+ ~ 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:https://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.
+ -->
+
+<html>
+<h2>Early Triggers</h2>
+<p>
+ Triggers allow Beam to emit early results, before all the data in a given window has arrived.
+ For example, emitting after a certain amount of time elapses, or after a certain number of
+ elements arrives.
+</p>
+<p>
+ <b>Kata:</b> Given that events are being generated every second and a fixed window of 1-day
+ duration, please implement an early trigger that emits the number of events count immediately
+ after new element is processed.
+</p>
+<br>
+<div class="hint">
+ Use <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/windowing/AfterWatermark.AfterWatermarkEarlyAndLate.html#withEarlyFirings-org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger-">
+ withEarlyFirings</a> to set early firing triggers.
+</div>
+<div class="hint">
+ Use <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/windowing/FixedWindows.html">
+ FixedWindows</a> with 1-day duration using
+ <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/windowing/AfterWatermark.html#pastEndOfWindow--">
+ AfterWatermark.pastEndOfWindow()</a> trigger.
+</div>
+<div class="hint">
+ Set the allowed lateness to 0 with discarding accumulation mode.
+</div>
+<div class="hint">
+ Use <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/Combine.html#globally-org.apache.beam.sdk.transforms.CombineFnBase.GlobalCombineFn-">
+ Combine.globally</a> and
+ <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/Count.html#combineFn--">
+ Count.combineFn</a> to calculate the count of events.
+</div>
+<div class="hint">
+ Refer to the Beam Programming Guide
+ <a href="https://beam.apache.org/documentation/programming-guide/#event-time-triggers">
+ "Event time triggers"</a> section for more information.
+</div>
+</html>