map and reduce operators

flink-clients/pom.xml

@@ -154,6 +154,17 @@ under the License.
  </execution>
  </executions>
  </plugin>
+ <plugin>
+ <!-- just define the Java version to be used for compiling and plugins -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version><!--$NO-MVN-MAN-VER$-->
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ <!-- High optimization, no debugging <compilerArgument>-g:none -O</compilerArgument> -->
+ </configuration>
+ </plugin>
  </plugins>
 
  <pluginManagement>

flink-compiler/src/test/java/org/apache/flink/compiler/FeedbackPropertiesMatchTest.java

@@ -23,7 +23,7 @@
 import static org.junit.Assert.*;
 
 import org.apache.flink.api.common.functions.GenericJoiner;
-import org.apache.flink.api.common.functions.GenericMap;
+import org.apache.flink.api.common.functions.MapFunctional;
 import org.apache.flink.api.common.operators.BinaryOperatorInformation;
 import org.apache.flink.api.common.operators.OperatorInformation;
 import org.apache.flink.api.common.operators.Order;
@@ -1426,7 +1426,7 @@ private static final DataSourceNode getSourceNode() {
  }
 
  private static final MapNode getMapNode() {
- return new MapNode(new MapOperatorBase<String, String, GenericMap<String,String>>(new IdentityMapper<String>(), new UnaryOperatorInformation<String, String>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO), "map op"));
+ return new MapNode(new MapOperatorBase<String, String, MapFunctional<String,String>>(new IdentityMapper<String>(), new UnaryOperatorInformation<String, String>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO), "map op"));
  }
 
  private static final MatchNode getJoinNode() {

flink-core/src/main/java/org/apache/flink/api/common/functions/GenericMap.java → flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunctional.java

@@ -20,7 +20,9 @@
 package org.apache.flink.api.common.functions;
 
 
-public interface GenericMap<T, O> extends RichFunction {
+import java.io.Serializable;
+
+public interface MapFunctional<T, O> extends Function, Serializable {
 
  /**
  * A user-implemented function that modifies or transforms an incoming object and

flink-core/src/main/java/org/apache/flink/api/common/functions/GenericReduce.java → flink-core/src/main/java/org/apache/flink/api/common/functions/ReduceFunctional.java

@@ -20,7 +20,9 @@
 package org.apache.flink.api.common.functions;
 
 
-public interface GenericReduce<T> extends RichFunction {
+import java.io.Serializable;
+
+public interface ReduceFunctional<T> extends Function, Serializable {
 
  T reduce(T value1, T value2) throws Exception;
 }

flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java

@@ -22,15 +22,15 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.api.common.functions.Function;
 import org.apache.flink.api.common.operators.util.UserCodeWrapper;
 
 /**
  * Abstract superclass for all contracts that represent actual operators.
  *
  * @param <FT> Type of the user function
  */
-public abstract class AbstractUdfOperator<OUT, FT extends RichFunction> extends Operator<OUT> {
+public abstract class AbstractUdfOperator<OUT, FT extends Function> extends Operator<OUT> {
 
  /**
  * The object or class containing the user function.

flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputOperator.java

@@ -21,7 +21,7 @@
 
 import java.util.List;
 
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.api.common.functions.Function;
 import org.apache.flink.api.common.operators.util.UserCodeWrapper;
 import org.apache.flink.util.Visitor;
 
@@ -32,7 +32,7 @@
  * @param <OUT> Output type of the user function
  * @param <FT> Type of the user function
  */
-public abstract class SingleInputOperator<IN, OUT, FT extends RichFunction> extends AbstractUdfOperator<OUT, FT> {
+public abstract class SingleInputOperator<IN, OUT, FT extends Function> extends AbstractUdfOperator<OUT, FT> {
 
  /**
  * The input which produces the data consumed by this operator.

flink-core/src/main/java/org/apache/flink/api/common/operators/base/MapOperatorBase.java

@@ -19,7 +19,7 @@
 
 package org.apache.flink.api.common.operators.base;
 
-import org.apache.flink.api.common.functions.GenericMap;
+import org.apache.flink.api.common.functions.MapFunctional;
 import org.apache.flink.api.common.operators.SingleInputOperator;
 import org.apache.flink.api.common.operators.UnaryOperatorInformation;
 import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
@@ -33,7 +33,7 @@
  * @param <OUT> The result type.
  * @param <FT> The type of the user-defined function.
  */
-public class MapOperatorBase<IN, OUT, FT extends GenericMap<IN, OUT>> extends SingleInputOperator<IN, OUT, FT> {
+public class MapOperatorBase<IN, OUT, FT extends MapFunctional<IN, OUT>> extends SingleInputOperator<IN, OUT, FT> {
 
  public MapOperatorBase(UserCodeWrapper<FT> udf, UnaryOperatorInformation<IN, OUT> operatorInfo, String name) {
  super(udf, operatorInfo, name);

flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java

@@ -19,7 +19,7 @@
 
 package org.apache.flink.api.common.operators.base;
 
-import org.apache.flink.api.common.functions.GenericReduce;
+import org.apache.flink.api.common.functions.ReduceFunctional;
 import org.apache.flink.api.common.operators.SingleInputOperator;
 import org.apache.flink.api.common.operators.UnaryOperatorInformation;
 import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
@@ -31,12 +31,12 @@
  * Base data flow operator for Reduce user-defined functions. Accepts reduce functions
  * and key positions. The key positions are expected in the flattened common data model.
  * 
- * @see GenericReduce
+ * @see org.apache.flink.api.common.functions.ReduceFunctional
  *
  * @param <T> The type (parameters and return type) of the reduce function.
  * @param <FT> The type of the reduce function.
  */
-public class ReduceOperatorBase<T, FT extends GenericReduce<T>> extends SingleInputOperator<T, T, FT> {
+public class ReduceOperatorBase<T, FT extends ReduceFunctional<T>> extends SingleInputOperator<T, T, FT> {
 
  /**
  * Creates a grouped reduce data flow operator.

flink-examples/flink-java-examples/pom.xml

@@ -318,6 +318,18 @@ under the License.
  </executions>
  </plugin>
 
+ <plugin>
+ <!-- just define the Java version to be used for compiling and plugins -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version><!--$NO-MVN-MAN-VER$-->
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ <!-- High optimization, no debugging <compilerArgument>-g:none -O</compilerArgument> -->
+ </configuration>
+ </plugin>
+
  </plugins>
  </build>
 

flink-examples/pom.xml

@@ -51,4 +51,20 @@ under the License.
  <module>flink-scala-examples</module>
  </modules>
 
+ <build>
+ <plugins>
+ <plugin>
+ <!-- just define the Java version to be used for compiling and plugins -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version><!--$NO-MVN-MAN-VER$-->
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ <!-- High optimization, no debugging <compilerArgument>-g:none -O</compilerArgument> -->
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
 </project>

flink-java/pom.xml

@@ -34,6 +34,22 @@ under the License.
 
  <packaging>jar</packaging>
 
+ <build>
+ <plugins>
+ <plugin>
+ <!-- just define the Java version to be used for compiling and plugins -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version><!--$NO-MVN-MAN-VER$-->
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ <!-- High optimization, no debugging <compilerArgument>-g:none -O</compilerArgument> -->
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
  <dependencies>
  <dependency>
  <groupId>org.apache.flink</groupId>

flink-java/src/main/java/org/apache/flink/api/java/DataSet.java

@@ -20,6 +20,8 @@
 
 import org.apache.commons.lang3.Validate;
 import org.apache.flink.api.common.functions.FilterFunctional;
+import org.apache.flink.api.common.functions.MapFunctional;
+import org.apache.flink.api.common.functions.ReduceFunctional;
 import org.apache.flink.api.common.io.FileOutputFormat;
 import org.apache.flink.api.common.io.OutputFormat;
 import org.apache.flink.api.java.aggregation.Aggregations;
@@ -135,7 +137,7 @@ public TypeInformation<T> getType() {
  * @see MapOperator
  * @see DataSet
  */
- public <R> MapOperator<T, R> map(MapFunction<T, R> mapper) {
+ public <R> MapOperator<T, R> map(MapFunctional<T, R> mapper) {
  if (mapper == null) {
  throw new NullPointerException("Map function must not be null.");
  }
@@ -276,7 +278,7 @@ public AggregateOperator<T> min (int field) {
  * @see ReduceOperator
  * @see DataSet
  */
- public ReduceOperator<T> reduce(ReduceFunction<T> reducer) {
+ public ReduceOperator<T> reduce(ReduceFunctional<T> reducer) {
  if (reducer == null) {
  throw new NullPointerException("Reduce function must not be null.");
  }

flink-java/src/main/java/org/apache/flink/api/java/functions/MapFunction.java

@@ -19,7 +19,7 @@
 package org.apache.flink.api.java.functions;
 
 import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.GenericMap;
+import org.apache.flink.api.common.functions.MapFunctional;
 
 /**
  * The abstract base class for Map functions. Map functions take elements and transform them,
@@ -40,7 +40,7 @@
  * @param <IN> Type of the input elements.
  * @param <OUT> Type of the returned elements.
  */
-public abstract class MapFunction<IN, OUT> extends AbstractRichFunction implements GenericMap<IN, OUT> {
+public abstract class MapFunction<IN, OUT> extends AbstractRichFunction implements MapFunctional<IN, OUT> {
 
  private static final long serialVersionUID = 1L;
 

flink-java/src/main/java/org/apache/flink/api/java/functions/ReduceFunction.java

@@ -19,7 +19,7 @@
 package org.apache.flink.api.java.functions;
 
 import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.GenericReduce;
+import org.apache.flink.api.common.functions.ReduceFunctional;
 
 /**
  * The abstract base class for Reduce functions. Reduce functions combine groups of elements to
@@ -44,7 +44,7 @@
  * 
  * @param <T> Type of the elements that this function processes.
  */
-public abstract class ReduceFunction<T> extends AbstractRichFunction implements GenericReduce<T> {
+public abstract class ReduceFunction<T> extends AbstractRichFunction implements ReduceFunctional<T> {
 
  private static final long serialVersionUID = 1L;
 

flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java

@@ -22,7 +22,7 @@
 
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.functions.GenericCoGrouper;
-import org.apache.flink.api.common.functions.GenericMap;
+import org.apache.flink.api.common.functions.MapFunctional;
 import org.apache.flink.api.common.operators.BinaryOperatorInformation;
 import org.apache.flink.api.common.operators.Operator;
 import org.apache.flink.api.common.operators.UnaryOperatorInformation;
@@ -199,10 +199,10 @@ private static <I1, I2, K, OUT> PlanUnwrappingCoGroupOperator<I1, I2, OUT, K> tr
  final KeyExtractingMapper<I1, K> extractor1 = new KeyExtractingMapper<I1, K>(keys1.getKeyExtractor());
  final KeyExtractingMapper<I2, K> extractor2 = new KeyExtractingMapper<I2, K>(keys2.getKeyExtractor());
 
- final MapOperatorBase<I1, Tuple2<K, I1>, GenericMap<I1, Tuple2<K, I1>>> keyMapper1 =
- new MapOperatorBase<I1, Tuple2<K, I1>, GenericMap<I1, Tuple2<K, I1>>>(extractor1, new UnaryOperatorInformation<I1, Tuple2<K, I1>>(inputType1, typeInfoWithKey1), "Key Extractor 1");
- final MapOperatorBase<I2, Tuple2<K, I2>, GenericMap<I2, Tuple2<K, I2>>> keyMapper2 =
- new MapOperatorBase<I2, Tuple2<K, I2>, GenericMap<I2, Tuple2<K, I2>>>(extractor2, new UnaryOperatorInformation<I2, Tuple2<K, I2>>(inputType2, typeInfoWithKey2), "Key Extractor 2");
+ final MapOperatorBase<I1, Tuple2<K, I1>, MapFunctional<I1, Tuple2<K, I1>>> keyMapper1 =
+ new MapOperatorBase<I1, Tuple2<K, I1>, MapFunctional<I1, Tuple2<K, I1>>>(extractor1, new UnaryOperatorInformation<I1, Tuple2<K, I1>>(inputType1, typeInfoWithKey1), "Key Extractor 1");
+ final MapOperatorBase<I2, Tuple2<K, I2>, MapFunctional<I2, Tuple2<K, I2>>> keyMapper2 =
+ new MapOperatorBase<I2, Tuple2<K, I2>, MapFunctional<I2, Tuple2<K, I2>>>(extractor2, new UnaryOperatorInformation<I2, Tuple2<K, I2>>(inputType2, typeInfoWithKey2), "Key Extractor 2");
  final PlanUnwrappingCoGroupOperator<I1, I2, OUT, K> cogroup = new PlanUnwrappingCoGroupOperator<I1, I2, OUT, K>(function, keys1, keys2, name, outputType, typeInfoWithKey1, typeInfoWithKey2);
 
  cogroup.setFirstInput(keyMapper1);
@@ -236,10 +236,10 @@ private static <I1, I2, K, OUT> PlanUnwrappingCoGroupOperator<I1, I2, OUT, K> tr
  final TupleKeyExtractingMapper<I1, K> extractor1 = new TupleKeyExtractingMapper<I1, K>(logicalKeyPositions1[0]);
  final KeyExtractingMapper<I2, K> extractor2 = new KeyExtractingMapper<I2, K>(keys2.getKeyExtractor());
 
- final MapOperatorBase<I1, Tuple2<K, I1>, GenericMap<I1, Tuple2<K, I1>>> keyMapper1 =
- new MapOperatorBase<I1, Tuple2<K, I1>, GenericMap<I1, Tuple2<K, I1>>>(extractor1, new UnaryOperatorInformation<I1, Tuple2<K, I1>>(inputType1, typeInfoWithKey1), "Key Extractor 1");
- final MapOperatorBase<I2, Tuple2<K, I2>, GenericMap<I2, Tuple2<K, I2>>> keyMapper2 =
- new MapOperatorBase<I2, Tuple2<K, I2>, GenericMap<I2, Tuple2<K, I2>>>(extractor2, new UnaryOperatorInformation<I2, Tuple2<K, I2>>(inputType2, typeInfoWithKey2), "Key Extractor 2");
+ final MapOperatorBase<I1, Tuple2<K, I1>, MapFunctional<I1, Tuple2<K, I1>>> keyMapper1 =
+ new MapOperatorBase<I1, Tuple2<K, I1>, MapFunctional<I1, Tuple2<K, I1>>>(extractor1, new UnaryOperatorInformation<I1, Tuple2<K, I1>>(inputType1, typeInfoWithKey1), "Key Extractor 1");
+ final MapOperatorBase<I2, Tuple2<K, I2>, MapFunctional<I2, Tuple2<K, I2>>> keyMapper2 =
+ new MapOperatorBase<I2, Tuple2<K, I2>, MapFunctional<I2, Tuple2<K, I2>>>(extractor2, new UnaryOperatorInformation<I2, Tuple2<K, I2>>(inputType2, typeInfoWithKey2), "Key Extractor 2");
 
  final PlanUnwrappingCoGroupOperator<I1, I2, OUT, K> cogroup = new PlanUnwrappingCoGroupOperator<I1, I2, OUT, K>(function, logicalKeyPositions1, keys2, name, outputType, typeInfoWithKey1, typeInfoWithKey2);
 
@@ -274,10 +274,10 @@ private static <I1, I2, K, OUT> PlanUnwrappingCoGroupOperator<I1, I2, OUT, K> tr
  final KeyExtractingMapper<I1, K> extractor1 = new KeyExtractingMapper<I1, K>(keys1.getKeyExtractor());
  final TupleKeyExtractingMapper<I2, K> extractor2 = new TupleKeyExtractingMapper<I2, K>(logicalKeyPositions2[0]);
 
- final MapOperatorBase<I1, Tuple2<K, I1>, GenericMap<I1, Tuple2<K, I1>>> keyMapper1 =
- new MapOperatorBase<I1, Tuple2<K, I1>, GenericMap<I1, Tuple2<K, I1>>>(extractor1, new UnaryOperatorInformation<I1, Tuple2<K, I1>>(inputType1, typeInfoWithKey1), "Key Extractor 1");
- final MapOperatorBase<I2, Tuple2<K, I2>, GenericMap<I2, Tuple2<K, I2>>> keyMapper2 =
- new MapOperatorBase<I2, Tuple2<K, I2>, GenericMap<I2, Tuple2<K, I2>>>(extractor2, new UnaryOperatorInformation<I2, Tuple2<K, I2>>(inputType2, typeInfoWithKey2), "Key Extractor 2");
+ final MapOperatorBase<I1, Tuple2<K, I1>, MapFunctional<I1, Tuple2<K, I1>>> keyMapper1 =
+ new MapOperatorBase<I1, Tuple2<K, I1>, MapFunctional<I1, Tuple2<K, I1>>>(extractor1, new UnaryOperatorInformation<I1, Tuple2<K, I1>>(inputType1, typeInfoWithKey1), "Key Extractor 1");
+ final MapOperatorBase<I2, Tuple2<K, I2>, MapFunctional<I2, Tuple2<K, I2>>> keyMapper2 =
+ new MapOperatorBase<I2, Tuple2<K, I2>, MapFunctional<I2, Tuple2<K, I2>>>(extractor2, new UnaryOperatorInformation<I2, Tuple2<K, I2>>(inputType2, typeInfoWithKey2), "Key Extractor 2");
 
  final PlanUnwrappingCoGroupOperator<I1, I2, OUT, K> cogroup = new PlanUnwrappingCoGroupOperator<I1, I2, OUT, K>(function, keys1, logicalKeyPositions2, name, outputType, typeInfoWithKey1, typeInfoWithKey2);
 

flink-java/src/main/java/org/apache/flink/api/java/operators/DistinctOperator.java

@@ -22,7 +22,7 @@
 
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.functions.GenericGroupReduce;
-import org.apache.flink.api.common.functions.GenericMap;
+import org.apache.flink.api.common.functions.MapFunctional;
 import org.apache.flink.api.common.operators.Operator;
 import org.apache.flink.api.common.operators.UnaryOperatorInformation;
 import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase;
@@ -127,7 +127,7 @@ private static <IN, OUT, K> PlanUnwrappingReduceGroupOperator<IN, OUT, K> transl
 
  PlanUnwrappingReduceGroupOperator<IN, OUT, K> reducer = new PlanUnwrappingReduceGroupOperator<IN, OUT, K>(function, keys, name, outputType, typeInfoWithKey, combinable);
 
- MapOperatorBase<IN, Tuple2<K, IN>, GenericMap<IN, Tuple2<K, IN>>> mapper = new MapOperatorBase<IN, Tuple2<K, IN>, GenericMap<IN, Tuple2<K, IN>>>(extractor, new UnaryOperatorInformation<IN, Tuple2<K, IN>>(inputType, typeInfoWithKey), "Key Extractor");
+ MapOperatorBase<IN, Tuple2<K, IN>, MapFunctional<IN, Tuple2<K, IN>>> mapper = new MapOperatorBase<IN, Tuple2<K, IN>, MapFunctional<IN, Tuple2<K, IN>>>(extractor, new UnaryOperatorInformation<IN, Tuple2<K, IN>>(inputType, typeInfoWithKey), "Key Extractor");
 
  reducer.setInput(mapper);
  mapper.setInput(input);