From 52c2044f5652ab152619d94e483c85757ec2a33d Mon Sep 17 00:00:00 2001 From: smarthi Date: Sun, 27 Dec 2015 21:21:02 -0500 Subject: [PATCH] FLINK-3186: Deprecate DataSink.sortLocalOutput() methods --- docs/apis/programming_guide.md | 20 +- .../common/operators/AbstractUdfOperator.java | 6 +- .../org/apache/flink/api/java/DataSet.java | 96 +++--- .../java/org/apache/flink/api/java/Utils.java | 4 +- .../flink/api/java/operators/DataSink.java | 9 +- .../flink/api/java/operator/DataSinkTest.java | 101 +++--- .../AccumulatorIterativeITCase.java | 4 +- .../test/javaApiOperators/DataSinkITCase.java | 44 +-- .../javaApiOperators/GroupCombineITCase.java | 60 ++-- .../javaApiOperators/SortPartitionITCase.java | 20 +- .../util/CollectionDataSets.java | 291 +++++++++--------- 11 files changed, 323 insertions(+), 332 deletions(-) diff --git a/docs/apis/programming_guide.md b/docs/apis/programming_guide.md index fb446f1bf8e623..002e8bc0ca4e8d 100644 --- a/docs/apis/programming_guide.md +++ b/docs/apis/programming_guide.md @@ -2213,19 +2213,19 @@ DataSet> pData = // [...] DataSet sData = // [...] // sort output on String field in ascending order -tData.print().sortLocalOutput(1, Order.ASCENDING); +tData.sortPartition(1, Order.ASCENDING).print(); // sort output on Double field in descending and Integer field in ascending order -tData.print().sortLocalOutput(2, Order.DESCENDING).sortLocalOutput(0, Order.ASCENDING); +tData.sortPartition(2, Order.DESCENDING).sortPartition(0, Order.ASCENDING).print(); // sort output on the "author" field of nested BookPojo in descending order -pData.writeAsText(...).sortLocalOutput("f0.author", Order.DESCENDING); +pData.sortPartition("f0.author", Order.DESCENDING).writeAsText(...); // sort output on the full tuple in ascending order -tData.writeAsCsv(...).sortLocalOutput("*", Order.ASCENDING); +tData.sortPartition("*", Order.ASCENDING).writeAsCsv(...); // sort atomic type (String) output in descending order -sData.writeAsText(...).sortLocalOutput("*", Order.DESCENDING); +sData.sortPartition("*", Order.DESCENDING).writeAsText(...); {% endhighlight %} @@ -2296,19 +2296,19 @@ val pData: DataSet[(BookPojo, Double)] = // [...] val sData: DataSet[String] = // [...] // sort output on String field in ascending order -tData.print.sortLocalOutput(1, Order.ASCENDING); +tData.sortPartition(1, Order.ASCENDING).print; // sort output on Double field in descending and Int field in ascending order -tData.print.sortLocalOutput(2, Order.DESCENDING).sortLocalOutput(0, Order.ASCENDING); +tData.sortPartition(2, Order.DESCENDING).sortPartition(0, Order.ASCENDING).print; // sort output on the "author" field of nested BookPojo in descending order -pData.writeAsText(...).sortLocalOutput("_1.author", Order.DESCENDING); +pData.sortPartition("_1.author", Order.DESCENDING).writeAsText(...); // sort output on the full tuple in ascending order -tData.writeAsCsv(...).sortLocalOutput("_", Order.ASCENDING); +tData.sortPartition("_", Order.ASCENDING).writeAsCsv(...); // sort atomic type (String) output in descending order -sData.writeAsText(...).sortLocalOutput("_", Order.DESCENDING); +sData.sortPartition("_", Order.DESCENDING).writeAsText(...); {% endhighlight %} diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java index 74b0d017bf06b6..d6451372aeb2aa 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java @@ -40,7 +40,7 @@ public abstract class AbstractUdfOperator extends Oper /** * The extra inputs which parameterize the user function. */ - protected final Map> broadcastInputs = new HashMap>(); + protected final Map> broadcastInputs = new HashMap<>(); // -------------------------------------------------------------------------------------------- @@ -137,7 +137,7 @@ public void setBroadcastVariables(Map> inputs) { * @param clazz The class object to be wrapped. * @return An array wrapping the class object. */ - protected static final Class[] asArray(Class clazz) { + protected static Class[] asArray(Class clazz) { @SuppressWarnings("unchecked") Class[] array = new Class[] { clazz }; return array; @@ -149,7 +149,7 @@ protected static final Class[] asArray(Class clazz) { * @param The type of the classes. * @return An empty array of type Class<U>. */ - protected static final Class[] emptyClassArray() { + protected static Class[] emptyClassArray() { @SuppressWarnings("unchecked") Class[] array = new Class[0]; return array; diff --git a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java index fd4e05072dfdba..c5a636c073c9dc 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java @@ -211,7 +211,7 @@ public MapOperator map(MapFunction mapper) { String callLocation = Utils.getCallLocationName(); TypeInformation resultType = TypeExtractor.getMapReturnTypes(mapper, getType(), callLocation, true); - return new MapOperator(this, resultType, clean(mapper), callLocation); + return new MapOperator<>(this, resultType, clean(mapper), callLocation); } @@ -240,7 +240,7 @@ public MapPartitionOperator mapPartition(MapPartitionFunction ma String callLocation = Utils.getCallLocationName(); TypeInformation resultType = TypeExtractor.getMapPartitionReturnTypes(mapPartition, getType(), callLocation, true); - return new MapPartitionOperator(this, resultType, clean(mapPartition), callLocation); + return new MapPartitionOperator<>(this, resultType, clean(mapPartition), callLocation); } /** @@ -262,7 +262,7 @@ public FlatMapOperator flatMap(FlatMapFunction flatMapper) { String callLocation = Utils.getCallLocationName(); TypeInformation resultType = TypeExtractor.getFlatMapReturnTypes(flatMapper, getType(), callLocation, true); - return new FlatMapOperator(this, resultType, clean(flatMapper), callLocation); + return new FlatMapOperator<>(this, resultType, clean(flatMapper), callLocation); } /** @@ -282,7 +282,7 @@ public FilterOperator filter(FilterFunction filter) { if (filter == null) { throw new NullPointerException("Filter function must not be null."); } - return new FilterOperator(this, clean(filter), Utils.getCallLocationName()); + return new FilterOperator<>(this, clean(filter), Utils.getCallLocationName()); } @@ -307,7 +307,7 @@ public FilterOperator filter(FilterFunction filter) { * @see ProjectOperator */ public ProjectOperator project(int... fieldIndexes) { - return new Projection(this, fieldIndexes).projectTupleX(); + return new Projection<>(this, fieldIndexes).projectTupleX(); } // -------------------------------------------------------------------------------------------- @@ -331,7 +331,7 @@ public ProjectOperator project(int... fieldIndexes) * @see DataSet */ public AggregateOperator aggregate(Aggregations agg, int field) { - return new AggregateOperator(this, agg, field, Utils.getCallLocationName()); + return new AggregateOperator<>(this, agg, field, Utils.getCallLocationName()); } /** @@ -405,7 +405,7 @@ public List collect() throws Exception { final String id = new AbstractID().toString(); final TypeSerializer serializer = getType().createSerializer(getExecutionEnvironment().getConfig()); - this.flatMap(new Utils.CollectHelper(id, serializer)).name("collect()") + this.flatMap(new Utils.CollectHelper<>(id, serializer)).name("collect()") .output(new DiscardingOutputFormat()).name("collect() sink"); JobExecutionResult res = getExecutionEnvironment().execute(); @@ -440,7 +440,7 @@ public ReduceOperator reduce(ReduceFunction reducer) { if (reducer == null) { throw new NullPointerException("Reduce function must not be null."); } - return new ReduceOperator(this, clean(reducer), Utils.getCallLocationName()); + return new ReduceOperator<>(this, clean(reducer), Utils.getCallLocationName()); } /** @@ -463,7 +463,7 @@ public GroupReduceOperator reduceGroup(GroupReduceFunction reduc String callLocation = Utils.getCallLocationName(); TypeInformation resultType = TypeExtractor.getGroupReduceReturnTypes(reducer, getType(), callLocation, true); - return new GroupReduceOperator(this, resultType, clean(reducer), callLocation); + return new GroupReduceOperator<>(this, resultType, clean(reducer), callLocation); } /** @@ -485,7 +485,7 @@ public GroupCombineOperator combineGroup(GroupCombineFunction co String callLocation = Utils.getCallLocationName(); TypeInformation resultType = TypeExtractor.getGroupCombineReturnTypes(combiner, getType(), callLocation, true); - return new GroupCombineOperator(this, resultType, clean(combiner), callLocation); + return new GroupCombineOperator<>(this, resultType, clean(combiner), callLocation); } /** @@ -520,7 +520,7 @@ public ReduceOperator minBy(int... fields) { throw new InvalidProgramException("DataSet#minBy(int...) only works on Tuple types."); } - return new ReduceOperator(this, new SelectByMinFunction( + return new ReduceOperator<>(this, new SelectByMinFunction( (TupleTypeInfo) getType(), fields), Utils.getCallLocationName()); } @@ -556,7 +556,7 @@ public ReduceOperator maxBy(int... fields) { throw new InvalidProgramException("DataSet#maxBy(int...) only works on Tuple types."); } - return new ReduceOperator(this, new SelectByMaxFunction( + return new ReduceOperator<>(this, new SelectByMaxFunction( (TupleTypeInfo) getType(), fields), Utils.getCallLocationName()); } @@ -589,7 +589,7 @@ public GroupReduceOperator first(int n) { */ public DistinctOperator distinct(KeySelector keyExtractor) { TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, getType()); - return new DistinctOperator(this, new Keys.SelectorFunctionKeys(keyExtractor, getType(), keyType), Utils.getCallLocationName()); + return new DistinctOperator<>(this, new Keys.SelectorFunctionKeys<>(keyExtractor, getType(), keyType), Utils.getCallLocationName()); } /** @@ -604,7 +604,7 @@ public DistinctOperator distinct(KeySelector keyExtractor) { * @return A DistinctOperator that represents the distinct DataSet. */ public DistinctOperator distinct(int... fields) { - return new DistinctOperator(this, new Keys.ExpressionKeys(fields, getType(), true), Utils.getCallLocationName()); + return new DistinctOperator<>(this, new Keys.ExpressionKeys<>(fields, getType(), true), Utils.getCallLocationName()); } /** @@ -618,7 +618,7 @@ public DistinctOperator distinct(int... fields) { * @return A DistinctOperator that represents the distinct DataSet. */ public DistinctOperator distinct(String... fields) { - return new DistinctOperator(this, new Keys.ExpressionKeys(fields, getType()), Utils.getCallLocationName()); + return new DistinctOperator<>(this, new Keys.ExpressionKeys<>(fields, getType()), Utils.getCallLocationName()); } /** @@ -630,7 +630,7 @@ public DistinctOperator distinct(String... fields) { * @return A DistinctOperator that represents the distinct DataSet. */ public DistinctOperator distinct() { - return new DistinctOperator(this, null, Utils.getCallLocationName()); + return new DistinctOperator<>(this, null, Utils.getCallLocationName()); } // -------------------------------------------------------------------------------------------- @@ -662,7 +662,7 @@ public DistinctOperator distinct() { */ public UnsortedGrouping groupBy(KeySelector keyExtractor) { TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, getType()); - return new UnsortedGrouping(this, new Keys.SelectorFunctionKeys(clean(keyExtractor), getType(), keyType)); + return new UnsortedGrouping<>(this, new Keys.SelectorFunctionKeys<>(clean(keyExtractor), getType(), keyType)); } /** @@ -689,7 +689,7 @@ public UnsortedGrouping groupBy(KeySelector keyExtractor) { * @see DataSet */ public UnsortedGrouping groupBy(int... fields) { - return new UnsortedGrouping(this, new Keys.ExpressionKeys(fields, getType(), false)); + return new UnsortedGrouping<>(this, new Keys.ExpressionKeys<>(fields, getType(), false)); } /** @@ -716,7 +716,7 @@ public UnsortedGrouping groupBy(int... fields) { * @see DataSet */ public UnsortedGrouping groupBy(String... fields) { - return new UnsortedGrouping(this, new Keys.ExpressionKeys(fields, getType())); + return new UnsortedGrouping<>(this, new Keys.ExpressionKeys<>(fields, getType())); } // -------------------------------------------------------------------------------------------- @@ -739,7 +739,7 @@ public UnsortedGrouping groupBy(String... fields) { * @see DataSet */ public JoinOperatorSets join(DataSet other) { - return new JoinOperatorSets(this, other); + return new JoinOperatorSets<>(this, other); } /** @@ -760,7 +760,7 @@ public JoinOperatorSets join(DataSet other) { * @see DataSet */ public JoinOperatorSets join(DataSet other, JoinHint strategy) { - return new JoinOperatorSets(this, other, strategy); + return new JoinOperatorSets<>(this, other, strategy); } /** @@ -781,7 +781,7 @@ public JoinOperatorSets join(DataSet other, JoinHint strategy) { * @see DataSet */ public JoinOperatorSets joinWithTiny(DataSet other) { - return new JoinOperatorSets(this, other, JoinHint.BROADCAST_HASH_SECOND); + return new JoinOperatorSets<>(this, other, JoinHint.BROADCAST_HASH_SECOND); } /** @@ -801,7 +801,7 @@ public JoinOperatorSets joinWithTiny(DataSet other) { * @see DataSet */ public JoinOperatorSets joinWithHuge(DataSet other) { - return new JoinOperatorSets(this, other, JoinHint.BROADCAST_HASH_FIRST); + return new JoinOperatorSets<>(this, other, JoinHint.BROADCAST_HASH_FIRST); } /** @@ -972,7 +972,7 @@ public JoinOperatorSetsBase fullOuterJoin(DataSet other, JoinHint s * @see DataSet */ public CoGroupOperator.CoGroupOperatorSets coGroup(DataSet other) { - return new CoGroupOperator.CoGroupOperatorSets(this, other); + return new CoGroupOperator.CoGroupOperatorSets<>(this, other); } // -------------------------------------------------------------------------------------------- @@ -1017,7 +1017,7 @@ public CoGroupOperator.CoGroupOperatorSets coGroup(DataSet other) { * @see Tuple2 */ public CrossOperator.DefaultCross cross(DataSet other) { - return new CrossOperator.DefaultCross(this, other, CrossHint.OPTIMIZER_CHOOSES, Utils.getCallLocationName()); + return new CrossOperator.DefaultCross<>(this, other, CrossHint.OPTIMIZER_CHOOSES, Utils.getCallLocationName()); } /** @@ -1047,7 +1047,7 @@ public CrossOperator.DefaultCross cross(DataSet other) { * @see Tuple2 */ public CrossOperator.DefaultCross crossWithTiny(DataSet other) { - return new CrossOperator.DefaultCross(this, other, CrossHint.SECOND_IS_SMALL, Utils.getCallLocationName()); + return new CrossOperator.DefaultCross<>(this, other, CrossHint.SECOND_IS_SMALL, Utils.getCallLocationName()); } /** @@ -1077,7 +1077,7 @@ public CrossOperator.DefaultCross crossWithTiny(DataSet other) { * @see Tuple2 */ public CrossOperator.DefaultCross crossWithHuge(DataSet other) { - return new CrossOperator.DefaultCross(this, other, CrossHint.FIRST_IS_SMALL, Utils.getCallLocationName()); + return new CrossOperator.DefaultCross<>(this, other, CrossHint.FIRST_IS_SMALL, Utils.getCallLocationName()); } // -------------------------------------------------------------------------------------------- @@ -1115,7 +1115,7 @@ public CrossOperator.DefaultCross crossWithHuge(DataSet other) { * @see org.apache.flink.api.java.operators.IterativeDataSet */ public IterativeDataSet iterate(int maxIterations) { - return new IterativeDataSet(getExecutionEnvironment(), getType(), this, maxIterations); + return new IterativeDataSet<>(getExecutionEnvironment(), getType(), this, maxIterations); } /** @@ -1168,8 +1168,8 @@ public DeltaIteration iterateDelta(DataSet workset, int maxIteratio Preconditions.checkNotNull(workset); Preconditions.checkNotNull(keyPositions); - Keys.ExpressionKeys keys = new Keys.ExpressionKeys(keyPositions, getType(), false); - return new DeltaIteration(getExecutionEnvironment(), getType(), this, workset, keys, maxIterations); + Keys.ExpressionKeys keys = new Keys.ExpressionKeys<>(keyPositions, getType(), false); + return new DeltaIteration<>(getExecutionEnvironment(), getType(), this, workset, keys, maxIterations); } // -------------------------------------------------------------------------------------------- @@ -1201,7 +1201,7 @@ public DataSet runOperation(CustomUnaryOperation operation) { * @return The resulting DataSet. */ public UnionOperator union(DataSet other){ - return new UnionOperator(this, other, Utils.getCallLocationName()); + return new UnionOperator<>(this, other, Utils.getCallLocationName()); } // -------------------------------------------------------------------------------------------- @@ -1217,7 +1217,7 @@ public UnionOperator union(DataSet other){ * @return The partitioned DataSet. */ public PartitionOperator partitionByHash(int... fields) { - return new PartitionOperator(this, PartitionMethod.HASH, new Keys.ExpressionKeys(fields, getType(), false), Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.HASH, new Keys.ExpressionKeys<>(fields, getType(), false), Utils.getCallLocationName()); } /** @@ -1229,7 +1229,7 @@ public PartitionOperator partitionByHash(int... fields) { * @return The partitioned DataSet. */ public PartitionOperator partitionByHash(String... fields) { - return new PartitionOperator(this, PartitionMethod.HASH, new Keys.ExpressionKeys(fields, getType()), Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.HASH, new Keys.ExpressionKeys<>(fields, getType()), Utils.getCallLocationName()); } /** @@ -1244,7 +1244,7 @@ public PartitionOperator partitionByHash(String... fields) { */ public > PartitionOperator partitionByHash(KeySelector keyExtractor) { final TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, getType()); - return new PartitionOperator(this, PartitionMethod.HASH, new Keys.SelectorFunctionKeys(clean(keyExtractor), this.getType(), keyType), Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.HASH, new Keys.SelectorFunctionKeys<>(clean(keyExtractor), this.getType(), keyType), Utils.getCallLocationName()); } /** @@ -1257,7 +1257,7 @@ public > PartitionOperator partitionByHash(KeySelecto * @return The partitioned DataSet. */ public PartitionOperator partitionByRange(int... fields) { - return new PartitionOperator(this, PartitionMethod.RANGE, new Keys.ExpressionKeys(fields, getType(), false), Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.RANGE, new Keys.ExpressionKeys<>(fields, getType(), false), Utils.getCallLocationName()); } /** @@ -1270,7 +1270,7 @@ public PartitionOperator partitionByRange(int... fields) { * @return The partitioned DataSet. */ public PartitionOperator partitionByRange(String... fields) { - return new PartitionOperator(this, PartitionMethod.RANGE, new Keys.ExpressionKeys(fields, getType()), Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.RANGE, new Keys.ExpressionKeys<>(fields, getType()), Utils.getCallLocationName()); } /** @@ -1286,7 +1286,7 @@ public PartitionOperator partitionByRange(String... fields) { */ public > PartitionOperator partitionByRange(KeySelector keyExtractor) { final TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, getType()); - return new PartitionOperator(this, PartitionMethod.RANGE, new Keys.SelectorFunctionKeys(clean(keyExtractor), this.getType(), keyType), Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.RANGE, new Keys.SelectorFunctionKeys<>(clean(keyExtractor), this.getType(), keyType), Utils.getCallLocationName()); } /** @@ -1300,7 +1300,7 @@ public > PartitionOperator partitionByRange(KeySelect * @return The partitioned DataSet. */ public PartitionOperator partitionCustom(Partitioner partitioner, int field) { - return new PartitionOperator(this, new Keys.ExpressionKeys(new int[] {field}, getType(), false), clean(partitioner), Utils.getCallLocationName()); + return new PartitionOperator<>(this, new Keys.ExpressionKeys<>(new int[] {field}, getType(), false), clean(partitioner), Utils.getCallLocationName()); } /** @@ -1314,7 +1314,7 @@ public PartitionOperator partitionCustom(Partitioner partitioner, int * @return The partitioned DataSet. */ public PartitionOperator partitionCustom(Partitioner partitioner, String field) { - return new PartitionOperator(this, new Keys.ExpressionKeys(new String[] {field}, getType()), clean(partitioner), Utils.getCallLocationName()); + return new PartitionOperator<>(this, new Keys.ExpressionKeys<>(new String[] {field}, getType()), clean(partitioner), Utils.getCallLocationName()); } /** @@ -1333,7 +1333,7 @@ public PartitionOperator partitionCustom(Partitioner partitioner, Stri */ public > PartitionOperator partitionCustom(Partitioner partitioner, KeySelector keyExtractor) { final TypeInformation keyType = TypeExtractor.getKeySelectorTypes(keyExtractor, getType()); - return new PartitionOperator(this, new Keys.SelectorFunctionKeys(keyExtractor, getType(), keyType), clean(partitioner), Utils.getCallLocationName()); + return new PartitionOperator<>(this, new Keys.SelectorFunctionKeys<>(keyExtractor, getType(), keyType), clean(partitioner), Utils.getCallLocationName()); } /** @@ -1345,7 +1345,7 @@ public > PartitionOperator partitionCustom(Partitione * @return The re-balanced DataSet. */ public PartitionOperator rebalance() { - return new PartitionOperator(this, PartitionMethod.REBALANCE, Utils.getCallLocationName()); + return new PartitionOperator<>(this, PartitionMethod.REBALANCE, Utils.getCallLocationName()); } // -------------------------------------------------------------------------------------------- @@ -1361,7 +1361,7 @@ public PartitionOperator rebalance() { * @return The DataSet with sorted local partitions. */ public SortPartitionOperator sortPartition(int field, Order order) { - return new SortPartitionOperator(this, field, order, Utils.getCallLocationName()); + return new SortPartitionOperator<>(this, field, order, Utils.getCallLocationName()); } /** @@ -1373,7 +1373,7 @@ public SortPartitionOperator sortPartition(int field, Order order) { * @return The DataSet with sorted local partitions. */ public SortPartitionOperator sortPartition(String field, Order order) { - return new SortPartitionOperator(this, field, order, Utils.getCallLocationName()); + return new SortPartitionOperator<>(this, field, order, Utils.getCallLocationName()); } // -------------------------------------------------------------------------------------------- @@ -1448,7 +1448,7 @@ public DataSink writeAsText(String filePath) { * @see DataSet#writeAsText(String) Output files and directories */ public DataSink writeAsText(String filePath, WriteMode writeMode) { - TextOutputFormat tof = new TextOutputFormat(new Path(filePath)); + TextOutputFormat tof = new TextOutputFormat<>(new Path(filePath)); tof.setWriteMode(writeMode); return output(tof); } @@ -1465,7 +1465,7 @@ public DataSink writeAsText(String filePath, WriteMode writeMode) { * @see DataSet#writeAsText(String) Output files and directories */ public DataSink writeAsFormattedText(String filePath, TextFormatter formatter) { - return map(new FormattingMapper(clean(formatter))).writeAsText(filePath); + return map(new FormattingMapper<>(clean(formatter))).writeAsText(filePath); } /** @@ -1481,7 +1481,7 @@ public DataSink writeAsFormattedText(String filePath, TextFormatter f * @see DataSet#writeAsText(String) Output files and directories */ public DataSink writeAsFormattedText(String filePath, WriteMode writeMode, TextFormatter formatter) { - return map(new FormattingMapper(clean(formatter))).writeAsText(filePath, writeMode); + return map(new FormattingMapper<>(clean(formatter))).writeAsText(filePath, writeMode); } /** @@ -1559,7 +1559,7 @@ public DataSink writeAsCsv(String filePath, String rowDelimiter, String field @SuppressWarnings("unchecked") private DataSink internalWriteAsCsv(Path filePath, String rowDelimiter, String fieldDelimiter, WriteMode wm) { Preconditions.checkArgument(getType().isTupleType(), "The writeAsCsv() method can only be used on data sets of tuples."); - CsvOutputFormat of = new CsvOutputFormat(filePath, rowDelimiter, fieldDelimiter); + CsvOutputFormat of = new CsvOutputFormat<>(filePath, rowDelimiter, fieldDelimiter); if(wm != null) { of.setWriteMode(wm); } @@ -1715,7 +1715,7 @@ public DataSink output(OutputFormat outputFormat) { ((InputTypeConfigurable) outputFormat).setInputType(getType(), context.getConfig() ); } - DataSink sink = new DataSink(this, outputFormat, getType()); + DataSink sink = new DataSink<>(this, outputFormat, getType()); this.context.registerDataSink(sink); return sink; } diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java index bf9ca4c8ca9598..665f35fde7166f 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java @@ -62,7 +62,7 @@ public static String getCallLocationName(int depth) { /** * Returns all GenericTypeInfos contained in a composite type. * - * @param typeInfo + * @param typeInfo {@link CompositeType} */ public static void getContainedGenericTypes(CompositeType typeInfo, List> target) { for(int i = 0; i < typeInfo.getArity(); i++) { @@ -118,7 +118,7 @@ public CollectHelper(String id, TypeSerializer serializer) { @Override public void open(Configuration parameters) throws Exception { - this.accumulator = new SerializedListAccumulator(); + this.accumulator = new SerializedListAccumulator<>(); } @Override diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java index 48209cf3c0733c..3371665c76063e 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java @@ -30,7 +30,6 @@ import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.java.typeutils.TupleTypeInfoBase; import org.apache.flink.configuration.Configuration; -import org.apache.flink.types.Nothing; import org.apache.flink.api.java.DataSet; import java.util.Arrays; @@ -107,6 +106,7 @@ public DataSink withParameters(Configuration parameters) { * @see org.apache.flink.api.java.tuple.Tuple * @see Order */ + @Deprecated public DataSink sortLocalOutput(int field, Order order) { if (!this.type.isTupleType()) { @@ -120,7 +120,7 @@ public DataSink sortLocalOutput(int field, Order order) { // get flat keys Keys.ExpressionKeys ek; try { - ek = new Keys.ExpressionKeys(new int[]{field}, this.type); + ek = new Keys.ExpressionKeys<>(new int[]{field}, this.type); } catch(IllegalArgumentException iae) { throw new InvalidProgramException("Invalid specification of field expression.", iae); } @@ -161,6 +161,7 @@ public DataSink sortLocalOutput(int field, Order order) { * * @see Order */ + @Deprecated public DataSink sortLocalOutput(String fieldExpression, Order order) { int numFields; @@ -173,7 +174,7 @@ public DataSink sortLocalOutput(String fieldExpression, Order order) { Keys.ExpressionKeys ek; try { isValidSortKeyType(fieldExpression); - ek = new Keys.ExpressionKeys(new String[]{fieldExpression}, this.type); + ek = new Keys.ExpressionKeys<>(new String[]{fieldExpression}, this.type); } catch(IllegalArgumentException iae) { throw new InvalidProgramException("Invalid specification of field expression.", iae); } @@ -255,7 +256,7 @@ public DataSink name(String name) { protected GenericDataSinkBase translateToDataFlow(Operator input) { // select the name (or create a default one) String name = this.name != null ? this.name : this.format.toString(); - GenericDataSinkBase sink = new GenericDataSinkBase(this.format, new UnaryOperatorInformation(this.type, new NothingTypeInfo()), name); + GenericDataSinkBase sink = new GenericDataSinkBase<>(this.format, new UnaryOperatorInformation<>(this.type, new NothingTypeInfo()), name); // set input sink.setInput(input); // set parameters diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java index 37ad381b656040..5024a0e375db01 100644 --- a/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java +++ b/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java @@ -20,6 +20,7 @@ import org.apache.flink.api.common.InvalidProgramException; import org.apache.flink.api.common.operators.Order; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.tuple.Tuple5; @@ -35,15 +36,15 @@ public class DataSinkTest { // TUPLE DATA - private final List> emptyTupleData = new ArrayList>(); + private final List> emptyTupleData = new ArrayList<>(); - private final TupleTypeInfo> tupleTypeInfo = new TupleTypeInfo>( + private final TupleTypeInfo> tupleTypeInfo = new TupleTypeInfo<>( BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO); // POJO DATA - private final List pojoData = new ArrayList(); + private final List pojoData = new ArrayList<>(); @Before public void fillPojoData() { @@ -62,7 +63,7 @@ public void testTupleSingleOrderIdx() { // should work try { - tupleDs.writeAsText("/tmp/willNotHappen").sortLocalOutput(0, Order.ANY); + tupleDs.sortPartition(0, Order.ANY).writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } @@ -78,9 +79,9 @@ public void testTupleTwoOrderIdx() { // should work try { - tupleDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput(0, Order.ASCENDING) - .sortLocalOutput(3, Order.DESCENDING); + tupleDs.sortPartition(0, Order.ASCENDING) + .sortPartition(3, Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } @@ -96,13 +97,13 @@ public void testTupleSingleOrderExp() { // should work try { - tupleDs.writeAsText("/tmp/willNotHappen").sortLocalOutput("f0", Order.ANY); + tupleDs.sortPartition("f0", Order.ANY).writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } } - @Test + @Test(expected = CompositeType.InvalidFieldReferenceException.class) public void testTupleSingleOrderExpFull() { final ExecutionEnvironment env = ExecutionEnvironment @@ -110,12 +111,8 @@ public void testTupleSingleOrderExpFull() { DataSet> tupleDs = env .fromCollection(emptyTupleData, tupleTypeInfo); - // should work - try { - tupleDs.writeAsText("/tmp/willNotHappen").sortLocalOutput("*", Order.ANY); - } catch (Exception e) { - Assert.fail(); - } + // should not work + tupleDs.sortPartition("*", Order.ANY).writeAsText("/tmp/willNotHappen"); } @Test @@ -128,9 +125,9 @@ public void testTupleTwoOrderExp() { // should work try { - tupleDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("f1", Order.ASCENDING) - .sortLocalOutput("f4", Order.DESCENDING); + tupleDs.sortPartition("f1", Order.ASCENDING) + .sortPartition("f4", Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } @@ -146,15 +143,15 @@ public void testTupleTwoOrderMixed() { // should work try { - tupleDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput(4, Order.ASCENDING) - .sortLocalOutput("f2", Order.DESCENDING); + tupleDs.sortPartition(4, Order.ASCENDING) + .sortPartition("f2", Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } } - @Test(expected = InvalidProgramException.class) + @Test(expected = IndexOutOfBoundsException.class) public void testFailTupleIndexOutOfBounds() { final ExecutionEnvironment env = ExecutionEnvironment @@ -163,12 +160,12 @@ public void testFailTupleIndexOutOfBounds() { .fromCollection(emptyTupleData, tupleTypeInfo); // must not work - tupleDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput(3, Order.ASCENDING) - .sortLocalOutput(5, Order.DESCENDING); + tupleDs.sortPartition(3, Order.ASCENDING) + .sortPartition(5, Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } - @Test(expected = InvalidProgramException.class) + @Test(expected = CompositeType.InvalidFieldReferenceException.class) public void testFailTupleInv() { final ExecutionEnvironment env = ExecutionEnvironment @@ -177,9 +174,9 @@ public void testFailTupleInv() { .fromCollection(emptyTupleData, tupleTypeInfo); // must not work - tupleDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("notThere", Order.ASCENDING) - .sortLocalOutput("f4", Order.DESCENDING); + tupleDs.sortPartition("notThere", Order.ASCENDING) + .sortPartition("f4", Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } @Test @@ -192,8 +189,7 @@ public void testPrimitiveOrder() { // should work try { - longDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("*", Order.ASCENDING); + longDs.sortPartition("*", Order.ASCENDING).writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } @@ -208,8 +204,7 @@ public void testFailPrimitiveOrder1() { .generateSequence(0,2); // must not work - longDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput(0, Order.ASCENDING); + longDs.sortPartition(0, Order.ASCENDING).writeAsText("/tmp/willNotHappen"); } @Test(expected = InvalidProgramException.class) @@ -221,8 +216,7 @@ public void testFailPrimitiveOrder2() { .generateSequence(0,2); // must not work - longDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("0", Order.ASCENDING); + longDs.sortPartition("0", Order.ASCENDING).writeAsText("/tmp/willNotHappen"); } @Test(expected = InvalidProgramException.class) @@ -234,8 +228,7 @@ public void testFailPrimitiveOrder3() { .generateSequence(0,2); // must not work - longDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("nope", Order.ASCENDING); + longDs.sortPartition("nope", Order.ASCENDING).writeAsText("/tmp/willNotHappen"); } @Test @@ -248,8 +241,7 @@ public void testPojoSingleOrder() { // should work try { - pojoDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("myString", Order.ASCENDING); + pojoDs.sortPartition("myString", Order.ASCENDING).writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } @@ -265,9 +257,9 @@ public void testPojoTwoOrder() { // should work try { - pojoDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("myLong", Order.ASCENDING) - .sortLocalOutput("myString", Order.DESCENDING); + pojoDs.sortPartition("myLong", Order.ASCENDING) + .sortPartition("myString", Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } catch (Exception e) { Assert.fail(); } @@ -282,11 +274,10 @@ public void testFailPojoIdx() { .fromCollection(pojoData); // must not work - pojoDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput(1, Order.DESCENDING); + pojoDs.sortPartition(1, Order.DESCENDING).writeAsText("/tmp/willNotHappen"); } - @Test(expected = InvalidProgramException.class) + @Test(expected = CompositeType.InvalidFieldReferenceException.class) public void testFailPojoInvalidField() { final ExecutionEnvironment env = ExecutionEnvironment @@ -295,12 +286,12 @@ public void testFailPojoInvalidField() { .fromCollection(pojoData); // must not work - pojoDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("myInt", Order.ASCENDING) - .sortLocalOutput("notThere", Order.DESCENDING); + pojoDs.sortPartition("myInt", Order.ASCENDING) + .sortPartition("notThere", Order.DESCENDING) + .writeAsText("/tmp/willNotHappen"); } - @Test(expected = InvalidProgramException.class) + @Test(expected = CompositeType.InvalidFieldReferenceException.class) public void testPojoSingleOrderFull() { final ExecutionEnvironment env = ExecutionEnvironment @@ -309,14 +300,14 @@ public void testPojoSingleOrderFull() { .fromCollection(pojoData); // must not work - pojoDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("*", Order.ASCENDING); + pojoDs.sortPartition("*", Order.ASCENDING) + .writeAsText("/tmp/willNotHappen"); } @Test(expected = InvalidProgramException.class) public void testArrayOrderFull() { - List arrayData = new ArrayList(); + List arrayData = new ArrayList<>(); arrayData.add(new Object[0]); final ExecutionEnvironment env = ExecutionEnvironment @@ -325,8 +316,8 @@ public void testArrayOrderFull() { .fromCollection(arrayData); // must not work - pojoDs.writeAsText("/tmp/willNotHappen") - .sortLocalOutput("*", Order.ASCENDING); + pojoDs.sortPartition("*", Order.ASCENDING) + .writeAsText("/tmp/willNotHappen"); } /** @@ -341,7 +332,7 @@ public static class CustomType implements Serializable { public String myString; public CustomType() { - }; + } public CustomType(int i, long l, String s) { myInt = i; diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java index 6dc0a0bb3b1a7d..dcbb451ccf58c7 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java @@ -46,9 +46,9 @@ protected void testProgram() throws Exception { IterativeDataSet iteration = env.fromElements(1, 2, 3).iterate(NUM_ITERATIONS); - iteration.closeWith(iteration.reduceGroup(new SumReducer())).output(new DiscardingOutputFormat()); + iteration.closeWith(iteration.reduceGroup(new SumReducer())).output(new DiscardingOutputFormat()); - Assert.assertEquals(Integer.valueOf(NUM_ITERATIONS * 6), (Integer)env.execute().getAccumulatorResult(ACC_NAME)); + Assert.assertEquals(NUM_ITERATIONS * 6, env.execute().getAccumulatorResult(ACC_NAME)); } static final class SumReducer extends RichGroupReduceFunction { diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/DataSinkITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/DataSinkITCase.java index b49bd33581aeb5..2e97424223cbf6 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/DataSinkITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/DataSinkITCase.java @@ -64,7 +64,7 @@ public void testIntSortingParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet ds = CollectionDataSets.getIntegerDataSet(env); - ds.writeAsText(resultPath).sortLocalOutput("*", Order.DESCENDING).setParallelism(1); + ds.sortPartition("*", Order.DESCENDING).writeAsText(resultPath).setParallelism(1); env.execute(); @@ -78,7 +78,7 @@ public void testStringSortingParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet ds = CollectionDataSets.getStringDataSet(env); - ds.writeAsText(resultPath).sortLocalOutput("*", Order.ASCENDING).setParallelism(1); + ds.sortPartition("*", Order.ASCENDING).writeAsText(resultPath).setParallelism(1); env.execute(); @@ -99,7 +99,7 @@ public void testTupleSortingSingleAscParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet> ds = CollectionDataSets.get3TupleDataSet(env); - ds.writeAsCsv(resultPath).sortLocalOutput(0, Order.ASCENDING).setParallelism(1); + ds.sortPartition(0, Order.ASCENDING).writeAsCsv(resultPath).setParallelism(1); env.execute(); @@ -133,7 +133,7 @@ public void testTupleSortingSingleDescParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet> ds = CollectionDataSets.get3TupleDataSet(env); - ds.writeAsCsv(resultPath).sortLocalOutput(0, Order.DESCENDING).setParallelism(1); + ds.sortPartition(0, Order.DESCENDING).writeAsCsv(resultPath).setParallelism(1); env.execute(); @@ -167,7 +167,7 @@ public void testTupleSortingDualParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet> ds = CollectionDataSets.get3TupleDataSet(env); - ds.writeAsCsv(resultPath).sortLocalOutput(1, Order.DESCENDING).sortLocalOutput(0, Order.ASCENDING).setParallelism(1); + ds.sortPartition(1, Order.DESCENDING).sortPartition(0, Order.ASCENDING).writeAsCsv(resultPath).setParallelism(1); env.execute(); @@ -202,9 +202,9 @@ public void testTupleSortingNestedParallelism1() throws Exception { DataSet, String, Integer>> ds = CollectionDataSets.getGroupSortedNestedTupleDataSet2(env); - ds.writeAsText(resultPath) - .sortLocalOutput("f0.f1", Order.ASCENDING) - .sortLocalOutput("f1", Order.DESCENDING) + ds.sortPartition("f0.f1", Order.ASCENDING) + .sortPartition("f1", Order.DESCENDING) + .writeAsText(resultPath) .setParallelism(1); env.execute(); @@ -227,9 +227,9 @@ public void testTupleSortingNestedParallelism1_2() throws Exception { DataSet, String, Integer>> ds = CollectionDataSets.getGroupSortedNestedTupleDataSet2(env); - ds.writeAsText(resultPath) - .sortLocalOutput(1, Order.ASCENDING) - .sortLocalOutput(2, Order.DESCENDING) + ds.sortPartition(1, Order.ASCENDING) + .sortPartition(2, Order.DESCENDING) + .writeAsText(resultPath) .setParallelism(1); env.execute(); @@ -251,7 +251,7 @@ public void testPojoSortingSingleParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet ds = CollectionDataSets.getMixedPojoDataSet(env); - ds.writeAsText(resultPath).sortLocalOutput("number", Order.ASCENDING).setParallelism(1); + ds.sortPartition("number", Order.ASCENDING).writeAsText(resultPath).setParallelism(1); env.execute(); @@ -272,9 +272,9 @@ public void testPojoSortingDualParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet ds = CollectionDataSets.getMixedPojoDataSet(env); - ds.writeAsText(resultPath) - .sortLocalOutput("str", Order.ASCENDING) - .sortLocalOutput("number", Order.DESCENDING) + ds.sortPartition("str", Order.ASCENDING) + .sortPartition("number", Order.DESCENDING) + .writeAsText(resultPath) .setParallelism(1); env.execute(); @@ -298,10 +298,10 @@ public void testPojoSortingNestedParallelism1() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet ds = CollectionDataSets.getMixedPojoDataSet(env); - ds.writeAsText(resultPath) - .sortLocalOutput("nestedTupleWithCustom.f0", Order.ASCENDING) - .sortLocalOutput("nestedTupleWithCustom.f1.myInt", Order.DESCENDING) - .sortLocalOutput("nestedPojo.longNumber", Order.ASCENDING) + ds.sortPartition("nestedTupleWithCustom.f0", Order.ASCENDING) + .sortPartition("nestedTupleWithCustom.f1.myInt", Order.DESCENDING) + .sortPartition("nestedPojo.longNumber", Order.ASCENDING) + .writeAsText(resultPath) .setParallelism(1); env.execute(); @@ -327,13 +327,13 @@ public void testSortingParallelism4() throws Exception { // randomize ds.map(new MapFunction() { - Random rand = new Random(1234l); + Random rand = new Random(1234L); @Override public Long map(Long value) throws Exception { return rand.nextLong(); } - }).writeAsText(resultPath) - .sortLocalOutput("*", Order.ASCENDING) + }).sortPartition("*", Order.ASCENDING) + .writeAsText(resultPath) .setParallelism(4); env.execute(); diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupCombineITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupCombineITCase.java index 7e6de0448ec954..9b56c63a532ab8 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupCombineITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupCombineITCase.java @@ -253,7 +253,7 @@ public void combine(Iterable> values, Collector(key, count)); } }).collect(); @@ -293,7 +293,7 @@ public void combine(Iterable> values, Collector(key, count)); } }).collect(); @@ -316,7 +316,7 @@ public void testAPI() throws Exception { DataSet> ds = CollectionDataSets.getStringDataSet(env).map(new MapFunction>() { @Override public Tuple1 map(String value) throws Exception { - return new Tuple1(value); + return new Tuple1<>(value); } }); @@ -361,23 +361,25 @@ public static class IdentityFunction implements GroupCombineFunction> values, Collector> out) throws Exception { for (Tuple3 value : values) { - out.collect(new Tuple3(value.f0, value.f1, value.f2)); + out.collect(new Tuple3<>(value.f0, value.f1, value.f2)); } } @Override public void reduce(Iterable> values, Collector> out) throws Exception { for (Tuple3 value : values) { - out.collect(new Tuple3(value.f0, value.f1, value.f2)); + out.collect(new Tuple3<>(value.f0, value.f1, value.f2)); } } } - public static class Tuple3toTuple3GroupReduce implements KvGroupReduce, Tuple3, Tuple3> { + public static class Tuple3toTuple3GroupReduce implements KvGroupReduce, + Tuple3, Tuple3> { @Override - public void combine(Iterable>> values, Collector>> out) throws Exception { + public void combine(Iterable>> values, Collector>> out) throws Exception { int i = 0; long l = 0; long key = 0; @@ -390,20 +392,23 @@ public void combine(Iterable>> values l += extracted.f1; } - Tuple3 result = new Tuple3(i, l, "combined"); - out.collect(new Tuple2>(key, result)); + Tuple3 result = new Tuple3<>(i, l, "combined"); + out.collect(new Tuple2<>(key, result)); } @Override - public void reduce(Iterable values, Collector out) throws Exception { + public void reduce(Iterable>> values, + Collector>> out) throws Exception { combine(values, out); } } - public static class Tuple3toTuple2GroupReduce implements KvGroupReduce, Tuple2, Tuple2> { + public static class Tuple3toTuple2GroupReduce implements KvGroupReduce, + Tuple2, Tuple2> { @Override - public void combine(Iterable>> values, Collector>> out) throws Exception { + public void combine(Iterable>> values, Collector>> out) throws Exception { int i = 0; long l = 0; long key = 0; @@ -416,20 +421,23 @@ public void combine(Iterable>> values l += extracted.f1 + extracted.f2.length(); } - Tuple2 result = new Tuple2(i, l); - out.collect(new Tuple2>(key, result)); + Tuple2 result = new Tuple2<>(i, l); + out.collect(new Tuple2<>(key, result)); } @Override - public void reduce(Iterable>> values, Collector>> out) throws Exception { + public void reduce(Iterable>> values, Collector>> out) throws Exception { new Tuple2toTuple2GroupReduce().reduce(values, out); } } - public static class Tuple2toTuple2GroupReduce implements KvGroupReduce, Tuple2, Tuple2> { + public static class Tuple2toTuple2GroupReduce implements KvGroupReduce, + Tuple2, Tuple2> { @Override - public void combine(Iterable>> values, Collector>> out) throws Exception { + public void combine(Iterable>> values, Collector>> out) throws Exception { int i = 0; long l = 0; long key = 0; @@ -442,29 +450,33 @@ public void combine(Iterable>> values, Collec l += extracted.f1; } - Tuple2 result = new Tuple2(i, l); + Tuple2 result = new Tuple2<>(i, l); - out.collect(new Tuple2>(key, result)); + out.collect(new Tuple2<>(key, result)); } @Override - public void reduce(Iterable>> values, Collector>> out) throws Exception { + public void reduce(Iterable>> values, Collector>> out) throws Exception { combine(values, out); } } - public class Tuple3KvWrapper implements MapFunction, Tuple2>> { + public class Tuple3KvWrapper implements MapFunction, Tuple2>> { @Override public Tuple2> map(Tuple3 value) throws Exception { - return new Tuple2>(value.f1, value); + return new Tuple2<>(value.f1, value); } } - public interface CombineAndReduceGroup extends GroupCombineFunction, GroupReduceFunction { + public interface CombineAndReduceGroup extends GroupCombineFunction, + GroupReduceFunction { } - public interface KvGroupReduce extends CombineAndReduceGroup, Tuple2, Tuple2> { + public interface KvGroupReduce extends CombineAndReduceGroup, Tuple2, + Tuple2> { } } diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/SortPartitionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/SortPartitionITCase.java index 1de013f4be6f35..2423420bcc97c6 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/SortPartitionITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/SortPartitionITCase.java @@ -57,7 +57,7 @@ public void testSortPartitionByKeyField() throws Exception { DataSet> ds = CollectionDataSets.get3TupleDataSet(env); List> result = ds - .map(new IdMapper()).setParallelism(4) // parallelize input + .map(new IdMapper>()).setParallelism(4) // parallelize input .sortPartition(1, Order.DESCENDING) .mapPartition(new OrderCheckMapper<>(new Tuple3Checker())) .distinct().collect(); @@ -78,7 +78,7 @@ public void testSortPartitionByTwoKeyFields() throws Exception { DataSet> ds = CollectionDataSets.get5TupleDataSet(env); List> result = ds - .map(new IdMapper()).setParallelism(2) // parallelize input + .map(new IdMapper>()).setParallelism(2) // parallelize input .sortPartition(4, Order.ASCENDING) .sortPartition(2, Order.DESCENDING) .mapPartition(new OrderCheckMapper<>(new Tuple5Checker())) @@ -122,7 +122,7 @@ public void testSortPartitionByTwoFieldExpressions() throws Exception { DataSet> ds = CollectionDataSets.get5TupleDataSet(env); List> result = ds - .map(new IdMapper()).setParallelism(2) // parallelize input + .map(new IdMapper>()).setParallelism(2) // parallelize input .sortPartition("f4", Order.ASCENDING) .sortPartition("f2", Order.DESCENDING) .mapPartition(new OrderCheckMapper<>(new Tuple5Checker())) @@ -144,7 +144,7 @@ public void testSortPartitionByNestedFieldExpression() throws Exception { DataSet, String>> ds = CollectionDataSets.getGroupSortedNestedTupleDataSet(env); List> result = ds - .map(new IdMapper()).setParallelism(3) // parallelize input + .map(new IdMapper, String>>()).setParallelism(3) // parallelize input .sortPartition("f0.f1", Order.ASCENDING) .sortPartition("f1", Order.DESCENDING) .mapPartition(new OrderCheckMapper<>(new NestedTupleChecker())) @@ -166,7 +166,7 @@ public void testSortPartitionPojoByNestedFieldExpression() throws Exception { DataSet ds = CollectionDataSets.getMixedPojoDataSet(env); List> result = ds - .map(new IdMapper()).setParallelism(1) // parallelize input + .map(new IdMapper()).setParallelism(1) // parallelize input .sortPartition("nestedTupleWithCustom.f1.myString", Order.ASCENDING) .sortPartition("number", Order.DESCENDING) .mapPartition(new OrderCheckMapper<>(new PojoChecker())) @@ -197,9 +197,8 @@ public void testSortPartitionParallelismChange() throws Exception { compareResultAsText(result, expected); } - public static interface OrderChecker extends Serializable { - - public boolean inOrder(T t1, T t2); + public interface OrderChecker extends Serializable { + boolean inOrder(T t1, T t2); } @SuppressWarnings("serial") @@ -215,7 +214,7 @@ public static class Tuple5Checker implements OrderChecker t1, Tuple5 t2) { - return t1.f4 < t2.f4 || t1.f4 == t2.f4 && t1.f2 >= t2.f2; + return t1.f4 < t2.f4 || t1.f4.equals(t2.f4) && t1.f2 >= t2.f2; } } @@ -225,7 +224,7 @@ public static class NestedTupleChecker implements OrderChecker, String> t1, Tuple2, String> t2) { return t1.f0.f1 < t2.f0.f1 || - t1.f0.f1 == t2.f0.f1 && t1.f1.compareTo(t2.f1) >= 0; + t1.f0.f1.equals(t2.f0.f1) && t1.f1.compareTo(t2.f1) >= 0; } } @@ -256,7 +255,6 @@ public void mapPartition(Iterable values, Collector> out) thr Iterator it = values.iterator(); if(!it.hasNext()) { out.collect(new Tuple1<>(true)); - return; } else { T last = it.next(); diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java index 9fb275f4b086b6..ba48e121a168dd 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java +++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java @@ -56,28 +56,28 @@ public class CollectionDataSets { public static DataSet> get3TupleDataSet(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple3(1, 1l, "Hi")); - data.add(new Tuple3(2, 2l, "Hello")); - data.add(new Tuple3(3, 2l, "Hello world")); - data.add(new Tuple3(4, 3l, "Hello world, how are you?")); - data.add(new Tuple3(5, 3l, "I am fine.")); - data.add(new Tuple3(6, 3l, "Luke Skywalker")); - data.add(new Tuple3(7, 4l, "Comment#1")); - data.add(new Tuple3(8, 4l, "Comment#2")); - data.add(new Tuple3(9, 4l, "Comment#3")); - data.add(new Tuple3(10, 4l, "Comment#4")); - data.add(new Tuple3(11, 5l, "Comment#5")); - data.add(new Tuple3(12, 5l, "Comment#6")); - data.add(new Tuple3(13, 5l, "Comment#7")); - data.add(new Tuple3(14, 5l, "Comment#8")); - data.add(new Tuple3(15, 5l, "Comment#9")); - data.add(new Tuple3(16, 6l, "Comment#10")); - data.add(new Tuple3(17, 6l, "Comment#11")); - data.add(new Tuple3(18, 6l, "Comment#12")); - data.add(new Tuple3(19, 6l, "Comment#13")); - data.add(new Tuple3(20, 6l, "Comment#14")); - data.add(new Tuple3(21, 6l, "Comment#15")); + List> data = new ArrayList<>(); + data.add(new Tuple3<>(1, 1L, "Hi")); + data.add(new Tuple3<>(2, 2L, "Hello")); + data.add(new Tuple3<>(3, 2L, "Hello world")); + data.add(new Tuple3<>(4, 3L, "Hello world, how are you?")); + data.add(new Tuple3<>(5, 3L, "I am fine.")); + data.add(new Tuple3<>(6, 3L, "Luke Skywalker")); + data.add(new Tuple3<>(7, 4L, "Comment#1")); + data.add(new Tuple3<>(8, 4L, "Comment#2")); + data.add(new Tuple3<>(9, 4L, "Comment#3")); + data.add(new Tuple3<>(10, 4L, "Comment#4")); + data.add(new Tuple3<>(11, 5L, "Comment#5")); + data.add(new Tuple3<>(12, 5L, "Comment#6")); + data.add(new Tuple3<>(13, 5L, "Comment#7")); + data.add(new Tuple3<>(14, 5L, "Comment#8")); + data.add(new Tuple3<>(15, 5L, "Comment#9")); + data.add(new Tuple3<>(16, 6L, "Comment#10")); + data.add(new Tuple3<>(17, 6L, "Comment#11")); + data.add(new Tuple3<>(18, 6L, "Comment#12")); + data.add(new Tuple3<>(19, 6L, "Comment#13")); + data.add(new Tuple3<>(20, 6L, "Comment#14")); + data.add(new Tuple3<>(21, 6L, "Comment#15")); Collections.shuffle(data); @@ -86,10 +86,10 @@ public static DataSet> get3TupleDataSet(ExecutionE public static DataSet> getSmall3TupleDataSet(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple3(1, 1l, "Hi")); - data.add(new Tuple3(2, 2l, "Hello")); - data.add(new Tuple3(3, 2l, "Hello world")); + List> data = new ArrayList<>(); + data.add(new Tuple3<>(1, 1L, "Hi")); + data.add(new Tuple3<>(2, 2L, "Hello")); + data.add(new Tuple3<>(3, 2L, "Hello world")); Collections.shuffle(data); @@ -98,27 +98,26 @@ public static DataSet> getSmall3TupleDataSet(Execu public static DataSet> get5TupleDataSet(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple5(1, 1l, 0, "Hallo", 1l)); - data.add(new Tuple5(2, 2l, 1, "Hallo Welt", 2l)); - data.add(new Tuple5(2, 3l, 2, "Hallo Welt wie", 1l)); - data.add(new Tuple5(3, 4l, 3, "Hallo Welt wie gehts?", 2l)); - data.add(new Tuple5(3, 5l, 4, "ABC", 2l)); - data.add(new Tuple5(3, 6l, 5, "BCD", 3l)); - data.add(new Tuple5(4, 7l, 6, "CDE", 2l)); - data.add(new Tuple5(4, 8l, 7, "DEF", 1l)); - data.add(new Tuple5(4, 9l, 8, "EFG", 1l)); - data.add(new Tuple5(4, 10l, 9, "FGH", 2l)); - data.add(new Tuple5(5, 11l, 10, "GHI", 1l)); - data.add(new Tuple5(5, 12l, 11, "HIJ", 3l)); - data.add(new Tuple5(5, 13l, 12, "IJK", 3l)); - data.add(new Tuple5(5, 14l, 13, "JKL", 2l)); - data.add(new Tuple5(5, 15l, 14, "KLM", 2l)); + List> data = new ArrayList<>(); + data.add(new Tuple5<>(1, 1L, 0, "Hallo", 1L)); + data.add(new Tuple5<>(2, 2L, 1, "Hallo Welt", 2L)); + data.add(new Tuple5<>(2, 3L, 2, "Hallo Welt wie", 1L)); + data.add(new Tuple5<>(3, 4L, 3, "Hallo Welt wie gehts?", 2L)); + data.add(new Tuple5<>(3, 5L, 4, "ABC", 2L)); + data.add(new Tuple5<>(3, 6L, 5, "BCD", 3L)); + data.add(new Tuple5<>(4, 7L, 6, "CDE", 2L)); + data.add(new Tuple5<>(4, 8L, 7, "DEF", 1L)); + data.add(new Tuple5<>(4, 9L, 8, "EFG", 1L)); + data.add(new Tuple5<>(4, 10L, 9, "FGH", 2L)); + data.add(new Tuple5<>(5, 11L, 10, "GHI", 1L)); + data.add(new Tuple5<>(5, 12L, 11, "HIJ", 3L)); + data.add(new Tuple5<>(5, 13L, 12, "IJK", 3L)); + data.add(new Tuple5<>(5, 14L, 13, "JKL", 2L)); + data.add(new Tuple5<>(5, 15L, 14, "KLM", 2L)); Collections.shuffle(data); - TupleTypeInfo> type = new - TupleTypeInfo>( + TupleTypeInfo> type = new TupleTypeInfo<>( BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, @@ -131,15 +130,14 @@ public static DataSet> get5TupleDat public static DataSet> getSmall5TupleDataSet(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple5(1, 1l, 0, "Hallo", 1l)); - data.add(new Tuple5(2, 2l, 1, "Hallo Welt", 2l)); - data.add(new Tuple5(2, 3l, 2, "Hallo Welt wie", 1l)); + List> data = new ArrayList<>(); + data.add(new Tuple5<>(1, 1L, 0, "Hallo", 1L)); + data.add(new Tuple5<>(2, 2L, 1, "Hallo Welt", 2L)); + data.add(new Tuple5<>(2, 3L, 2, "Hallo Welt wie", 1L)); Collections.shuffle(data); - TupleTypeInfo> type = new - TupleTypeInfo>( + TupleTypeInfo> type = new TupleTypeInfo<>( BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, @@ -152,13 +150,12 @@ public static DataSet> getSmall5Tup public static DataSet, String>> getSmallNestedTupleDataSet(ExecutionEnvironment env) { - List, String>> data = new ArrayList, String>>(); - data.add(new Tuple2, String>(new Tuple2(1, 1), "one")); - data.add(new Tuple2, String>(new Tuple2(2, 2), "two")); - data.add(new Tuple2, String>(new Tuple2(3, 3), "three")); + List, String>> data = new ArrayList<>(); + data.add(new Tuple2<>(new Tuple2<>(1, 1), "one")); + data.add(new Tuple2<>(new Tuple2<>(2, 2), "two")); + data.add(new Tuple2<>(new Tuple2<>(3, 3), "three")); - TupleTypeInfo, String>> type = new - TupleTypeInfo, String>>( + TupleTypeInfo, String>> type = new TupleTypeInfo<>( new TupleTypeInfo>(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO), BasicTypeInfo.STRING_TYPE_INFO ); @@ -168,17 +165,16 @@ public static DataSet, String>> getSmallNestedTu public static DataSet, String>> getGroupSortedNestedTupleDataSet(ExecutionEnvironment env) { - List, String>> data = new ArrayList, String>>(); - data.add(new Tuple2, String>(new Tuple2(1, 3), "a")); - data.add(new Tuple2, String>(new Tuple2(1, 2), "a")); - data.add(new Tuple2, String>(new Tuple2(2, 1), "a")); - data.add(new Tuple2, String>(new Tuple2(2, 2), "b")); - data.add(new Tuple2, String>(new Tuple2(3, 3), "c")); - data.add(new Tuple2, String>(new Tuple2(3, 6), "c")); - data.add(new Tuple2, String>(new Tuple2(4, 9), "c")); - - TupleTypeInfo, String>> type = new - TupleTypeInfo, String>>( + List, String>> data = new ArrayList<>(); + data.add(new Tuple2<>(new Tuple2<>(1, 3), "a")); + data.add(new Tuple2<>(new Tuple2<>(1, 2), "a")); + data.add(new Tuple2<>(new Tuple2<>(2, 1), "a")); + data.add(new Tuple2<>(new Tuple2<>(2, 2), "b")); + data.add(new Tuple2<>(new Tuple2<>(3, 3), "c")); + data.add(new Tuple2<>(new Tuple2<>(3, 6), "c")); + data.add(new Tuple2<>(new Tuple2<>(4, 9), "c")); + + TupleTypeInfo, String>> type = new TupleTypeInfo<>( new TupleTypeInfo>(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO), BasicTypeInfo.STRING_TYPE_INFO ); @@ -188,17 +184,16 @@ public static DataSet, String>> getGroupSortedNe public static DataSet, String, Integer>> getGroupSortedNestedTupleDataSet2(ExecutionEnvironment env) { - List, String, Integer>> data = new ArrayList, String, Integer>>(); - data.add(new Tuple3, String, Integer>(new Tuple2(1, 3), "a", 2)); - data.add(new Tuple3, String, Integer>(new Tuple2(1, 2), "a", 1)); - data.add(new Tuple3, String, Integer>(new Tuple2(2, 1), "a", 3)); - data.add(new Tuple3, String, Integer>(new Tuple2(2, 2), "b", 4)); - data.add(new Tuple3, String, Integer>(new Tuple2(3, 3), "c", 5)); - data.add(new Tuple3, String, Integer>(new Tuple2(3, 6), "c", 6)); - data.add(new Tuple3, String, Integer>(new Tuple2(4, 9), "c", 7)); - - TupleTypeInfo, String, Integer>> type = new - TupleTypeInfo, String, Integer>>( + List, String, Integer>> data = new ArrayList<>(); + data.add(new Tuple3<>(new Tuple2<>(1, 3), "a", 2)); + data.add(new Tuple3<>(new Tuple2<>(1, 2), "a", 1)); + data.add(new Tuple3<>(new Tuple2<>(2, 1), "a", 3)); + data.add(new Tuple3<>(new Tuple2<>(2, 2), "b", 4)); + data.add(new Tuple3<>(new Tuple2<>(3, 3), "c", 5)); + data.add(new Tuple3<>(new Tuple2<>(3, 6), "c", 6)); + data.add(new Tuple3<>(new Tuple2<>(4, 9), "c", 7)); + + TupleTypeInfo, String, Integer>> type = new TupleTypeInfo<>( new TupleTypeInfo>(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO), BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO @@ -208,15 +203,15 @@ public static DataSet, String, Integer>> getGrou } public static DataSet> getTuple2WithByteArrayDataSet(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple2(new byte[]{0, 4}, 1)); - data.add(new Tuple2(new byte[]{2, 0}, 1)); - data.add(new Tuple2(new byte[]{2, 0, 4}, 4)); - data.add(new Tuple2(new byte[]{2, 1}, 3)); - data.add(new Tuple2(new byte[]{0}, 0)); - data.add(new Tuple2(new byte[]{2, 0}, 1)); + List> data = new ArrayList<>(); + data.add(new Tuple2<>(new byte[]{0, 4}, 1)); + data.add(new Tuple2<>(new byte[]{2, 0}, 1)); + data.add(new Tuple2<>(new byte[]{2, 0, 4}, 4)); + data.add(new Tuple2<>(new byte[]{2, 1}, 3)); + data.add(new Tuple2<>(new byte[]{0}, 0)); + data.add(new Tuple2<>(new byte[]{2, 0}, 1)); - TupleTypeInfo> type = new TupleTypeInfo>( + TupleTypeInfo> type = new TupleTypeInfo<>( PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO ); @@ -226,7 +221,7 @@ public static DataSet> getTuple2WithByteArrayDataSet(Exe public static DataSet getStringDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add("Hi"); data.add("Hello"); data.add("Hello world"); @@ -243,7 +238,7 @@ public static DataSet getStringDataSet(ExecutionEnvironment env) { public static DataSet getIntegerDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(1); data.add(2); data.add(2); @@ -267,28 +262,28 @@ public static DataSet getIntegerDataSet(ExecutionEnvironment env) { public static DataSet getCustomTypeDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); - data.add(new CustomType(1, 0l, "Hi")); - data.add(new CustomType(2, 1l, "Hello")); - data.add(new CustomType(2, 2l, "Hello world")); - data.add(new CustomType(3, 3l, "Hello world, how are you?")); - data.add(new CustomType(3, 4l, "I am fine.")); - data.add(new CustomType(3, 5l, "Luke Skywalker")); - data.add(new CustomType(4, 6l, "Comment#1")); - data.add(new CustomType(4, 7l, "Comment#2")); - data.add(new CustomType(4, 8l, "Comment#3")); - data.add(new CustomType(4, 9l, "Comment#4")); - data.add(new CustomType(5, 10l, "Comment#5")); - data.add(new CustomType(5, 11l, "Comment#6")); - data.add(new CustomType(5, 12l, "Comment#7")); - data.add(new CustomType(5, 13l, "Comment#8")); - data.add(new CustomType(5, 14l, "Comment#9")); - data.add(new CustomType(6, 15l, "Comment#10")); - data.add(new CustomType(6, 16l, "Comment#11")); - data.add(new CustomType(6, 17l, "Comment#12")); - data.add(new CustomType(6, 18l, "Comment#13")); - data.add(new CustomType(6, 19l, "Comment#14")); - data.add(new CustomType(6, 20l, "Comment#15")); + List data = new ArrayList<>(); + data.add(new CustomType(1, 0L, "Hi")); + data.add(new CustomType(2, 1L, "Hello")); + data.add(new CustomType(2, 2L, "Hello world")); + data.add(new CustomType(3, 3L, "Hello world, how are you?")); + data.add(new CustomType(3, 4L, "I am fine.")); + data.add(new CustomType(3, 5L, "Luke Skywalker")); + data.add(new CustomType(4, 6L, "Comment#1")); + data.add(new CustomType(4, 7L, "Comment#2")); + data.add(new CustomType(4, 8L, "Comment#3")); + data.add(new CustomType(4, 9L, "Comment#4")); + data.add(new CustomType(5, 10L, "Comment#5")); + data.add(new CustomType(5, 11L, "Comment#6")); + data.add(new CustomType(5, 12L, "Comment#7")); + data.add(new CustomType(5, 13L, "Comment#8")); + data.add(new CustomType(5, 14L, "Comment#9")); + data.add(new CustomType(6, 15L, "Comment#10")); + data.add(new CustomType(6, 16L, "Comment#11")); + data.add(new CustomType(6, 17L, "Comment#12")); + data.add(new CustomType(6, 18L, "Comment#13")); + data.add(new CustomType(6, 19L, "Comment#14")); + data.add(new CustomType(6, 20L, "Comment#15")); Collections.shuffle(data); @@ -298,10 +293,10 @@ public static DataSet getCustomTypeDataSet(ExecutionEnvironment env) public static DataSet getSmallCustomTypeDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); - data.add(new CustomType(1, 0l, "Hi")); - data.add(new CustomType(2, 1l, "Hello")); - data.add(new CustomType(2, 2l, "Hello world")); + List data = new ArrayList<>(); + data.add(new CustomType(1, 0L, "Hi")); + data.add(new CustomType(2, 1L, "Hello")); + data.add(new CustomType(2, 2L, "Hello world")); Collections.shuffle(data); @@ -346,30 +341,24 @@ public int compare(CustomType o1, CustomType o2) { } public static DataSet> getSmallTuplebasedDataSet(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple7(1, "First", 10, 100, 1000L, "One", 10000L)); - data.add(new Tuple7(2, "Second", 20, 200, 2000L, "Two", 20000L)); - data.add(new Tuple7(3, "Third", 30, 300, 3000L, "Three", 30000L)); + List> data = new ArrayList<>(); + data.add(new Tuple7<>(1, "First", 10, 100, 1000L, "One", 10000L)); + data.add(new Tuple7<>(2, "Second", 20, 200, 2000L, "Two", 20000L)); + data.add(new Tuple7<>(3, "Third", 30, 300, 3000L, "Three", 30000L)); return env.fromCollection(data); } public static DataSet> getSmallTuplebasedDataSetMatchingPojo(ExecutionEnvironment env) { - List> data = - new ArrayList>(); - data.add(new Tuple7 - (10000L, 10, 100, 1000L, "One", 1, "First")); - - data.add(new Tuple7 - (20000L, 20, 200, 2000L, "Two", 2, "Second")); - - data.add(new Tuple7 - (30000L, 30, 300, 3000L, "Three", 3, "Third")); + List> data = new ArrayList<>(); + data.add(new Tuple7<>(10000L, 10, 100, 1000L, "One", 1, "First")); + data.add(new Tuple7<>(20000L, 20, 200, 2000L, "Two", 2, "Second")); + data.add(new Tuple7<>(30000L, 30, 300, 3000L, "Three", 3, "Third")); return env.fromCollection(data); } public static DataSet getSmallPojoDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new POJO(1 /*number*/, "First" /*str*/, 10 /*f0*/, 100/*f1.myInt*/, 1000L/*f1.myLong*/, "One" /*f1.myString*/, 10000L /*nestedPojo.longNumber*/)); data.add(new POJO(2, "Second", 20, 200, 2000L, "Two", 20000L)); data.add(new POJO(3, "Third", 30, 300, 3000L, "Three", 30000L)); @@ -377,7 +366,7 @@ public static DataSet getSmallPojoDataSet(ExecutionEnvironment env) { } public static DataSet getDuplicatePojoDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L)); // 5x data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L)); data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10000L)); @@ -390,7 +379,7 @@ public static DataSet getDuplicatePojoDataSet(ExecutionEnvironment env) { } public static DataSet getMixedPojoDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new POJO(1, "First", 10, 100, 1000L, "One", 10100L)); // 5x data.add(new POJO(2, "First_", 10, 105, 1000L, "One", 10200L)); data.add(new POJO(3, "First", 11, 102, 3000L, "One", 10200L)); @@ -414,7 +403,7 @@ public POJO(int i0, String s0, long l1) { this.number = i0; this.str = s0; - this.nestedTupleWithCustom = new Tuple2(i1, new CustomType(i2, l0, s1)); + this.nestedTupleWithCustom = new Tuple2<>(i1, new CustomType(i2, l0, s1)); this.nestedPojo = new NestedPojo(); this.nestedPojo.longNumber = l1; } @@ -437,7 +426,7 @@ public NestedPojo() { } public static DataSet getCrazyNestedDataSet(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new CrazyNested("aa")); data.add(new CrazyNested("bb")); data.add(new CrazyNested("bb")); @@ -506,7 +495,7 @@ public FromTupleWithCTor(int special, long tupleField) { } public static DataSet getPojoExtendingFromTuple(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new FromTupleWithCTor(1, 10L)); // 3x data.add(new FromTupleWithCTor(1, 10L)); data.add(new FromTupleWithCTor(1, 10L)); @@ -527,12 +516,12 @@ public PojoContainingTupleAndWritable() { public PojoContainingTupleAndWritable(int i, long l1, long l2) { hadoopFan = new IntWritable(i); someInt = i; - theTuple = new Tuple2(l1, l2); + theTuple = new Tuple2<>(l1, l2); } } public static DataSet getPojoContainingTupleAndWritable(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new PojoContainingTupleAndWritable(1, 10L, 100L)); // 1x data.add(new PojoContainingTupleAndWritable(2, 20L, 200L)); // 5x data.add(new PojoContainingTupleAndWritable(2, 20L, 200L)); @@ -545,7 +534,7 @@ public static DataSet getPojoContainingTupleAndW public static DataSet getGroupSortedPojoContainingTupleAndWritable(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new PojoContainingTupleAndWritable(1, 10L, 100L)); // 1x data.add(new PojoContainingTupleAndWritable(2, 20L, 200L)); // 5x data.add(new PojoContainingTupleAndWritable(2, 20L, 201L)); @@ -556,12 +545,12 @@ public static DataSet getGroupSortedPojoContaini } public static DataSet> getTupleContainingPojos(ExecutionEnvironment env) { - List> data = new ArrayList>(); - data.add(new Tuple3(1, new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // 3x - data.add(new Tuple3(1, new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); - data.add(new Tuple3(1, new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); + List> data = new ArrayList<>(); + data.add(new Tuple3<>(1, new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // 3x + data.add(new Tuple3<>(1, new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); + data.add(new Tuple3<>(1, new CrazyNested("one", "uno", 1L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // POJO is not initialized according to the first two fields. - data.add(new Tuple3(2, new CrazyNested("two", "duo", 2L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // 1x + data.add(new Tuple3<>(2, new CrazyNested("two", "duo", 2L), new POJO(1, "First", 10, 100, 1000L, "One", 10000L))); // 1x return env.fromCollection(data); } @@ -602,7 +591,7 @@ public PojoWithMultiplePojos(String a, String b, String a1, String b1, Integer i } public static DataSet getPojoWithMultiplePojos(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); data.add(new PojoWithMultiplePojos("a", "aa", "b", "bb", 1)); data.add(new PojoWithMultiplePojos("b", "bb", "c", "cc", 2)); data.add(new PojoWithMultiplePojos("b", "bb", "c", "cc", 2)); @@ -613,7 +602,7 @@ public static DataSet getPojoWithMultiplePojos(ExecutionE } public enum Category { - CAT_A, CAT_B; + CAT_A, CAT_B } public static class PojoWithDateAndEnum { @@ -623,7 +612,7 @@ public static class PojoWithDateAndEnum { } public static DataSet getPojoWithDateAndEnum(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); PojoWithDateAndEnum one = new PojoWithDateAndEnum(); one.group = "a"; one.date = new Date(666); one.cat = Category.CAT_A; @@ -688,13 +677,13 @@ public String toString() { } public static DataSet getPojoWithCollection(ExecutionEnvironment env) { - List data = new ArrayList(); + List data = new ArrayList<>(); - List pojosList1 = new ArrayList(); + List pojosList1 = new ArrayList<>(); pojosList1.add(new Pojo1("a", "aa")); pojosList1.add(new Pojo1("b", "bb")); - List pojosList2 = new ArrayList(); + List pojosList2 = new ArrayList<>(); pojosList2.add(new Pojo1("a2", "aa2")); pojosList2.add(new Pojo1("b2", "bb2")); @@ -706,10 +695,10 @@ public static DataSet getPojoWithCollection(ExecutionEnviron pwc1.bigDecimalKeepItNull = null; // use calendar to make it stable across time zones - GregorianCalendar gcl1 = new GregorianCalendar(2033, 04, 18); + GregorianCalendar gcl1 = new GregorianCalendar(2033, 4, 18); pwc1.sqlDate = new java.sql.Date(gcl1.getTimeInMillis()); - pwc1.mixed = new ArrayList(); - Map map = new HashMap(); + pwc1.mixed = new ArrayList<>(); + Map map = new HashMap<>(); map.put("someKey", 1); // map.put("anotherKey", 2); map.put("third", 3); pwc1.mixed.add(map); pwc1.mixed.add(new File("/this/is/wrong"));