| /* |
| * 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.wayang.api; |
| |
| import org.apache.wayang.core.api.WayangContext; |
| import org.apache.wayang.core.util.WayangCollections; |
| import org.apache.wayang.java.Java; |
| //import org.apache.wayang.spark.Spark; |
| //import org.apache.wayang.sqlite3.Sqlite3; |
| //import org.apache.wayang.sqlite3.operators.Sqlite3TableSource; |
| import org.junit.Assert; |
| import org.junit.Test; |
| |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.List; |
| |
| /** |
| * Test suite for the Java API. |
| */ |
| public class JavaApiTest { |
| |
| // private Configuration sqlite3Configuration; |
| // |
| // @Before |
| // public void setUp() throws SQLException, IOException { |
| // // Generate test data. |
| // this.sqlite3Configuration = new Configuration(); |
| // File sqlite3dbFile = File.createTempFile("wayang-sqlite3", "db"); |
| // sqlite3dbFile.deleteOnExit(); |
| // this.sqlite3Configuration.setProperty("wayang.sqlite3.jdbc.url", "jdbc:sqlite:" + sqlite3dbFile.getAbsolutePath()); |
| // try (Connection connection = Sqlite3.platform().createDatabaseDescriptor(this.sqlite3Configuration).createJdbcConnection()) { |
| // Statement statement = connection.createStatement(); |
| // statement.addBatch("DROP TABLE IF EXISTS customer;"); |
| // statement.addBatch("CREATE TABLE customer (name TEXT, age INT);"); |
| // statement.addBatch("INSERT INTO customer VALUES ('John', 20)"); |
| // statement.addBatch("INSERT INTO customer VALUES ('Timmy', 16)"); |
| // statement.addBatch("INSERT INTO customer VALUES ('Evelyn', 35)"); |
| // statement.executeBatch(); |
| // } |
| // } |
| |
| // @Test |
| // public void testMapReduce() { |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder javaPlanBuilder = new JavaPlanBuilder(wayangContext); |
| // |
| // List<Integer> inputCollection = Arrays.asList(0, 1, 2, 3, 4); |
| // Collection<Integer> outputCollection = javaPlanBuilder |
| // .loadCollection(inputCollection).withName("load numbers") |
| // .map(i -> i * i).withName("square") |
| // .reduce((a, b) -> a + b).withName("sum") |
| // .collect(); |
| // |
| // Assert.assertEquals(WayangCollections.asSet(1 + 4 + 9 + 16), WayangCollections.asSet(outputCollection)); |
| // } |
| |
| // @Test |
| // public void testMapReduceBy() { |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder javaPlanBuilder = new JavaPlanBuilder(wayangContext); |
| // |
| // List<Integer> inputCollection = Arrays.asList(0, 1, 2, 3, 4); |
| // Collection<Integer> outputCollection = javaPlanBuilder |
| // .loadCollection(inputCollection).withName("load numbers") |
| // .map(i -> i * i).withName("square") |
| // .reduceByKey(i -> i & 1, (a, b) -> a + b).withName("sum") |
| // .collect(); |
| // |
| // Assert.assertEquals(WayangCollections.asSet(4 + 16, 1 + 9), WayangCollections.asSet(outputCollection)); |
| // } |
| // |
| // @Test |
| // public void testBroadcast2() { |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder javaPlanBuilder = new JavaPlanBuilder(wayangContext); |
| // |
| // List<Integer> inputCollection = Arrays.asList(0, 1, 2, 3, 4); |
| // List<Integer> offsetCollection = Collections.singletonList(-2); |
| // |
| // LoadCollectionDataQuantaBuilder<Integer> offsetDataQuanta = javaPlanBuilder |
| // .loadCollection(offsetCollection) |
| // .withName("load offset"); |
| // |
| // Collection<Integer> outputCollection = javaPlanBuilder |
| // .loadCollection(inputCollection).withName("load numbers") |
| // .map(new AddOffset("offset")).withName("add offset").withBroadcast(offsetDataQuanta, "offset") |
| // .collect(); |
| // |
| // Assert.assertEquals(WayangCollections.asSet(-2, -1, 0, 1, 2), WayangCollections.asSet(outputCollection)); |
| // } |
| // |
| // @Test |
| // public void testCustomOperatorShortCut() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // |
| // final List<Integer> inputValues = WayangArrays.asList(0, 1, 2, 3); |
| // |
| // // Build and execute a Wayang plan. |
| // final Collection<Integer> outputValues = new JavaPlanBuilder(wayangContext) |
| // .loadCollection(inputValues).withName("Load input values") |
| // .<Integer>customOperator(new JavaMapOperator<>( |
| // DataSetType.createDefault(Integer.class), |
| // DataSetType.createDefault(Integer.class), |
| // new TransformationDescriptor<>( |
| // i -> i + 2, |
| // Integer.class, Integer.class |
| // ) |
| // )).withName("Add 2") |
| // .collect(); |
| // |
| // // Check the outcome. |
| // final List<Integer> expectedOutputValues = WayangArrays.asList(2, 3, 4, 5); |
| // Assert.assertEquals(WayangCollections.asSet(expectedOutputValues), WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testWordCount() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // |
| // final List<String> inputValues = Arrays.asList("Big data is big.", "Is data big data?"); |
| // |
| // // Build and execute a Wayang plan. |
| // final Collection<Tuple2<String, Integer>> outputValues = new JavaPlanBuilder(wayangContext) |
| // .loadCollection(inputValues).withName("Load input values") |
| // .flatMap(line -> Arrays.asList(line.split("\\s+"))).withName("Split words") |
| // .map(token -> token.replaceAll("\\W+", "").toLowerCase()).withName("To lower case") |
| // .map(word -> new Tuple2<>(word, 1)).withName("Attach counter") |
| // .reduceByKey(Tuple2::getField0, (t1, t2) -> new Tuple2<>(t1.field0, t1.field1 + t2.field1)).withName("Sum counters") |
| // .collect(); |
| // |
| // // Check the outcome. |
| // final Set<Tuple2<String, Integer>> expectedOutputValues = WayangCollections.asSet( |
| // new Tuple2<>("big", 3), |
| // new Tuple2<>("is", 2), |
| // new Tuple2<>("data", 3) |
| // ); |
| // Assert.assertEquals(WayangCollections.asSet(expectedOutputValues), WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testWordCountOnSparkAndJava() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()).with(Spark.basicPlugin()); |
| // |
| // final List<String> inputValues = Arrays.asList("Big data is big.", "Is data big data?"); |
| // |
| // // Build and execute a Wayang plan. |
| // final Collection<Tuple2<String, Integer>> outputValues = new JavaPlanBuilder(wayangContext) |
| // .loadCollection(inputValues).withName("Load input values") |
| // .flatMap(line -> Arrays.asList(line.split("\\s+"))).withName("Split words") |
| // .map(token -> token.replaceAll("\\W+", "").toLowerCase()).withName("To lower case") |
| // .map(word -> new Tuple2<>(word, 1)).withName("Attach counter") |
| // .reduceByKey(Tuple2::getField0, (t1, t2) -> new Tuple2<>(t1.field0, t1.field1 + t2.field1)).withName("Sum counters") |
| // .collect(); |
| // |
| // // Check the outcome. |
| // final Set<Tuple2<String, Integer>> expectedOutputValues = WayangCollections.asSet( |
| // new Tuple2<>("big", 3), |
| // new Tuple2<>("is", 2), |
| // new Tuple2<>("data", 3) |
| // ); |
| // Assert.assertEquals(WayangCollections.asSet(expectedOutputValues), WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testSample() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()).with(Spark.basicPlugin()); |
| // |
| // // Create some input values. |
| // final List<Integer> inputValues = WayangArrays.asList(WayangArrays.range(100)); |
| // |
| // // Build and execute a Wayang plan. |
| // final Collection<Integer> outputValues = new JavaPlanBuilder(wayangContext) |
| // .loadCollection(inputValues).withName("Load input values") |
| // .sample(10).withName("Sample") |
| // .collect(); |
| // |
| // // Check the outcome. |
| // Assert.assertEquals(10, outputValues.size()); |
| // Assert.assertEquals(10, WayangCollections.asSet(outputValues).size()); |
| // |
| // } |
| // |
| // @Test |
| // public void testDoWhile() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // |
| // // Generate test data. |
| // final List<Integer> inputValues = WayangArrays.asList(1, 2); |
| // |
| // // Build and execute a word count WayangPlan. |
| // |
| // final Collection<Integer> outputValues = new JavaPlanBuilder(wayangContext) |
| // .loadCollection(inputValues).withName("Load input values") |
| // .doWhile( |
| // values -> values.stream().mapToInt(i -> i).sum() > 100, |
| // start -> { |
| // final GlobalReduceDataQuantaBuilder<Integer> sum = |
| // start.reduce((a, b) -> a + b).withName("sum"); |
| // return new Tuple<>( |
| // start.union(sum).withName("Old+new"), |
| // sum |
| // ); |
| // } |
| // ).withConditionClass(Integer.class).withName("While <= 100") |
| // .collect(); |
| // |
| // Set<Integer> expectedValues = WayangCollections.asSet(1, 2, 3, 6, 12, 24, 48, 96, 192); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // private static class AddOffset implements FunctionDescriptor.ExtendedSerializableFunction<Integer, Integer> { |
| // |
| // private final String broadcastName; |
| // |
| // private int offset; |
| // |
| // public AddOffset(String broadcastName) { |
| // this.broadcastName = broadcastName; |
| // } |
| // |
| // @Override |
| // public void open(ExecutionContext ctx) { |
| // this.offset = WayangCollections.getSingle(ctx.<Integer>getBroadcast(this.broadcastName)); |
| // } |
| // |
| // @Override |
| // public Integer apply(Integer input) { |
| // return input + this.offset; |
| // } |
| // } |
| // |
| // @Test |
| // public void testRepeat() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // |
| // // Generate test data. |
| // final List<Integer> inputValues = WayangArrays.asList(1, 2); |
| // |
| // // Build and execute a word count WayangPlan. |
| // |
| // final Collection<Integer> outputValues = new JavaPlanBuilder(wayangContext) |
| // .loadCollection(inputValues).withName("Load input values") |
| // .repeat(3, start -> start |
| // .reduce((a, b) -> a * b).withName("Multiply") |
| // .flatMap(v -> Arrays.asList(v, v + 1)).withName("Duplicate").withOutputClass(Integer.class) |
| // ).withName("Repeat 3x") |
| // .collect(); |
| // |
| // Set<Integer> expectedValues = WayangCollections.asSet(42, 43); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // private static class SelectWords implements PredicateDescriptor.ExtendedSerializablePredicate<String> { |
| // |
| // private final String broadcastName; |
| // |
| // private Collection<Character> selectors; |
| // |
| // public SelectWords(String broadcastName) { |
| // this.broadcastName = broadcastName; |
| // } |
| // |
| // @Override |
| // public void open(ExecutionContext ctx) { |
| // this.selectors = ctx.getBroadcast(this.broadcastName); |
| // } |
| // |
| // @Override |
| // public boolean test(String word) { |
| // return this.selectors.stream().anyMatch(c -> word.indexOf(c) >= 0); |
| // } |
| // } |
| // |
| // @Test |
| // public void testBroadcast() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<String> inputValues = Arrays.asList("Hello", "World", "Hi", "Mars"); |
| // final List<Character> selectors = Arrays.asList('o', 'l'); |
| // |
| // // Execute the job. |
| // final DataQuantaBuilder<?, Character> selectorsDataSet = builder.loadCollection(selectors).withName("Load selectors"); |
| // final Collection<String> outputValues = builder |
| // .loadCollection(inputValues).withName("Load input values") |
| // .filter(new SelectWords("selectors")).withName("Filter words") |
| // .withBroadcast(selectorsDataSet, "selectors") |
| // .collect(); |
| // |
| // // Verify the outcome. |
| // Set<String> expectedValues = WayangCollections.asSet("Hello", "World"); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testGroupBy() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Integer> inputValues = Arrays.asList(1, 2, 3, 4, 5, 7, 8, 9, 10); |
| // |
| // // Execute the job. |
| // final Collection<Double> outputValues = builder |
| // .loadCollection(inputValues).withName("Load input values") |
| // .groupByKey(i -> i % 2).withName("group odd and even") |
| // .map(group -> { |
| // List<Integer> sortedGroup = StreamSupport.stream(group.spliterator(), false) |
| // .sorted() |
| // .collect(Collectors.toList()); |
| // int sizeDivTwo = sortedGroup.size() / 2; |
| // return sortedGroup.size() % 2 == 0 ? |
| // (sortedGroup.get(sizeDivTwo - 1) + sortedGroup.get(sizeDivTwo)) / 2d : |
| // (double) sortedGroup.get(sizeDivTwo); |
| // }) |
| // .collect(); |
| // |
| // // Verify the outcome. |
| // Set<Double> expectedValues = WayangCollections.asSet(5d, 6d); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testJoin() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Tuple2<String, Integer>> inputValues1 = Arrays.asList( |
| // new Tuple2<>("Water", 0), |
| // new Tuple2<>("Tonic", 5), |
| // new Tuple2<>("Juice", 10) |
| // ); |
| // final List<Tuple2<String, String>> inputValues2 = Arrays.asList( |
| // new Tuple2<>("Apple juice", "Juice"), |
| // new Tuple2<>("Tap water", "Water"), |
| // new Tuple2<>("Orange juice", "Juice") |
| // ); |
| // |
| // // Execute the job. |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, Integer>> dataQuanta1 = builder.loadCollection(inputValues1); |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, String>> dataQuanta2 = builder.loadCollection(inputValues2); |
| // final Collection<Tuple2<String, Integer>> outputValues = dataQuanta1 |
| // .join(Tuple2::getField0, dataQuanta2, Tuple2::getField1) |
| // .map(joinTuple -> new Tuple2<>(joinTuple.getField1().getField0(), joinTuple.getField0().getField1())) |
| // .collect(); |
| // |
| // // Verify the outcome. |
| // Set<Tuple2<String, Integer>> expectedValues = WayangCollections.asSet( |
| // new Tuple2<>("Apple juice", 10), |
| // new Tuple2<>("Orange juice", 10), |
| // new Tuple2<>("Tap water", 0) |
| // ); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testJoinAndAssemble() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Tuple2<String, Integer>> inputValues1 = Arrays.asList( |
| // new Tuple2<>("Water", 0), |
| // new Tuple2<>("Tonic", 5), |
| // new Tuple2<>("Juice", 10) |
| // ); |
| // final List<Tuple2<String, String>> inputValues2 = Arrays.asList( |
| // new Tuple2<>("Apple juice", "Juice"), |
| // new Tuple2<>("Tap water", "Water"), |
| // new Tuple2<>("Orange juice", "Juice") |
| // ); |
| // |
| // // Execute the job. |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, Integer>> dataQuanta1 = builder.loadCollection(inputValues1); |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, String>> dataQuanta2 = builder.loadCollection(inputValues2); |
| // final Collection<Tuple2<String, Integer>> outputValues = dataQuanta1.keyBy(Tuple2::getField0) |
| // .join(dataQuanta2.keyBy(Tuple2::getField1)) |
| // .assemble((val1, val2) -> new Tuple2<>(val2.getField0(), val1.getField1())) |
| // .collect(); |
| // |
| // // Verify the outcome. |
| // Set<Tuple2<String, Integer>> expectedValues = WayangCollections.asSet( |
| // new Tuple2<>("Apple juice", 10), |
| // new Tuple2<>("Orange juice", 10), |
| // new Tuple2<>("Tap water", 0) |
| // ); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testCoGroup() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()).with(Spark.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Tuple2<String, Integer>> inputValues1 = Arrays.asList( |
| // new Tuple2<>("Water", 0), |
| // new Tuple2<>("Cola", 5), |
| // new Tuple2<>("Juice", 10) |
| // ); |
| // final List<Tuple2<String, String>> inputValues2 = Arrays.asList( |
| // new Tuple2<>("Apple juice", "Juice"), |
| // new Tuple2<>("Tap water", "Water"), |
| // new Tuple2<>("Orange juice", "Juice") |
| // ); |
| // |
| // // Execute the job. |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, Integer>> dataQuanta1 = builder.loadCollection(inputValues1); |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, String>> dataQuanta2 = builder.loadCollection(inputValues2); |
| // final Collection<Tuple2<Set<Tuple2<String, Integer>>, Set<Tuple2<String, String>>>> outputValues = dataQuanta1 |
| // .coGroup(Tuple2::getField0, dataQuanta2, Tuple2::getField1) |
| // .map(joinTuple -> new Tuple2<>( |
| // WayangCollections.asSet(joinTuple.getField0()), |
| // WayangCollections.asSet(joinTuple.getField1()) |
| // )) |
| // .collect(); |
| // |
| // // Verify the outcome. |
| // Set<Tuple2<Set<Tuple2<String, Integer>>, Set<Tuple2<String, String>>>> expectedValues = WayangCollections.asSet( |
| // new Tuple2<>( |
| // WayangCollections.asSet(new Tuple2<>("Water", 0)), |
| // WayangCollections.asSet(new Tuple2<>("Tap water", "Water")) |
| // ), |
| // new Tuple2<>( |
| // WayangCollections.asSet(new Tuple2<>("Cola", 5)), |
| // WayangCollections.asSet() |
| // ), new Tuple2<>( |
| // WayangCollections.asSet(new Tuple2<>("Juice", 10)), |
| // WayangCollections.asSet(new Tuple2<>("Apple juice", "Juice"), new Tuple2<>("Orange juice", "Juice")) |
| // ) |
| // ); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testCoGroupViaKeyBy() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()).with(Spark.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Tuple2<String, Integer>> inputValues1 = Arrays.asList( |
| // new Tuple2<>("Water", 0), |
| // new Tuple2<>("Cola", 5), |
| // new Tuple2<>("Juice", 10) |
| // ); |
| // final List<Tuple2<String, String>> inputValues2 = Arrays.asList( |
| // new Tuple2<>("Apple juice", "Juice"), |
| // new Tuple2<>("Tap water", "Water"), |
| // new Tuple2<>("Orange juice", "Juice") |
| // ); |
| // |
| // // Execute the job. |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, Integer>> dataQuanta1 = builder.loadCollection(inputValues1); |
| // final LoadCollectionDataQuantaBuilder<Tuple2<String, String>> dataQuanta2 = builder.loadCollection(inputValues2); |
| // final Collection<Tuple2<Set<Tuple2<String, Integer>>, Set<Tuple2<String, String>>>> outputValues = |
| // dataQuanta1.keyBy(Tuple2::getField0) |
| // .coGroup(dataQuanta2.keyBy(Tuple2::getField1)) |
| // .map(joinTuple -> new Tuple2<>( |
| // WayangCollections.asSet(joinTuple.getField0()), |
| // WayangCollections.asSet(joinTuple.getField1()) |
| // )) |
| // .collect(); |
| // |
| // // Verify the outcome. |
| // Set<Tuple2<Set<Tuple2<String, Integer>>, Set<Tuple2<String, String>>>> expectedValues = WayangCollections.asSet( |
| // new Tuple2<>( |
| // WayangCollections.asSet(new Tuple2<>("Water", 0)), |
| // WayangCollections.asSet(new Tuple2<>("Tap water", "Water")) |
| // ), |
| // new Tuple2<>( |
| // WayangCollections.asSet(new Tuple2<>("Cola", 5)), |
| // WayangCollections.asSet() |
| // ), new Tuple2<>( |
| // WayangCollections.asSet(new Tuple2<>("Juice", 10)), |
| // WayangCollections.asSet(new Tuple2<>("Apple juice", "Juice"), new Tuple2<>("Orange juice", "Juice")) |
| // ) |
| // ); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testIntersect() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Integer> inputValues1 = Arrays.asList(1, 2, 3, 4, 5, 7, 8, 9, 10); |
| // final List<Integer> inputValues2 = Arrays.asList(0, 2, 3, 3, 4, 5, 7, 8, 9, 11); |
| // |
| // // Execute the job. |
| // final LoadCollectionDataQuantaBuilder<Integer> dataQuanta1 = builder.loadCollection(inputValues1); |
| // final LoadCollectionDataQuantaBuilder<Integer> dataQuanta2 = builder.loadCollection(inputValues2); |
| // final Collection<Integer> outputValues = dataQuanta1.intersect(dataQuanta2).collect(); |
| // |
| // // Verify the outcome. |
| // Set<Integer> expectedValues = WayangCollections.asSet(2, 3, 4, 5, 7, 8, 9); |
| // Assert.assertEquals(expectedValues, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testSort() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // final List<Integer> inputValues1 = Arrays.asList(3, 4, 5, 2, 1); |
| // |
| // // Execute the job. |
| // final LoadCollectionDataQuantaBuilder<Integer> dataQuanta1 = builder.loadCollection(inputValues1); |
| // final Collection<Integer> outputValues = dataQuanta1.sort(r -> r).collect(); |
| // |
| // // Verify the outcome. |
| // List<Integer> expectedValues = Arrays.asList(1, 2, 3, 4, 5); |
| // Assert.assertEquals(expectedValues, WayangCollections.asList(outputValues)); |
| // } |
| // |
| // |
| // @Test |
| // public void testPageRank() { |
| // // Set up WayangContext. |
| // WayangContext wayangContext = new WayangContext() |
| // .with(Java.basicPlugin()) |
| // .with(Java.graphPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Create a test graph. |
| // Collection<Tuple2<Long, Long>> edges = Arrays.asList( |
| // new Tuple2<>(0L, 1L), |
| // new Tuple2<>(0L, 2L), |
| // new Tuple2<>(0L, 3L), |
| // new Tuple2<>(1L, 0L), |
| // new Tuple2<>(2L, 1L), |
| // new Tuple2<>(3L, 2L), |
| // new Tuple2<>(3L, 1L) |
| // ); |
| // |
| // // Execute the job. |
| // Collection<Tuple2<Long, Float>> pageRanks = builder.loadCollection(edges).asEdges() |
| // .pageRank(20) |
| // .collect(); |
| // List<Tuple2<Long, Float>> sortedPageRanks = new ArrayList<>(pageRanks); |
| // sortedPageRanks.sort((pr1, pr2) -> Float.compare(pr2.field1, pr1.field1)); |
| // |
| // System.out.println(sortedPageRanks); |
| // Assert.assertEquals(1L, sortedPageRanks.get(0).field0.longValue()); |
| // Assert.assertEquals(0L, sortedPageRanks.get(1).field0.longValue()); |
| // Assert.assertEquals(2L, sortedPageRanks.get(2).field0.longValue()); |
| // Assert.assertEquals(3L, sortedPageRanks.get(3).field0.longValue()); |
| // } |
| // |
| // @Test |
| // public void testMapPartitions() { |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // List<Integer> inputValues = WayangArrays.asList(0, 1, 2, 3, 4, 6, 8); |
| // |
| // // Execute the job. |
| // Collection<Tuple2<String, Integer>> outputValues = builder.loadCollection(inputValues) |
| // .mapPartitions(partition -> { |
| // int numEvens = 0, numOdds = 0; |
| // for (Integer value : partition) { |
| // if ((value & 1) == 0) numEvens++; |
| // else numOdds++; |
| // } |
| // return Arrays.asList( |
| // new Tuple2<>("odd", numOdds), |
| // new Tuple2<>("even", numEvens) |
| // ); |
| // }) |
| // .reduceByKey(Tuple2::getField0, (t1, t2) -> new Tuple2<>(t1.getField0(), t1.getField1() + t2.getField1())) |
| // .collect(); |
| // |
| // // Check the output. |
| // Set<Tuple2<String, Integer>> expectedOutput = WayangCollections.asSet( |
| // new Tuple2<>("even", 5), new Tuple2<>("odd", 2) |
| // ); |
| // Assert.assertEquals(expectedOutput, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testZipWithId() { |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // List<Integer> inputValues = new ArrayList<>(42 * 100); |
| // for (int i = 0; i < 100; i++) { |
| // for (int j = 0; j < 42; j++) { |
| // inputValues.add(i); |
| // } |
| // } |
| // |
| // // Execute the job. |
| // Collection<Tuple2<Integer, Integer>> outputValues = builder.loadCollection(inputValues) |
| // .zipWithId() |
| // .groupByKey(Tuple2::getField1) |
| // .map(group -> { |
| // int distinctIds = (int) StreamSupport.stream(group.spliterator(), false) |
| // .map(Tuple2::getField0) |
| // .distinct() |
| // .count(); |
| // return new Tuple2<>(distinctIds, 1); |
| // }) |
| // .reduceByKey(Tuple2::getField0, (t1, t2) -> new Tuple2<>(t1.getField0(), t1.getField1() + t2.getField1())) |
| // .collect(); |
| // |
| // // Check the output. |
| // Set<Tuple2<Integer, Integer>> expectedOutput = Collections.singleton(new Tuple2<>(42, 100)); |
| // Assert.assertEquals(expectedOutput, WayangCollections.asSet(outputValues)); |
| // } |
| // |
| // @Test |
| // public void testWriteTextFile() throws IOException, URISyntaxException { |
| // WayangContext wayangContext = new WayangContext().with(Java.basicPlugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangContext); |
| // |
| // // Generate test data. |
| // List<Double> inputValues = Arrays.asList(0d, 1 / 3d, 2 / 3d, 1d, 4 / 3d, 5 / 3d); |
| // |
| // // Execute the job. |
| // File tempDir = LocalFileSystem.findTempDir(); |
| // String targetUrl = LocalFileSystem.toURL(new File(tempDir, "testWriteTextFile.txt")); |
| // |
| // builder |
| // .loadCollection(inputValues) |
| // .writeTextFile(targetUrl, d -> String.format("%.2f", d), "testWriteTextFile()"); |
| // |
| // // Check the output. |
| // Set<String> actualLines = Files.lines(Paths.get(new URI(targetUrl))).collect(Collectors.toSet()); |
| // Set<String> expectedLines = inputValues.stream().map(d -> String.format("%.2f", d)).collect(Collectors.toSet()); |
| // Assert.assertEquals(expectedLines, actualLines); |
| // } |
| // |
| // @Test |
| // public void testSqlOnJava() throws IOException, SQLException { |
| // // Execute job. |
| // final WayangContext wayangCtx = new WayangContext(this.sqlite3Configuration) |
| // .with(Java.basicPlugin()) |
| // .with(Sqlite3.plugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangCtx, "testSqlOnJava()"); |
| // final Collection<String> outputValues = builder |
| // .readTable(new Sqlite3TableSource("customer", "name", "age")) |
| // .filter(r -> (Integer) r.getField(1) >= 18).withSqlUdf("age >= 18").withTargetPlatform(Java.platform()) |
| // .asRecords().projectRecords(new String[]{"name"}) |
| // .map(record -> (String) record.getField(0)) |
| // .collect(); |
| // |
| // // Test the outcome. |
| // Assert.assertEquals( |
| // WayangCollections.asSet("John", "Evelyn"), |
| // WayangCollections.asSet(outputValues) |
| // ); |
| // } |
| // |
| // @Test |
| // public void testSqlOnSqlite3() throws IOException, SQLException { |
| // // Execute job. |
| // final WayangContext wayangCtx = new WayangContext(this.sqlite3Configuration) |
| // .with(Java.basicPlugin()) |
| // .with(Sqlite3.plugin()); |
| // JavaPlanBuilder builder = new JavaPlanBuilder(wayangCtx, "testSqlOnSqlite3()"); |
| // final Collection<String> outputValues = builder |
| // .readTable(new Sqlite3TableSource("customer", "name", "age")) |
| // .filter(r -> (Integer) r.getField(1) >= 18).withSqlUdf("age >= 18") |
| // .asRecords().projectRecords(new String[]{"name"}).withTargetPlatform(Sqlite3.platform()) |
| // .map(record -> (String) record.getField(0)) |
| // .collect(); |
| // |
| // // Test the outcome. |
| // Assert.assertEquals( |
| // WayangCollections.asSet("John", "Evelyn"), |
| // WayangCollections.asSet(outputValues) |
| // ); |
| // } |
| |
| } |
