| /* |
| * 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.extensions.euphoria.core.docs; |
| |
| import static java.util.Arrays.asList; |
| |
| import java.io.Serializable; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.stream.Stream; |
| import org.apache.beam.sdk.Pipeline; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.functional.UnaryFunction; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.io.Collector; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.AssignEventTime; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.CompositeOperator; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.CountByKey; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.Distinct; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.Filter; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.FlatMap; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.FullJoin; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.Join; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.LeftJoin; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.MapElements; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.ReduceByKey; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.ReduceWindow; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.RightJoin; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.SumByKey; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.TopPerKey; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.Union; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.operator.base.Operator; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.util.Fold; |
| import org.apache.beam.sdk.extensions.euphoria.core.client.util.Triple; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.BroadcastHashJoinTranslator; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.CompositeOperatorTranslator; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.FlatMapTranslator; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.OperatorTranslator; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.TranslatorProvider; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.provider.CompositeProvider; |
| import org.apache.beam.sdk.extensions.euphoria.core.translate.provider.GenericTranslatorProvider; |
| import org.apache.beam.sdk.extensions.kryo.KryoCoderProvider; |
| import org.apache.beam.sdk.extensions.kryo.KryoOptions; |
| import org.apache.beam.sdk.io.TextIO; |
| import org.apache.beam.sdk.options.PipelineOptions; |
| import org.apache.beam.sdk.options.PipelineOptionsFactory; |
| 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.transforms.windowing.DefaultTrigger; |
| import org.apache.beam.sdk.transforms.windowing.FixedWindows; |
| import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; |
| import org.apache.beam.sdk.transforms.windowing.Window.OnTimeBehavior; |
| import org.apache.beam.sdk.values.KV; |
| import org.apache.beam.sdk.values.PCollection; |
| import org.apache.beam.sdk.values.TypeDescriptor; |
| import org.apache.beam.sdk.values.TypeDescriptors; |
| import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Splitter; |
| import org.joda.time.Duration; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Ignore; |
| import org.junit.Rule; |
| import org.junit.Test; |
| |
| /** |
| * Contains all the examples from documentation page. Not all of them contains asserts, some do, but |
| * the rest is often here just to confirm that they compile. Once something break or changes, the |
| * documentation needs to change too. |
| */ |
| public class DocumentationExamplesTest { |
| private List<String> textLineByLine = |
| Arrays.asList( |
| "Lorem ipsum dolor sit amet, consectetur adipiscing elit. ", |
| "Vestibulum volutpat pellentesque risus at sodales.", |
| "Interdum et malesuada fames ac ante ipsum primis in faucibus.", |
| "Donec sit amet arcu nec tellus sodales ultricies.", |
| "Quisque ipsum fermentum nisl at libero accumsan consectetur.", |
| "Praesent lobortis ex eget ex rhoncus, quis malesuada risus tristique.", |
| "Aliquam risus at orci, porttitor eu turpis et, porttitor semper ligula."); |
| |
| @Rule public final TestPipeline pipeline = TestPipeline.create(); |
| |
| @Before |
| public void setup() { |
| KryoCoderProvider.of(k -> {}).registerTo(pipeline); |
| } |
| |
| @Ignore("We do not want to actually write output files from this test.") |
| @Test |
| public void wordCountExample() { |
| final PipelineOptions options = PipelineOptionsFactory.create(); |
| |
| Pipeline pipeline = Pipeline.create(options); |
| |
| // Use Kryo as coder fallback |
| KryoCoderProvider.of().registerTo(pipeline); |
| |
| // Source of data loaded from Beam IO. |
| PCollection<String> lines = |
| pipeline |
| .apply(Create.of(textLineByLine)) |
| .setTypeDescriptor(TypeDescriptor.of(String.class)); |
| |
| // FlatMap processes one input element at a time and allows user code to emit |
| // zero, one, or more output elements. From input lines we will get data set of words. |
| PCollection<String> words = |
| FlatMap.named("TOKENIZER") |
| .of(lines) |
| .using( |
| (String line, Collector<String> context) -> { |
| for (String word : Splitter.onPattern("\\s+").split(line)) { |
| context.collect(word); |
| } |
| }) |
| .output(); |
| |
| // Now we can count input words - the operator ensures that all values for the same |
| // key (word in this case) end up being processed together. Then it counts number of appearances |
| // of the same key in 'words' dataset and emits it to output. |
| PCollection<KV<String, Long>> counted = |
| CountByKey.named("COUNT").of(words).keyBy(w -> w).output(); |
| |
| // Format output. |
| PCollection<String> output = |
| MapElements.named("FORMAT") |
| .of(counted) |
| .using(p -> p.getKey() + ": " + p.getValue()) |
| .output(); |
| |
| // Now we can again use Beam transformation. In this case we save words and their count |
| // into the text file. |
| output.apply(TextIO.write().to("counted_words")); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void inputsAndOutputsSection() { |
| |
| PCollection<String> input = |
| pipeline |
| .apply(Create.of("mouse", "rat", "elephant", "cat", "X", "duck")) |
| .setTypeDescriptor(TypeDescriptor.of(String.class)); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void addOperatorSection() { |
| PCollection<Integer> input = pipeline.apply(Create.of(1, 2, 4, 3)); |
| |
| PCollection<String> mappedElements = |
| MapElements.named("Int2Str").of(input).using(String::valueOf).output(); |
| |
| PAssert.that(mappedElements).containsInAnyOrder("1", "2", "4", "3"); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void metricsAndAccumulatorsSection() { |
| final PipelineOptions options = PipelineOptionsFactory.create(); |
| Pipeline pipeline = Pipeline.create(options); |
| |
| PCollection<String> dataset = pipeline.apply(Create.of("a", "x")); |
| |
| PCollection<String> flatMapped = |
| FlatMap.named("FlatMap1") |
| .of(dataset) |
| .using( |
| (String value, Collector<String> context) -> { |
| context.getCounter("my-counter").increment(); |
| context.collect(value); |
| }) |
| .output(); |
| |
| PCollection<String> mapped = |
| MapElements.named("MapThem") |
| .of(dataset) |
| .using( |
| (value, context) -> { |
| // use simple counter |
| context.getCounter("my-counter").increment(); |
| |
| return value.toLowerCase(); |
| }) |
| .output(); |
| } |
| |
| @Test |
| public void codersAndTypesSection() { |
| final PipelineOptions options = PipelineOptionsFactory.create(); |
| Pipeline pipeline = Pipeline.create(options); |
| |
| // Register `KryoCoderProvider` which attempt to use `KryoCoder` to every non-primitive type |
| KryoCoderProvider.of().registerTo(pipeline); |
| |
| // Do not allow `KryoCoderProvider` to return `KryoCoder` for unregistered types |
| options.as(KryoOptions.class).setKryoRegistrationRequired(true); |
| |
| KryoCoderProvider.of( |
| kryo -> { // KryoRegistrar of your uwn |
| kryo.register(KryoSerializedElementType.class); // other may follow |
| }) |
| .registerTo(pipeline); |
| |
| PCollection<Integer> input = |
| pipeline.apply(Create.of(1, 2, 3, 4)).setTypeDescriptor(TypeDescriptors.integers()); |
| |
| MapElements.named("Int2Str") |
| .of(input) |
| .using(String::valueOf, TypeDescriptors.strings()) |
| .output(); |
| } |
| |
| @Test |
| public void windowingSection() { |
| |
| PCollection<Integer> input = |
| pipeline.apply(Create.of(1, 2, 3, 4)).setTypeDescriptor(TypeDescriptors.integers()); |
| |
| PCollection<KV<Integer, Long>> countedElements = |
| CountByKey.of(input) |
| .keyBy(e -> e) |
| .windowBy(FixedWindows.of(Duration.standardSeconds(1))) |
| .triggeredBy(DefaultTrigger.of()) |
| .discardingFiredPanes() |
| .withAllowedLateness(Duration.standardSeconds(5)) |
| .withOnTimeBehavior(OnTimeBehavior.FIRE_IF_NON_EMPTY) |
| .withTimestampCombiner(TimestampCombiner.EARLIEST) |
| .output(); |
| |
| pipeline.run(); |
| } |
| |
| private static class KryoSerializedElementType {} |
| |
| @Test |
| public void countByKeyOperator() { |
| |
| PCollection<Integer> input = pipeline.apply(Create.of(1, 2, 4, 1, 1, 3)); |
| |
| // suppose input: [1, 2, 4, 1, 1, 3] |
| PCollection<KV<Integer, Long>> output = CountByKey.of(input).keyBy(e -> e).output(); |
| // Output will contain: [KV(1, 3), KV(2, 1), KV(3, 1), (4, 1)] |
| |
| PAssert.that(output) |
| .containsInAnyOrder(asList(KV.of(1, 3L), KV.of(2, 1L), KV.of(3, 1L), KV.of(4, 1L))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void distinctOperator() { |
| |
| PCollection<Integer> input = pipeline.apply("input", Create.of(1, 2, 3, 3, 2, 1)); |
| |
| // suppose input: [1, 2, 3, 3, 2, 1] |
| Distinct.named("unique-integers-only").of(input).output(); |
| // Output will contain: 1, 2, 3 |
| |
| PCollection<KV<Integer, Long>> keyValueInput = |
| pipeline.apply( |
| "keyValueInput", |
| Create.of( |
| KV.of(1, 100L), KV.of(3, 100_000L), KV.of(42, 10L), KV.of(1, 0L), KV.of(3, 0L))); |
| |
| // suppose input: [KV(1, 100L), KV(3, 100_000L), KV(42, 10L), KV(1, 0L), KV(3, 0L)] |
| PCollection<KV<Integer, Long>> distinct = |
| Distinct.named("unique-keys-only").of(keyValueInput).projected(KV::getKey).output(); |
| |
| // Output will contain: 1, 3, 42 |
| PCollection<Integer> uniqueKeys = MapElements.of(distinct).using(KV::getKey).output(); |
| |
| PAssert.that(uniqueKeys).containsInAnyOrder(1, 3, 42); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void batchJoinOperator() { |
| |
| PCollection<Integer> left = |
| pipeline |
| .apply("left", Create.of(1, 2, 3, 0, 4, 3, 1)) |
| .setTypeDescriptor(TypeDescriptors.integers()); |
| PCollection<String> right = |
| pipeline |
| .apply("right", Create.of("mouse", "rat", "elephant", "cat", "X", "duck")) |
| .setTypeDescriptor(TypeDescriptors.strings()); |
| |
| // suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| // suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, String>> joined = |
| Join.named("join-length-to-words") |
| .of(left, right) |
| .by(le -> le, String::length) // key extractors |
| .using((Integer l, String r, Collector<String> c) -> c.collect(l + "+" + r)) |
| .output(); |
| |
| // joined will contain: [ KV(1, "1+X"), KV(3, "3+cat"), KV(3, "3+rat"), KV(4, "4+duck"), |
| // KV(3, "3+cat"), KV(3, "3+rat"), KV(1, "1+X")] |
| |
| PAssert.that(joined) |
| .containsInAnyOrder( |
| asList( |
| KV.of(1, "1+X"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(4, "4+duck"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(1, "1+X"))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void batchLeftJoinOperator() { |
| |
| PCollection<Integer> left = |
| pipeline |
| .apply("left", Create.of(1, 2, 3, 0, 4, 3, 1)) |
| .setTypeDescriptor(TypeDescriptors.integers()); |
| PCollection<String> right = |
| pipeline |
| .apply("right", Create.of("mouse", "rat", "elephant", "cat", "X", "duck")) |
| .setTypeDescriptor(TypeDescriptors.strings()); |
| |
| // suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| // suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, String>> joined = |
| LeftJoin.named("left-join-length-to-words") |
| .of(left, right) |
| .by(le -> le, String::length) // key extractors |
| .using( |
| (Integer l, Optional<String> r, Collector<String> c) -> |
| c.collect(l + "+" + r.orElse(null))) |
| .output(); |
| |
| // joined will contain: [KV(1, "1+X"), KV(2, "2+null"), KV(3, "3+cat"), |
| // KV(3, "3+rat"), KV(0, "0+null"), KV(4, "4+duck"), KV(3, "3+cat"), |
| // KV(3, "3+rat"), KV(1, "1+X")] |
| |
| PAssert.that(joined) |
| .containsInAnyOrder( |
| asList( |
| KV.of(1, "1+X"), |
| KV.of(2, "2+null"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(0, "0+null"), |
| KV.of(4, "4+duck"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(1, "1+X"))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void batchRightJoinFullOperator() { |
| |
| PCollection<Integer> left = |
| pipeline |
| .apply("left", Create.of(1, 2, 3, 0, 4, 3, 1)) |
| .setTypeDescriptor(TypeDescriptors.integers()); |
| PCollection<String> right = |
| pipeline |
| .apply("right", Create.of("mouse", "rat", "elephant", "cat", "X", "duck")) |
| .setTypeDescriptor(TypeDescriptors.strings()); |
| |
| // suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| // suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, String>> joined = |
| RightJoin.named("right-join-length-to-words") |
| .of(left, right) |
| .by(le -> le, String::length) // key extractors |
| .using( |
| (Optional<Integer> l, String r, Collector<String> c) -> |
| c.collect(l.orElse(null) + "+" + r)) |
| .output(); |
| |
| // joined will contain: [ KV(1, "1+X"), KV(3, "3+cat"), KV(3, "3+rat"), |
| // KV(4, "4+duck"), KV(3, "3+cat"), KV(3, "3+rat"), KV(1, "1+X"), |
| // KV(8, "null+elephant"), KV(5, "null+mouse")] |
| |
| PAssert.that(joined) |
| .containsInAnyOrder( |
| asList( |
| KV.of(1, "1+X"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(4, "4+duck"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(1, "1+X"), |
| KV.of(8, "null+elephant"), |
| KV.of(5, "null+mouse"))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void batchFullJoinOperator() { |
| |
| PCollection<Integer> left = |
| pipeline |
| .apply("left", Create.of(1, 2, 3, 0, 4, 3, 1)) |
| .setTypeDescriptor(TypeDescriptors.integers()); |
| PCollection<String> right = |
| pipeline |
| .apply("right", Create.of("mouse", "rat", "elephant", "cat", "X", "duck")) |
| .setTypeDescriptor(TypeDescriptors.strings()); |
| |
| // suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| // suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, String>> joined = |
| FullJoin.named("join-length-to-words") |
| .of(left, right) |
| .by(le -> le, String::length) // key extractors |
| .using( |
| (Optional<Integer> l, Optional<String> r, Collector<String> c) -> |
| c.collect(l.orElse(null) + "+" + r.orElse(null))) |
| .output(); |
| |
| // joined will contain: [ KV(1, "1+X"), KV(2, "2+null"), KV(3, "3+cat"), KV(3, "3+rat"), |
| // KV(0, "0+null"), KV(4, "4+duck"), KV(3, "3+cat"), KV(3, "3+rat"),KV(1, "1+X"), |
| // KV(1, "null+elephant"), KV(5, "null+mouse")]; |
| |
| PAssert.that(joined) |
| .containsInAnyOrder( |
| asList( |
| KV.of(1, "1+X"), |
| KV.of(2, "2+null"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(0, "0+null"), |
| KV.of(4, "4+duck"), |
| KV.of(3, "3+cat"), |
| KV.of(3, "3+rat"), |
| KV.of(1, "1+X"), |
| KV.of(8, "null+elephant"), |
| KV.of(5, "null+mouse"))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void mapElementsOperator() { |
| |
| PCollection<Integer> input = |
| pipeline.apply(Create.of(0, 1, 2, 3, 4, 5)).setTypeDescriptor(TypeDescriptors.integers()); |
| |
| // suppose inputs contains: [ 0, 1, 2, 3, 4, 5] |
| PCollection<String> strings = |
| MapElements.named("int2str").of(input).using(i -> "#" + i).output(); |
| // strings will contain: [ "#0", "#1", "#2", "#3", "#4", "#5"] |
| |
| PAssert.that(strings).containsInAnyOrder("#0", "#1", "#2", "#3", "#4", "#5"); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void flatMapOperator() { |
| |
| PCollection<String> words = pipeline.apply(Create.of(asList("Brown", "fox", ".", ""))); |
| |
| // suppose words contain: ["Brown", "fox", ".", ""] |
| PCollection<String> letters = |
| FlatMap.named("str2char") |
| .of(words) |
| .using( |
| (String s, Collector<String> collector) -> { |
| for (int i = 0; i < s.length(); i++) { |
| char c = s.charAt(i); |
| collector.collect(String.valueOf(c)); |
| } |
| }) |
| .output(); |
| // characters will contain: ["B", "r", "o", "w", "n", "f", "o", "x", "."] |
| |
| PAssert.that(letters).containsInAnyOrder("B", "r", "o", "w", "n", "f", "o", "x", "."); |
| pipeline.run(); |
| } |
| |
| @Test |
| public void flatMapWithTimeExtractorOperator() { |
| |
| PCollection<SomeEventObject> events = |
| pipeline.apply( |
| Create.of( |
| new SomeEventObject(0), |
| new SomeEventObject(1), |
| new SomeEventObject(2), |
| new SomeEventObject(3), |
| new SomeEventObject(4))); |
| |
| // suppose events contain events of SomeEventObject, its 'getEventTimeInMillis()' methods |
| // returns time-stamp |
| PCollection<SomeEventObject> timeStampedEvents = |
| FlatMap.named("extract-event-time") |
| .of(events) |
| .using((SomeEventObject e, Collector<SomeEventObject> c) -> c.collect(e)) |
| .eventTimeBy(SomeEventObject::getEventTimeInMillis) |
| .output(); |
| // Euphoria will now know event time for each event |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void filterOperator() { |
| |
| PCollection<Integer> nums = pipeline.apply(Create.of(asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))); |
| |
| // suppose nums contains: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] |
| PCollection<Integer> divisibleBythree = |
| Filter.named("divisibleByFive").of(nums).by(e -> e % 3 == 0).output(); |
| // divisibleBythree will contain: [ 0, 3, 6, 9] |
| |
| PAssert.that(divisibleBythree).containsInAnyOrder(0, 3, 6, 9); |
| pipeline.run(); |
| } |
| |
| @Test |
| public void reduceByKeyTestOperator1() { |
| |
| PCollection<String> animals = |
| pipeline.apply(Create.of("mouse", "rat", "elephant", "cat", "X", "duck")); |
| |
| // suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, Long>> countOfAnimalNamesByLength = |
| ReduceByKey.named("to-letters-couts") |
| .of(animals) |
| .keyBy(String::length) // length of animal name will be used as groupping key |
| // we need to count each animal name once, so why not to optimize each string to 1 |
| .valueBy(e -> 1) |
| .reduceBy(Stream::count) |
| .output(); |
| // countOfAnimalNamesByLength wil contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, |
| // 1L), KV.of(8, 1L) ] |
| |
| PAssert.that(countOfAnimalNamesByLength) |
| .containsInAnyOrder( |
| asList(KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void reduceByKeyTestOperatorCombinable() { |
| |
| PCollection<String> animals = |
| pipeline.apply(Create.of("mouse", "rat", "elephant", "cat", "X", "duck")); |
| |
| // suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, Long>> countOfAnimalNamesByLength = |
| ReduceByKey.named("to-letters-couts") |
| .of(animals) |
| .keyBy(String::length) // length of animal name will be used as grouping key |
| // we need to count each animal name once, so why not to optimize each string to 1 |
| .valueBy(e -> 1L) |
| .combineBy(s -> s.mapToLong(l -> l).sum()) |
| .output(); |
| // countOfAnimalNamesByLength wil contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, |
| // 1L), KV.of(8, 1L) ] |
| |
| PAssert.that(countOfAnimalNamesByLength) |
| .containsInAnyOrder( |
| asList(KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void reduceByKeyTestOperatorContext() { |
| |
| PCollection<String> animals = |
| pipeline.apply(Create.of("mouse", "rat", "elephant", "cat", "X", "duck")); |
| |
| // suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, Long>> countOfAnimalNamesByLength = |
| ReduceByKey.named("to-letters-couts") |
| .of(animals) |
| .keyBy(String::length) // length of animal name will e used as grouping key |
| // we need to count each animal name once, so why not to optimize each string to 1 |
| .valueBy(e -> 1) |
| .reduceBy( |
| (Stream<Integer> s, Collector<Long> collector) -> { |
| collector.collect(s.count()); |
| collector.asContext().getCounter("num-of-keys").increment(); |
| }) |
| .output(); |
| // countOfAnimalNamesByLength wil contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, |
| // 1L), KV.of(8, 1L) ] |
| |
| PAssert.that(countOfAnimalNamesByLength) |
| .containsInAnyOrder( |
| asList(KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L))); |
| |
| pipeline.run(); |
| } |
| |
| /** |
| * Note that this one is not mentioned in documentation due to high number of RBK examples and |
| * rather lower explanation value. Please consider to include it in future |
| */ |
| @Test |
| public void reduceByKeyTestOperatorContextManyOutputs() { |
| |
| PCollection<String> animals = |
| pipeline.apply(Create.of("mouse", "rat", "elephant", "cat", "X", "duck")); |
| |
| PCollection<KV<Integer, Long>> countOfAnimalNamesByLength = |
| ReduceByKey.named("to-letters-couts") |
| .of(animals) |
| .keyBy(String::length) // length of animal name will e used as grouping key |
| // we need to count each animal name once, so why not to optimize each string to 1 |
| .valueBy(e -> 1) |
| .reduceBy( |
| (Stream<Integer> s, Collector<Long> collector) -> { |
| long count = s.count(); |
| collector.collect(count); |
| collector.collect(2L * count); |
| }) |
| .output(); |
| |
| PAssert.that(countOfAnimalNamesByLength) |
| .containsInAnyOrder( |
| asList( |
| KV.of(1, 1L), |
| KV.of(3, 2L), |
| KV.of(4, 1L), |
| KV.of(5, 1L), |
| KV.of(8, 1L), |
| KV.of(1, 2L), |
| KV.of(3, 4L), |
| KV.of(4, 2L), |
| KV.of(5, 2L), |
| KV.of(8, 2L))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void reduceByKeyTestOperatorFold() { |
| |
| PCollection<String> animals = |
| pipeline.apply(Create.of("mouse", "rat", "elephant", "cat", "X", "duck")); |
| |
| // suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| PCollection<KV<Integer, Long>> countOfAnimalNamesByLength = |
| ReduceByKey.named("to-letters-couts") |
| .of(animals) |
| .keyBy(String::length) // length of animal name will be used as grouping key |
| // we need to count each animal name once, so why not to optimize each string to 1 |
| .valueBy(e -> 1L) |
| .combineBy(Fold.of((l1, l2) -> l1 + l2)) |
| .output(); |
| // countOfAnimalNamesByLength will contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, |
| // 1L), KV.of(8, 1L) ] |
| |
| PAssert.that(countOfAnimalNamesByLength) |
| .containsInAnyOrder( |
| asList(KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L))); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void testSumByKeyOperator() { |
| PCollection<Integer> input = pipeline.apply(Create.of(asList(1, 2, 3, 4, 5, 6, 7, 8, 9))); |
| |
| // suppose input contains: [ 1, 2, 3, 4, 5, 6, 7, 8, 9 ] |
| PCollection<KV<Integer, Long>> output = |
| SumByKey.named("sum-odd-and-even") |
| .of(input) |
| .keyBy(e -> e % 2) |
| .valueBy(e -> (long) e) |
| .output(); |
| // output will contain: [ KV.of(0, 20L), KV.of(1, 25L)] |
| |
| PAssert.that(output).containsInAnyOrder(asList(KV.of(0, 20L), KV.of(1, 25L))); |
| pipeline.run(); |
| } |
| |
| @Test |
| public void testUnionOperator() { |
| |
| PCollection<String> cats = |
| pipeline |
| .apply("cats", Create.of(asList("cheetah", "cat", "lynx", "jaguar"))) |
| .setTypeDescriptor(TypeDescriptors.strings()); |
| |
| PCollection<String> rodents = |
| pipeline |
| .apply("rodents", Create.of("squirrel", "mouse", "rat", "lemming", "beaver")) |
| .setTypeDescriptor(TypeDescriptors.strings()); |
| |
| // suppose cats contains: [ "cheetah", "cat", "lynx", "jaguar" ] |
| // suppose rodents contains: [ "squirrel", "mouse", "rat", "lemming", "beaver" ] |
| PCollection<String> animals = Union.named("to-animals").of(cats, rodents).output(); |
| |
| // animal will contain: "cheetah", "cat", "lynx", "jaguar", "squirrel", "mouse", "rat", |
| // "lemming", "beaver" |
| PAssert.that(animals) |
| .containsInAnyOrder( |
| "cheetah", "cat", "lynx", "jaguar", "squirrel", "mouse", "rat", "lemming", "beaver"); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void testAssignEventTimeOperator() { |
| |
| PCollection<SomeEventObject> events = |
| pipeline.apply( |
| Create.of( |
| asList( |
| new SomeEventObject(0), |
| new SomeEventObject(1), |
| new SomeEventObject(2), |
| new SomeEventObject(3), |
| new SomeEventObject(4)))); |
| |
| // suppose events contain events of SomeEventObject, its 'getEventTimeInMillis()' methods |
| // returns time-stamp |
| PCollection<SomeEventObject> timeStampedEvents = |
| AssignEventTime.named("extract-event-time") |
| .of(events) |
| .using(SomeEventObject::getEventTimeInMillis) |
| .output(); |
| // Euphoria will now know event time for each event |
| |
| pipeline.run(); |
| } |
| |
| private static class SomeEventObject implements Serializable { |
| |
| private long timestamp; |
| |
| SomeEventObject(long timestamp) { |
| this.timestamp = timestamp; |
| } |
| |
| long getEventTimeInMillis() { |
| return timestamp; |
| } |
| } |
| |
| @Test |
| public void testReduceWithWindowOperator() { |
| |
| PCollection<Integer> input = pipeline.apply(Create.of(asList(1, 2, 3, 4, 5, 6, 7, 8))); |
| |
| // suppose input contains [ 1, 2, 3, 4, 5, 6, 7, 8 ] |
| // lets assign time-stamp to each input element |
| PCollection<Integer> withEventTime = AssignEventTime.of(input).using(i -> 1000L * i).output(); |
| |
| PCollection<Integer> output = |
| ReduceWindow.of(withEventTime) |
| .combineBy(Fold.of((i1, i2) -> i1 + i2)) |
| .windowBy(FixedWindows.of(Duration.millis(5000))) |
| .triggeredBy(DefaultTrigger.of()) |
| .discardingFiredPanes() |
| .output(); |
| // output will contain: [ 10, 26 ] |
| |
| PAssert.that(output).containsInAnyOrder(10, 26); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void testTopPerKeyOperator() { |
| |
| PCollection<String> animals = |
| pipeline.apply( |
| Create.of( |
| "mouse", |
| "elk", |
| "rat", |
| "mule", |
| "elephant", |
| "dinosaur", |
| "cat", |
| "duck", |
| "caterpillar")); |
| |
| // suppose 'animals contain: [ "mouse", "elk", "rat", "mule", "elephant", "dinosaur", "cat", |
| // "duck", "caterpillar" ] |
| PCollection<Triple<Character, String, Integer>> longestNamesByLetter = |
| TopPerKey.named("longest-animal-names") |
| .of(animals) |
| .keyBy(name -> name.charAt(0)) // first character is the key |
| .valueBy(UnaryFunction.identity()) // value type is the same as input element type |
| .scoreBy(String::length) // length defines score, note that Integer implements |
| // Comparable<Integer> |
| .output(); |
| // longestNamesByLetter will contain: [ ('m', "mouse", 5), ('r', "rat", 3), ('e', "elephant", |
| // 8), ('d', "dinosaur", 8), ('c', "caterpillar", 11) ] |
| |
| PAssert.that(longestNamesByLetter) |
| .containsInAnyOrder( |
| Triple.of('m', "mouse", 5), |
| Triple.of('r', "rat", 3), |
| Triple.of('e', "elephant", 8), |
| Triple.of('d', "dinosaur", 8), |
| Triple.of('c', "caterpillar", 11)); |
| |
| pipeline.run(); |
| } |
| |
| @Test |
| public void testGenericTranslatorProvider() { |
| |
| GenericTranslatorProvider provider = |
| GenericTranslatorProvider.newBuilder() |
| .register(FlatMap.class, new FlatMapTranslator<>()) // register by operator class |
| .register( |
| Join.class, |
| (Join op) -> { |
| String name = ((Optional<String>) op.getName()).orElse(""); |
| return name.toLowerCase().startsWith("broadcast"); |
| }, |
| new BroadcastHashJoinTranslator<>()) // register by class and predicate |
| .register( |
| op -> op instanceof CompositeOperator, |
| new CompositeOperatorTranslator<>()) // register by predicate only |
| .build(); |
| |
| Assert.assertNotNull(provider); |
| } |
| |
| private static class CustomTranslatorProvider implements TranslatorProvider { |
| |
| static CustomTranslatorProvider of() { |
| return new CustomTranslatorProvider(); |
| } |
| |
| @Override |
| public <InputT, OutputT, OperatorT extends Operator<OutputT>> |
| Optional<OperatorTranslator<InputT, OutputT, OperatorT>> findTranslator( |
| OperatorT operator) { |
| return Optional.empty(); |
| } |
| } |
| |
| @Test |
| public void testCompositeTranslationProviderExample() { |
| |
| CompositeProvider compositeProvider = |
| CompositeProvider.of( |
| CustomTranslatorProvider.of(), // first ask CustomTranslatorProvider for translator |
| GenericTranslatorProvider |
| .createWithDefaultTranslators()); // then ask default provider if needed |
| |
| Assert.assertNotNull(compositeProvider); |
| } |
| } |
