Add composite transform Java kata

mransley · Jun 9, 2019 · ac6faff · ac6faff
1 parent 29f0553
commit ac6faff
Show file tree

Hide file tree

Showing 4 changed files with 252 additions and 0 deletions.
diff --git a/learning/katas/java/.idea/study_project.xml b/learning/katas/java/.idea/study_project.xml
diff --git a/...Composite Transform/src/org/apache/beam/learning/katas/coretransforms/composite/Task.java b/...Composite Transform/src/org/apache/beam/learning/katas/coretransforms/composite/Task.java
@@ -0,0 +1,69 @@
+/*
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.beam.learning.katas.coretransforms.composite;
+
+import static org.apache.beam.sdk.values.TypeDescriptors.integers;
+
+import java.util.Arrays;
+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.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import util.Log;
+
+public class Task {
+
+ public static void main(String[] args) {
+ PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
+ Pipeline pipeline = Pipeline.create(options);
+
+ pipeline
+ .apply(Create.of("1,2,3,4,5", "6,7,8,9,10"))
+ .apply(new ExtractAndMultiplyNumbers())
+ .apply(Log.ofElements());
+
+ pipeline.run();
+ }
+
+ static class ExtractAndMultiplyNumbers
+ extends PTransform<PCollection<String>, PCollection<Integer>> {
+
+ @Override
+ public PCollection<Integer> expand(PCollection<String> input) {
+ return input
+ .apply(ParDo.of(new DoFn<String, Integer>() {
+
+ @ProcessElement
+ public void processElement(@Element String numbers, OutputReceiver<Integer> out) {
+ Arrays.stream(numbers.split(","))
+ .forEach(numStr -> out.output(Integer.parseInt(numStr)));
+ }
+
+ }))
+
+ .apply(MapElements.into(integers()).via(number -> number * 10));
+ }
+
+ }
+}
diff --git a/learning/katas/java/Core Transforms/Composite Transform/Composite Transform/task.html b/learning/katas/java/Core Transforms/Composite Transform/Composite Transform/task.html
@@ -0,0 +1,51 @@
+<!--
+ ~ 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
+ ~
+ ~ 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>Composite Transform</h2>
+<p>
+ Transforms can have a nested structure, where a complex transform performs multiple simpler
+ transforms (such as more than one ParDo, Combine, GroupByKey, or even other composite transforms).
+ These transforms are called composite transforms. Nesting multiple transforms inside a single
+ composite transform can make your code more modular and easier to understand.
+</p>
+<p>
+ To create your own composite transform, create a subclass of the PTransform class and override
+ the expand method to specify the actual processing logic. You can then use this transform just as
+ you would a built-in transform from the Beam SDK. For the PTransform class type parameters, you
+ pass the PCollection types that your transform takes as input, and produces as output. Within
+ your PTransform subclass, you’ll need to override the expand method. The expand method is where
+ you add the processing logic for the PTransform. Your override of expand must accept the
+ appropriate type of input PCollection as a parameter, and specify the output PCollection as the
+ return value.
+</p>
+<p>
+ <b>Kata:</b> Please implement a composite transform "ExtractAndMultiplyNumbers" that extracts
+ numbers from comma separated line and then multiplies each number by 10.
+</p>
+<br>
+<div class="hint">
+ Refer to <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/PTransform.html">
+ PTransform</a>.
+</div>
+<div class="hint">
+ Refer to the Beam Programming Guide
+ <a href="https://beam.apache.org/documentation/programming-guide/#composite-transforms">
+ "Composite transforms"</a> section for more information.
+</div>
+</html>
diff --git a/...mposite Transform/test/org/apache/beam/learning/katas/coretransforms/composite/Tests.java b/...mposite Transform/test/org/apache/beam/learning/katas/coretransforms/composite/Tests.java
@@ -0,0 +1,49 @@
+/*
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.beam.learning.katas.coretransforms.composite;
+
+import org.apache.beam.learning.katas.coretransforms.composite.Task.ExtractAndMultiplyNumbers;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class Tests {
+
+ @Rule
+ public final transient TestPipeline testPipeline = TestPipeline.create();
+
+ @Test
+ public void compositeTransform() {
+ Create.Values<String> values = Create.of("1,2,3,4,5", "6,7,8,9,10");
+
+ PCollection<Integer> results =
+ testPipeline
+ .apply(values)
+ .apply(new ExtractAndMultiplyNumbers());
+
+ PAssert.that(results)
+ .containsInAnyOrder(10, 20, 30, 40, 50, 60, 70, 80, 90, 100);
+
+ testPipeline.run().waitUntilFinish();
+ }
+
+}