| /* |
| * 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.flink.test.operators; |
| |
| import org.apache.flink.api.common.functions.GroupReduceFunction; |
| import org.apache.flink.api.common.functions.ReduceFunction; |
| import org.apache.flink.api.java.DataSet; |
| import org.apache.flink.api.java.ExecutionEnvironment; |
| import org.apache.flink.api.java.tuple.Tuple2; |
| import org.apache.flink.test.util.MultipleProgramsTestBase; |
| import org.apache.flink.util.Collector; |
| |
| import org.junit.Test; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.Parameterized; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.List; |
| |
| import static org.junit.Assert.assertEquals; |
| |
| /** These check whether the object-reuse execution mode does really reuse objects. */ |
| @SuppressWarnings("serial") |
| @RunWith(Parameterized.class) |
| public class ObjectReuseITCase extends MultipleProgramsTestBase { |
| |
| private static final List<Tuple2<String, Integer>> REDUCE_DATA = |
| Arrays.asList( |
| new Tuple2<>("a", 1), |
| new Tuple2<>("a", 2), |
| new Tuple2<>("a", 3), |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 50)); |
| |
| private static final List<Tuple2<String, Integer>> GROUP_REDUCE_DATA = |
| Arrays.asList( |
| new Tuple2<>("a", 1), |
| new Tuple2<>("a", 2), |
| new Tuple2<>("a", 3), |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 5)); |
| |
| private final boolean objectReuse; |
| |
| public ObjectReuseITCase(boolean objectReuse) { |
| super(TestExecutionMode.CLUSTER); |
| this.objectReuse = objectReuse; |
| } |
| |
| @Test |
| public void testKeyedReduce() throws Exception { |
| |
| final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); |
| if (objectReuse) { |
| env.getConfig().enableObjectReuse(); |
| } else { |
| env.getConfig().disableObjectReuse(); |
| } |
| |
| DataSet<Tuple2<String, Integer>> input = env.fromCollection(REDUCE_DATA); |
| |
| DataSet<Tuple2<String, Integer>> result = |
| input.groupBy(0) |
| .reduce( |
| new ReduceFunction<Tuple2<String, Integer>>() { |
| |
| @Override |
| public Tuple2<String, Integer> reduce( |
| Tuple2<String, Integer> value1, |
| Tuple2<String, Integer> value2) { |
| value2.f1 += value1.f1; |
| return value2; |
| } |
| }); |
| |
| Tuple2<String, Integer> res = result.collect().get(0); |
| assertEquals(new Tuple2<>("a", 60), res); |
| } |
| |
| @Test |
| public void testGlobalReduce() throws Exception { |
| |
| final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); |
| if (objectReuse) { |
| env.getConfig().enableObjectReuse(); |
| } else { |
| env.getConfig().disableObjectReuse(); |
| } |
| |
| DataSet<Tuple2<String, Integer>> input = env.fromCollection(REDUCE_DATA); |
| |
| DataSet<Tuple2<String, Integer>> result = |
| input.reduce( |
| new ReduceFunction<Tuple2<String, Integer>>() { |
| |
| @Override |
| public Tuple2<String, Integer> reduce( |
| Tuple2<String, Integer> value1, |
| Tuple2<String, Integer> value2) { |
| |
| if (value1.f1 % 3 == 0) { |
| value1.f1 += value2.f1; |
| return value1; |
| } else { |
| value2.f1 += value1.f1; |
| return value2; |
| } |
| } |
| }); |
| |
| Tuple2<String, Integer> res = result.collect().get(0); |
| assertEquals(new Tuple2<>("a", 60), res); |
| } |
| |
| @Test |
| public void testKeyedGroupReduce() throws Exception { |
| |
| final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); |
| if (objectReuse) { |
| env.getConfig().enableObjectReuse(); |
| } else { |
| env.getConfig().disableObjectReuse(); |
| } |
| |
| DataSet<Tuple2<String, Integer>> input = env.fromCollection(GROUP_REDUCE_DATA); |
| |
| DataSet<Tuple2<String, Integer>> result = |
| input.groupBy(0) |
| .reduceGroup( |
| new GroupReduceFunction< |
| Tuple2<String, Integer>, Tuple2<String, Integer>>() { |
| |
| @Override |
| public void reduce( |
| Iterable<Tuple2<String, Integer>> values, |
| Collector<Tuple2<String, Integer>> out) { |
| List<Tuple2<String, Integer>> list = new ArrayList<>(); |
| for (Tuple2<String, Integer> val : values) { |
| list.add(val); |
| } |
| |
| for (Tuple2<String, Integer> val : list) { |
| out.collect(val); |
| } |
| } |
| }); |
| |
| List<Tuple2<String, Integer>> is = result.collect(); |
| Collections.sort(is, new TupleComparator<Tuple2<String, Integer>>()); |
| |
| List<Tuple2<String, Integer>> expected = |
| env.getConfig().isObjectReuseEnabled() |
| ? Arrays.asList( |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 5), |
| new Tuple2<>("a", 5), |
| new Tuple2<>("a", 5)) |
| : Arrays.asList( |
| new Tuple2<>("a", 1), |
| new Tuple2<>("a", 2), |
| new Tuple2<>("a", 3), |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 5)); |
| |
| assertEquals(expected, is); |
| } |
| |
| @Test |
| public void testGlobalGroupReduce() throws Exception { |
| |
| final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); |
| if (objectReuse) { |
| env.getConfig().enableObjectReuse(); |
| } else { |
| env.getConfig().disableObjectReuse(); |
| } |
| |
| DataSet<Tuple2<String, Integer>> input = env.fromCollection(GROUP_REDUCE_DATA); |
| |
| DataSet<Tuple2<String, Integer>> result = |
| input.reduceGroup( |
| new GroupReduceFunction< |
| Tuple2<String, Integer>, Tuple2<String, Integer>>() { |
| |
| @Override |
| public void reduce( |
| Iterable<Tuple2<String, Integer>> values, |
| Collector<Tuple2<String, Integer>> out) { |
| List<Tuple2<String, Integer>> list = new ArrayList<>(); |
| for (Tuple2<String, Integer> val : values) { |
| list.add(val); |
| } |
| |
| for (Tuple2<String, Integer> val : list) { |
| out.collect(val); |
| } |
| } |
| }); |
| |
| List<Tuple2<String, Integer>> is = result.collect(); |
| Collections.sort(is, new TupleComparator<Tuple2<String, Integer>>()); |
| |
| List<Tuple2<String, Integer>> expected = |
| env.getConfig().isObjectReuseEnabled() |
| ? Arrays.asList( |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 5), |
| new Tuple2<>("a", 5), |
| new Tuple2<>("a", 5)) |
| : Arrays.asList( |
| new Tuple2<>("a", 1), |
| new Tuple2<>("a", 2), |
| new Tuple2<>("a", 3), |
| new Tuple2<>("a", 4), |
| new Tuple2<>("a", 5)); |
| |
| assertEquals(expected, is); |
| } |
| |
| @Parameterized.Parameters(name = "Execution mode = CLUSTER, Reuse = {0}") |
| public static Collection<Object[]> executionModes() { |
| return Arrays.asList( |
| new Object[] { |
| false, |
| }, |
| new Object[] {true}); |
| } |
| } |