You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2014/10/03 18:25:02 UTC
[06/23] git commit: [FLINK-1110] Framework for collection-based
execution
[FLINK-1110] Framework for collection-based execution
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/d10d59eb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/d10d59eb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/d10d59eb
Branch: refs/heads/master
Commit: d10d59eb0ddcb5b3712f0646b0efe91cbb8b991f
Parents: 8e21204
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Sep 22 15:12:58 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 3 16:22:33 2014 +0200
----------------------------------------------------------------------
.../org/apache/flink/compiler/PactCompiler.java | 4 +-
.../compiler/contextcheck/ContextChecker.java | 2 +-
.../apache/flink/compiler/dag/DataSinkNode.java | 2 +-
.../flink/compiler/dag/DataSourceNode.java | 2 +-
.../compiler/postpass/JavaApiPostPass.java | 2 +-
.../compiler/postpass/RecordModelPostPass.java | 2 +-
.../flink/compiler/util/NoOpBinaryUdfOp.java | 8 +
.../flink/compiler/util/NoOpUnaryUdfOp.java | 8 +
.../CachedMatchStrategyCompilerTest.java | 2 +-
.../apache/flink/compiler/CompilerTestBase.java | 2 +-
.../compiler/FeedbackPropertiesMatchTest.java | 2 +-
.../apache/flink/compiler/plan/ChannelTest.java | 2 +-
.../java/org/apache/flink/api/common/Plan.java | 2 +-
.../common/functions/util/ListCollector.java | 40 +++
.../functions/util/RuntimeUDFContext.java | 171 ++++++++++
.../flink/api/common/io/FileInputFormat.java | 2 +-
.../common/operators/CollectionExecutor.java | 178 +++++++++++
.../api/common/operators/DualInputOperator.java | 6 +-
.../common/operators/GenericDataSinkBase.java | 310 +++++++++++++++++++
.../common/operators/GenericDataSourceBase.java | 208 +++++++++++++
.../flink/api/common/operators/Operator.java | 4 +-
.../common/operators/SingleInputOperator.java | 5 +
.../flink/api/common/operators/Union.java | 12 +
.../common/operators/base/FileDataSinkBase.java | 1 +
.../operators/base/FileDataSourceBase.java | 1 +
.../operators/base/FilterOperatorBase.java | 26 +-
.../operators/base/GenericDataSinkBase.java | 300 ------------------
.../operators/base/GenericDataSourceBase.java | 178 -----------
.../common/operators/base/MapOperatorBase.java | 24 ++
.../api/common/operators/util/OperatorUtil.java | 4 +-
.../common/operators/util/OperatorUtilTest.java | 4 +-
.../CollectionExecutionExample.java | 48 +++
.../flink/api/java/CollectionEnvironment.java | 43 +++
.../flink/api/java/operators/DataSink.java | 2 +-
.../flink/api/java/operators/DataSource.java | 2 +-
.../api/java/operators/OperatorTranslation.java | 2 +-
.../java/operators/translation/JavaPlan.java | 2 +-
.../record/operators/CollectionDataSource.java | 2 +-
.../java/record/operators/GenericDataSink.java | 2 +-
.../record/operators/GenericDataSource.java | 2 +-
.../flink/api/java/MultipleInvokationsTest.java | 3 +-
.../SemanticPropertiesProjectionTest.java | 2 +-
.../SemanticPropertiesTranslationTest.java | 3 +-
.../translation/AggregateTranslationTest.java | 5 +-
.../DeltaIterationTranslationTest.java | 2 +-
.../translation/ReduceTranslationTests.java | 4 +-
.../task/AbstractIterativePactTask.java | 2 +-
.../runtime/operators/RegularPactTask.java | 2 +-
.../operators/chaining/ChainedDriver.java | 2 +-
.../operators/udf/RuntimeUDFContext.java | 170 ----------
.../test/compiler/util/CompilerTestBase.java | 2 +-
51 files changed, 1126 insertions(+), 690 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java b/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
index 0ff0aee..6492c5c 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/PactCompiler.java
@@ -32,6 +32,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.Union;
import org.apache.flink.api.common.operators.base.BulkIterationBase;
@@ -40,8 +42,6 @@ import org.apache.flink.api.common.operators.base.CrossOperatorBase;
import org.apache.flink.api.common.operators.base.DeltaIterationBase;
import org.apache.flink.api.common.operators.base.FilterOperatorBase;
import org.apache.flink.api.common.operators.base.FlatMapOperatorBase;
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase;
import org.apache.flink.api.common.operators.base.JoinOperatorBase;
import org.apache.flink.api.common.operators.base.MapOperatorBase;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/contextcheck/ContextChecker.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/contextcheck/ContextChecker.java b/flink-compiler/src/main/java/org/apache/flink/compiler/contextcheck/ContextChecker.java
index b051cf7..baeaadd 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/contextcheck/ContextChecker.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/contextcheck/ContextChecker.java
@@ -25,12 +25,12 @@ import java.util.Set;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.operators.DualInputOperator;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.SingleInputOperator;
import org.apache.flink.api.common.operators.base.BulkIterationBase;
import org.apache.flink.api.common.operators.base.FileDataSinkBase;
import org.apache.flink.api.common.operators.base.FileDataSourceBase;
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase;
import org.apache.flink.core.fs.Path;
import org.apache.flink.util.Visitor;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java
index 098e1e0..28a4d7c 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java
@@ -25,9 +25,9 @@ import java.util.List;
import java.util.Map;
import org.apache.flink.api.common.distributions.DataDistribution;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.Ordering;
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase;
import org.apache.flink.compiler.CompilerException;
import org.apache.flink.compiler.DataStatistics;
import org.apache.flink.compiler.costs.CostEstimator;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java
index 3de4e03..39c09a1 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java
@@ -28,8 +28,8 @@ import org.apache.flink.api.common.io.FileInputFormat;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.io.NonParallelInput;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.compiler.DataStatistics;
import org.apache.flink.compiler.PactCompiler;
import org.apache.flink.compiler.costs.CostEstimator;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/JavaApiPostPass.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/JavaApiPostPass.java b/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/JavaApiPostPass.java
index ea3cc87..4756bc9 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/JavaApiPostPass.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/JavaApiPostPass.java
@@ -23,11 +23,11 @@ import java.util.HashSet;
import java.util.Set;
import org.apache.flink.api.common.operators.DualInputOperator;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.SingleInputOperator;
import org.apache.flink.api.common.operators.base.BulkIterationBase;
import org.apache.flink.api.common.operators.base.DeltaIterationBase;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase;
import org.apache.flink.api.common.operators.util.FieldList;
import org.apache.flink.api.common.typeinfo.AtomicType;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/RecordModelPostPass.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/RecordModelPostPass.java b/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/RecordModelPostPass.java
index ef13ca7..f25d141 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/RecordModelPostPass.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/postpass/RecordModelPostPass.java
@@ -19,11 +19,11 @@
package org.apache.flink.compiler.postpass;
import org.apache.flink.api.common.operators.DualInputOperator;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
import org.apache.flink.api.common.operators.Ordering;
import org.apache.flink.api.common.operators.RecordOperator;
import org.apache.flink.api.common.operators.SingleInputOperator;
import org.apache.flink.api.common.operators.base.CoGroupOperatorBase;
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase;
import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase;
import org.apache.flink.api.common.operators.util.FieldList;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpBinaryUdfOp.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpBinaryUdfOp.java b/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpBinaryUdfOp.java
index f5558ef..1abaeac 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpBinaryUdfOp.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpBinaryUdfOp.java
@@ -18,6 +18,9 @@
package org.apache.flink.compiler.util;
+import java.util.List;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.NoOpFunction;
import org.apache.flink.api.common.operators.BinaryOperatorInformation;
import org.apache.flink.api.common.operators.DualInputOperator;
@@ -38,5 +41,10 @@ public class NoOpBinaryUdfOp<OUT> extends DualInputOperator<OUT, OUT, OUT, NoOpF
public Class<? extends Key<?>>[] getKeyClasses() {
return (Class<? extends Key<?>>[]) new Class[0];
}
+
+ @Override
+ protected List<OUT> executeOnCollections(List<OUT> inputData1, List<OUT> inputData2, RuntimeContext runtimeContext) {
+ throw new UnsupportedOperationException();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpUnaryUdfOp.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpUnaryUdfOp.java b/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpUnaryUdfOp.java
index bafa249..8f88168 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpUnaryUdfOp.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/util/NoOpUnaryUdfOp.java
@@ -18,6 +18,9 @@
package org.apache.flink.compiler.util;
+import java.util.List;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.NoOpFunction;
import org.apache.flink.api.common.operators.RecordOperator;
import org.apache.flink.api.common.operators.SingleInputOperator;
@@ -49,5 +52,10 @@ public class NoOpUnaryUdfOp<OUT> extends SingleInputOperator<OUT, OUT, NoOpFunct
TypeInformation<OUT> previousOut = input.getOperatorInfo().getOutputType();
return new UnaryOperatorInformation<OUT, OUT>(previousOut, previousOut);
}
+
+ @Override
+ protected List<OUT> executeOnCollections(List<OUT> inputData, RuntimeContext runtimeContext) {
+ return inputData;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/test/java/org/apache/flink/compiler/CachedMatchStrategyCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/compiler/CachedMatchStrategyCompilerTest.java b/flink-compiler/src/test/java/org/apache/flink/compiler/CachedMatchStrategyCompilerTest.java
index d615c37..1103099 100644
--- a/flink-compiler/src/test/java/org/apache/flink/compiler/CachedMatchStrategyCompilerTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/compiler/CachedMatchStrategyCompilerTest.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.fail;
import org.junit.Test;
import org.apache.flink.api.common.Plan;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.IterativeDataSet;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/test/java/org/apache/flink/compiler/CompilerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/compiler/CompilerTestBase.java b/flink-compiler/src/test/java/org/apache/flink/compiler/CompilerTestBase.java
index 46c1bc0..f7d2fd6 100644
--- a/flink-compiler/src/test/java/org/apache/flink/compiler/CompilerTestBase.java
+++ b/flink-compiler/src/test/java/org/apache/flink/compiler/CompilerTestBase.java
@@ -29,8 +29,8 @@ import java.util.Set;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.io.FileInputFormat.FileBaseStatistics;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.api.java.record.operators.BulkIteration;
import org.apache.flink.api.java.record.operators.DeltaIteration;
import org.apache.flink.compiler.costs.DefaultCostEstimator;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/test/java/org/apache/flink/compiler/FeedbackPropertiesMatchTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/compiler/FeedbackPropertiesMatchTest.java b/flink-compiler/src/test/java/org/apache/flink/compiler/FeedbackPropertiesMatchTest.java
index b854cd8..5d45159 100644
--- a/flink-compiler/src/test/java/org/apache/flink/compiler/FeedbackPropertiesMatchTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/compiler/FeedbackPropertiesMatchTest.java
@@ -25,11 +25,11 @@ import static org.junit.Assert.*;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.operators.BinaryOperatorInformation;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.OperatorInformation;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.common.operators.Ordering;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.api.common.operators.base.JoinOperatorBase;
import org.apache.flink.api.common.operators.base.MapOperatorBase;
import org.apache.flink.api.common.operators.util.FieldList;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-compiler/src/test/java/org/apache/flink/compiler/plan/ChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/test/java/org/apache/flink/compiler/plan/ChannelTest.java b/flink-compiler/src/test/java/org/apache/flink/compiler/plan/ChannelTest.java
index 0746efe..f585c06 100644
--- a/flink-compiler/src/test/java/org/apache/flink/compiler/plan/ChannelTest.java
+++ b/flink-compiler/src/test/java/org/apache/flink/compiler/plan/ChannelTest.java
@@ -19,8 +19,8 @@
package org.apache.flink.compiler.plan;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.OperatorInformation;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.compiler.dag.DataSourceNode;
import org.apache.flink.compiler.plan.Channel;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/Plan.java b/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
index 71e8cb5..f7e93b4 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
@@ -36,8 +36,8 @@ import java.net.URI;
import java.net.URISyntaxException;
import org.apache.flink.api.common.cache.DistributedCache.DistributedCacheEntry;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.util.Visitable;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/functions/util/ListCollector.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/ListCollector.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/ListCollector.java
new file mode 100644
index 0000000..f7f7865
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/ListCollector.java
@@ -0,0 +1,40 @@
+/**
+ * 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.api.common.functions.util;
+
+import java.util.List;
+
+import org.apache.flink.util.Collector;
+
+public class ListCollector<T> implements Collector<T> {
+
+ private final List<T> list;
+
+ public ListCollector(List<T> list) {
+ this.list = list;
+ }
+
+ @Override
+ public void collect(T record) {
+ list.add(record);
+ }
+
+ @Override
+ public void close() {}
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
new file mode 100644
index 0000000..c3d9076
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
@@ -0,0 +1,171 @@
+/*
+ * 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.api.common.functions.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.FutureTask;
+
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+import org.apache.flink.api.common.accumulators.DoubleCounter;
+import org.apache.flink.api.common.accumulators.Histogram;
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.core.fs.Path;
+
+/**
+ * Implementation of the {@link RuntimeContext}, created by runtime UDF operators.
+ */
+public class RuntimeUDFContext implements RuntimeContext {
+
+ private final String name;
+
+ private final int numParallelSubtasks;
+
+ private final int subtaskIndex;
+
+ private final ClassLoader userCodeClassLoader;
+
+ private final DistributedCache distributedCache = new DistributedCache();
+
+ private final HashMap<String, Accumulator<?, ?>> accumulators = new HashMap<String, Accumulator<?, ?>>();
+
+ private final HashMap<String, List<?>> broadcastVars = new HashMap<String, List<?>>();
+
+ public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader) {
+ this.name = name;
+ this.numParallelSubtasks = numParallelSubtasks;
+ this.subtaskIndex = subtaskIndex;
+ this.userCodeClassLoader = userCodeClassLoader;
+ }
+
+ public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, Map<String, FutureTask<Path>> cpTasks) {
+ this.name = name;
+ this.numParallelSubtasks = numParallelSubtasks;
+ this.subtaskIndex = subtaskIndex;
+ this.userCodeClassLoader = userCodeClassLoader;
+ this.distributedCache.setCopyTasks(cpTasks);
+ }
+
+ @Override
+ public String getTaskName() {
+ return this.name;
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return this.numParallelSubtasks;
+ }
+
+ @Override
+ public int getIndexOfThisSubtask() {
+ return this.subtaskIndex;
+ }
+
+ @Override
+ public IntCounter getIntCounter(String name) {
+ return (IntCounter) getAccumulator(name, IntCounter.class);
+ }
+
+ @Override
+ public LongCounter getLongCounter(String name) {
+ return (LongCounter) getAccumulator(name, LongCounter.class);
+ }
+
+ @Override
+ public Histogram getHistogram(String name) {
+ return (Histogram) getAccumulator(name, Histogram.class);
+ }
+
+ @Override
+ public DoubleCounter getDoubleCounter(String name) {
+ return (DoubleCounter) getAccumulator(name, DoubleCounter.class);
+ }
+
+ @Override
+ public <V, A> void addAccumulator(String name, Accumulator<V, A> accumulator) {
+ if (accumulators.containsKey(name)) {
+ throw new UnsupportedOperationException("The counter '" + name
+ + "' already exists and cannot be added.");
+ }
+ accumulators.put(name, accumulator);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <V, A> Accumulator<V, A> getAccumulator(String name) {
+ return (Accumulator<V, A>) accumulators.get(name);
+ }
+
+ @SuppressWarnings("unchecked")
+ private <V, A> Accumulator<V, A> getAccumulator(String name,
+ Class<? extends Accumulator<V, A>> accumulatorClass) {
+
+ Accumulator<?, ?> accumulator = accumulators.get(name);
+
+ if (accumulator != null) {
+ AccumulatorHelper.compareAccumulatorTypes(name, accumulator.getClass(), accumulatorClass);
+ } else {
+ // Create new accumulator
+ try {
+ accumulator = accumulatorClass.newInstance();
+ } catch (InstantiationException e) {
+ e.printStackTrace();
+ } catch (IllegalAccessException e) {
+ e.printStackTrace();
+ }
+ accumulators.put(name, accumulator);
+ }
+ return (Accumulator<V, A>) accumulator;
+ }
+
+ @Override
+ public HashMap<String, Accumulator<?, ?>> getAllAccumulators() {
+ return this.accumulators;
+ }
+
+ public void setBroadcastVariable(String name, List<?> value) {
+ this.broadcastVars.put(name, value);
+ }
+
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <RT> List<RT> getBroadcastVariable(String name) {
+ if (!this.broadcastVars.containsKey(name)) {
+ throw new IllegalArgumentException("Trying to access an unbound broadcast variable '"
+ + name + "'.");
+ }
+ return (List<RT>) this.broadcastVars.get(name);
+ }
+
+ @Override
+ public DistributedCache getDistributedCache() {
+ return this.distributedCache;
+ }
+
+ @Override
+ public ClassLoader getUserCodeClassLoader() {
+ return this.userCodeClassLoader;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
index 25f1299..13a54f6 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
@@ -28,7 +28,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
new file mode 100644
index 0000000..6643703
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
@@ -0,0 +1,178 @@
+/**
+ * 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.api.common.operators;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.InvalidProgramException;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
+
+public class CollectionExecutor {
+
+ private final Map<Operator<?>, List<?>> intermediateResults;
+
+ private final Map<String, Accumulator<?, ?>> accumulators;
+
+
+ public CollectionExecutor() {
+ this.intermediateResults = new HashMap<Operator<?>, List<?>>();
+ this.accumulators = new HashMap<String, Accumulator<?,?>>();
+ }
+
+
+ public JobExecutionResult execute(Plan program) throws Exception {
+ long startTime = System.currentTimeMillis();
+
+ Collection<? extends GenericDataSinkBase<?>> sinks = program.getDataSinks();
+ for (Operator<?> sink : sinks) {
+ execute(sink);
+ }
+
+ long endTime = System.currentTimeMillis();
+ Map<String, Object> accumulatorResults = AccumulatorHelper.toResultMap(accumulators);
+ return new JobExecutionResult(endTime - startTime, accumulatorResults);
+ }
+
+// private void execute(Operator<?>... operators) throws Exception {
+// for (Operator<?> op : operators) {
+// execute(op);
+// }
+// }
+
+
+ private List<?> execute(Operator<?> operator) throws Exception {
+ List<?> result = this.intermediateResults.get(operator);
+
+ // if it has already been computed, use the cached variant
+ if (result != null) {
+ return result;
+ }
+
+ if (operator instanceof SingleInputOperator) {
+ return executeUnaryOperator((SingleInputOperator<?, ?, ?>) operator);
+ }
+ else if (operator instanceof DualInputOperator) {
+ return executeBinaryOperator((DualInputOperator<?, ?, ?, ?>) operator);
+ }
+ else if (operator instanceof GenericDataSourceBase) {
+ return executeDataSource((GenericDataSourceBase<?, ?>) operator);
+ }
+ else if (operator instanceof GenericDataSinkBase) {
+ executeDataSink((GenericDataSinkBase<?>) operator);
+ result = Collections.emptyList();
+ }
+
+ this.intermediateResults.put(operator, result);
+
+ return result;
+ }
+
+
+
+ private <IN> void executeDataSink(GenericDataSinkBase<?> sink) throws Exception {
+ Operator<?> inputOp = sink.getInput();
+ if (inputOp == null) {
+ throw new InvalidProgramException("The data sink " + sink.getName() + " has no input.");
+ }
+
+ @SuppressWarnings("unchecked")
+ List<IN> input = (List<IN>) execute(inputOp);
+
+ @SuppressWarnings("unchecked")
+ GenericDataSinkBase<IN> typedSink = (GenericDataSinkBase<IN>) sink;
+
+ typedSink.executeOnCollections(input);
+ }
+
+ private <OUT> List<OUT> executeDataSource(GenericDataSourceBase<?, ?> source) throws Exception {
+ @SuppressWarnings("unchecked")
+ GenericDataSourceBase<OUT, ?> typedSource = (GenericDataSourceBase<OUT, ?>) source;
+ return typedSource.executeOnCollections();
+ }
+
+ private <IN, OUT> List<OUT> executeUnaryOperator(SingleInputOperator<?, ?, ?> operator) throws Exception {
+ Operator<?> inputOp = operator.getInput();
+ if (inputOp == null) {
+ throw new InvalidProgramException("The unary operation " + operator.getName() + " has no input.");
+ }
+
+ @SuppressWarnings("unchecked")
+ List<IN> inputData = (List<IN>) execute(inputOp);
+
+ @SuppressWarnings("unchecked")
+ SingleInputOperator<IN, OUT, ?> typedOp = (SingleInputOperator<IN, OUT, ?>) operator;
+
+ RuntimeUDFContext ctx;
+ if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
+ ctx = new RuntimeUDFContext(operator.getName(), 1, 0);
+ } else {
+ ctx = null;
+ }
+
+ List<OUT> result = typedOp.executeOnCollections(inputData, ctx);
+
+ if (ctx != null) {
+ AccumulatorHelper.mergeInto(this.accumulators, ctx.getAllAccumulators());
+ }
+ return result;
+ }
+
+ private <IN1, IN2, OUT> List<OUT> executeBinaryOperator(DualInputOperator<?, ?, ?, ?> operator) throws Exception {
+ Operator<?> inputOp1 = operator.getFirstInput();
+ Operator<?> inputOp2 = operator.getSecondInput();
+
+ if (inputOp1 == null) {
+ throw new InvalidProgramException("The binary operation " + operator.getName() + " has no first input.");
+ }
+ if (inputOp2 == null) {
+ throw new InvalidProgramException("The binary operation " + operator.getName() + " has no second input.");
+ }
+
+ @SuppressWarnings("unchecked")
+ List<IN1> inputData1 = (List<IN1>) execute(inputOp1);
+ @SuppressWarnings("unchecked")
+ List<IN2> inputData2 = (List<IN2>) execute(inputOp2);
+
+ @SuppressWarnings("unchecked")
+ DualInputOperator<IN1, IN2, OUT, ?> typedOp = (DualInputOperator<IN1, IN2, OUT, ?>) operator;
+
+ RuntimeUDFContext ctx;
+ if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
+ ctx = new RuntimeUDFContext(operator.getName(), 1, 0);
+ } else {
+ ctx = null;
+ }
+
+ List<OUT> result = typedOp.executeOnCollections(inputData1, inputData2, ctx);
+
+ if (ctx != null) {
+ AccumulatorHelper.mergeInto(this.accumulators, ctx.getAllAccumulators());
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputOperator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputOperator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputOperator.java
index dc31785..2d28180 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputOperator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputOperator.java
@@ -22,6 +22,7 @@ package org.apache.flink.api.common.operators;
import java.util.List;
import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.util.Visitor;
@@ -270,7 +271,6 @@ public abstract class DualInputOperator<IN1, IN2, OUT, FT extends Function> exte
// --------------------------------------------------------------------------------------------
-
@Override
public void accept(Visitor<Operator<?>> visitor) {
boolean descend = visitor.preVisit(this);
@@ -283,4 +283,8 @@ public abstract class DualInputOperator<IN1, IN2, OUT, FT extends Function> exte
visitor.postVisit(this);
}
}
+
+ // --------------------------------------------------------------------------------------------
+
+ protected abstract List<OUT> executeOnCollections(List<IN1> inputData1, List<IN2> inputData2, RuntimeContext runtimeContext) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
new file mode 100644
index 0000000..ca55855
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
@@ -0,0 +1,310 @@
+/*
+ * 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.api.common.operators;
+
+import java.util.List;
+
+import org.apache.flink.api.common.distributions.DataDistribution;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
+import org.apache.flink.api.common.operators.util.UserCodeWrapper;
+import org.apache.flink.types.Nothing;
+import org.apache.flink.util.Visitor;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Operator for nodes that act as data sinks, storing the data they receive.
+ * The way the data is stored is handled by the {@link org.apache.flink.api.common.io.OutputFormat}.
+ */
+public class GenericDataSinkBase<IN> extends Operator<Nothing> {
+
+ protected final UserCodeWrapper<? extends OutputFormat<IN>> formatWrapper;
+
+ protected Operator<IN> input = null;
+
+ private Ordering localOrdering;
+
+ private Ordering partitionOrdering;
+
+ private DataDistribution distribution;
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Creates a GenericDataSink with the provided {@link org.apache.flink.api.common.io.OutputFormat} implementation
+ * and the given name.
+ *
+ * @param f The {@link org.apache.flink.api.common.io.OutputFormat} implementation used to sink the data.
+ * @param name The given name for the sink, used in plans, logs and progress messages.
+ */
+ public GenericDataSinkBase(OutputFormat<IN> f, UnaryOperatorInformation<IN, Nothing> operatorInfo, String name) {
+ super(operatorInfo, name);
+
+ Preconditions.checkNotNull(f, "The OutputFormat may not be null.");
+ this.formatWrapper = new UserCodeObjectWrapper<OutputFormat<IN>>(f);
+ }
+
+ /**
+ * Creates a GenericDataSink with the provided {@link org.apache.flink.api.common.io.OutputFormat} implementation
+ * and the given name.
+ *
+ * @param f The {@link org.apache.flink.api.common.io.OutputFormat} implementation used to sink the data.
+ * @param name The given name for the sink, used in plans, logs and progress messages.
+ */
+ public GenericDataSinkBase(UserCodeWrapper<? extends OutputFormat<IN>> f, UnaryOperatorInformation<IN, Nothing> operatorInfo, String name) {
+ super(operatorInfo, name);
+ Preconditions.checkNotNull(f, "The OutputFormat class may not be null.");
+ this.formatWrapper = f;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Returns this operator's input operator.
+ *
+ * @return This operator's input.
+ */
+ public Operator<IN> getInput() {
+ return this.input;
+ }
+
+ /**
+ * Sets the given operator as the input to this operator.
+ *
+ * @param input The operator to use as the input.
+ */
+ public void setInput(Operator<IN> input) {
+ Preconditions.checkNotNull(input, "The input may not be null.");
+ this.input = input;
+ }
+
+ /**
+ * Sets the input to the union of the given operators.
+ *
+ * @param inputs The operator(s) that form the input.
+ * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
+ */
+ @Deprecated
+ public void setInputs(Operator<IN>... inputs) {
+ Preconditions.checkNotNull(inputs, "The inputs may not be null.");
+ this.input = Operator.createUnionCascade(inputs);
+ }
+
+ /**
+ * Sets the input to the union of the given operators.
+ *
+ * @param inputs The operator(s) that form the input.
+ * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
+ */
+ @Deprecated
+ public void setInputs(List<Operator<IN>> inputs) {
+ Preconditions.checkNotNull(inputs, "The inputs may not be null.");
+ this.input = Operator.createUnionCascade(inputs);
+ }
+
+ /**
+ * Adds to the input the union of the given operators.
+ *
+ * @param inputs The operator(s) to be unioned with the input.
+ * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
+ */
+ @Deprecated
+ public void addInput(Operator<IN>... inputs) {
+ Preconditions.checkNotNull(inputs, "The input may not be null.");
+ this.input = Operator.createUnionCascade(this.input, inputs);
+ }
+
+ /**
+ * Adds to the input the union of the given operators.
+ *
+ * @param inputs The operator(s) to be unioned with the input.
+ * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
+ */
+ @SuppressWarnings("unchecked")
+ @Deprecated
+ public void addInputs(List<? extends Operator<IN>> inputs) {
+ Preconditions.checkNotNull(inputs, "The inputs may not be null.");
+ this.input = createUnionCascade(this.input, (Operator<IN>[]) inputs.toArray(new Operator[inputs.size()]));
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Sets the order in which the sink must write its data. For any value other then <tt>NONE</tt>,
+ * this will cause the system to perform a global sort, or try to reuse an order from a
+ * previous operation.
+ *
+ * @param globalOrder The order to write the data in.
+ */
+ public void setGlobalOrder(Ordering globalOrder) {
+ this.localOrdering = globalOrder;
+ setRangePartitioned(globalOrder);
+ }
+
+ /**
+ * Sets the order in which the sink must write its data. For any value other then <tt>NONE</tt>,
+ * this will cause the system to perform a global sort, or try to reuse an order from a
+ * previous operation.
+ *
+ * @param globalOrder The order to write the data in.
+ * @param distribution The distribution to use for the range partitioning.
+ */
+ public void setGlobalOrder(Ordering globalOrder, DataDistribution distribution) {
+ this.localOrdering = globalOrder;
+ setRangePartitioned(globalOrder, distribution);
+ }
+
+ /**
+ * Gets the order, in which the data sink writes its data locally. Local order means that
+ * with in each fragment of the file inside the distributed file system, the data is ordered,
+ * but not across file fragments.
+ *
+ * @return NONE, if the sink writes data in any order, or ASCENDING (resp. DESCENDING),
+ * if the sink writes it data with a local ascending (resp. descending) order.
+ */
+ public Ordering getLocalOrder() {
+ return this.localOrdering;
+ }
+
+ /**
+ * Sets the order in which the sink must write its data within each fragment in the distributed
+ * file system. For any value other then <tt>NONE</tt>, this will cause the system to perform a
+ * local sort, or try to reuse an order from a previous operation.
+ *
+ * @param localOrder The local order to write the data in.
+ */
+ public void setLocalOrder(Ordering localOrder) {
+ this.localOrdering = localOrder;
+ }
+
+ /**
+ * Gets the record ordering over which the sink partitions in ranges.
+ *
+ * @return The record ordering over which to partition in ranges.
+ */
+ public Ordering getPartitionOrdering() {
+ return this.partitionOrdering;
+ }
+
+ /**
+ * Sets the sink to partition the records into ranges over the given ordering.
+ *
+ * @param partitionOrdering The record ordering over which to partition in ranges.
+ */
+ public void setRangePartitioned(Ordering partitionOrdering) {
+ throw new UnsupportedOperationException(
+ "Range partitioning is currently only supported with a user supplied data distribution.");
+ }
+
+ /**
+ * Sets the sink to partition the records into ranges over the given ordering.
+ * The bucket boundaries are determined using the given data distribution.
+ *
+ * @param partitionOrdering The record ordering over which to partition in ranges.
+ * @param distribution The distribution to use for the range partitioning.
+ */
+ public void setRangePartitioned(Ordering partitionOrdering, DataDistribution distribution) {
+ if (partitionOrdering.getNumberOfFields() != distribution.getNumberOfFields()) {
+ throw new IllegalArgumentException("The number of keys in the distribution must match number of ordered fields.");
+ }
+
+ // TODO: check compatibility of distribution and ordering (number and order of keys, key types, etc.
+ // TODO: adapt partition ordering to data distribution (use prefix of ordering)
+
+ this.partitionOrdering = partitionOrdering;
+ this.distribution = distribution;
+ }
+
+ /**
+ * Gets the distribution to use for the range partitioning.
+ *
+ * @return The distribution to use for the range partitioning.
+ */
+ public DataDistribution getDataDistribution() {
+ return this.distribution;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Gets the class describing this sinks output format.
+ *
+ * @return The output format class.
+ */
+ public UserCodeWrapper<? extends OutputFormat<IN>> getFormatWrapper() {
+ return this.formatWrapper;
+ }
+
+ /**
+ * Gets the class describing the output format.
+ * <p>
+ * This method is basically identical to {@link #getFormatWrapper()}.
+ *
+ * @return The class describing the output format.
+ *
+ * @see org.apache.flink.api.common.operators.Operator#getUserCodeWrapper()
+ */
+ @Override
+ public UserCodeWrapper<? extends OutputFormat<IN>> getUserCodeWrapper() {
+ return this.formatWrapper;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Accepts the visitor and applies it this instance. This method applies the visitor in a depth-first traversal.
+ * The visitors pre-visit method is called and, if returning
+ * <tt>true</tt>, the visitor is recursively applied on the single input. After the recursion returned,
+ * the post-visit method is called.
+ *
+ * @param visitor The visitor.
+ *
+ * @see org.apache.flink.util.Visitable#accept(org.apache.flink.util.Visitor)
+ */
+ @Override
+ public void accept(Visitor<Operator<?>> visitor) {
+ boolean descend = visitor.preVisit(this);
+ if (descend) {
+ this.input.accept(visitor);
+ visitor.postVisit(this);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ protected void executeOnCollections(List<IN> inputData) throws Exception {
+ OutputFormat<IN> format = this.formatWrapper.getUserCodeObject();
+ format.configure(this.parameters);
+
+ format.open(0, 1);
+ for (IN element : inputData) {
+ format.writeRecord(element);
+ }
+ format.close();
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public String toString() {
+ return this.name;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
new file mode 100644
index 0000000..957f2e2
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
@@ -0,0 +1,208 @@
+/*
+ * 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.api.common.operators;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
+import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
+import org.apache.flink.api.common.operators.util.UserCodeWrapper;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.util.Visitor;
+
+/**
+ * Abstract superclass for data sources in a Pact plan.
+ *
+ * @param <OUT> The output type of the data source
+ * @param <T> The type of input format invoked by instances of this data source.
+ */
+public class GenericDataSourceBase<OUT, T extends InputFormat<OUT, ?>> extends Operator<OUT> {
+
+ private static final String DEFAULT_NAME = "<Unnamed Generic Data Source>";
+
+ protected final UserCodeWrapper<? extends T> formatWrapper;
+
+ protected String statisticsKey;
+
+ /**
+ * Creates a new instance for the given file using the given input format.
+ *
+ * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
+ * @param operatorInfo The type information for the operator.
+ * @param name The given name for the Pact, used in plans, logs and progress messages.
+ */
+ public GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name) {
+ super(operatorInfo, name);
+
+ if (format == null) {
+ throw new IllegalArgumentException("Input format may not be null.");
+ }
+
+ this.formatWrapper = new UserCodeObjectWrapper<T>(format);
+ }
+
+ /**
+ * Creates a new instance for the given file using the given input format, using the default name.
+ *
+ * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
+ * @param operatorInfo The type information for the operator.
+ */
+ public GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo) {
+ super(operatorInfo, DEFAULT_NAME);
+
+ if (format == null) {
+ throw new IllegalArgumentException("Input format may not be null.");
+ }
+
+ this.formatWrapper = new UserCodeObjectWrapper<T>(format);
+ }
+
+ /**
+ * Creates a new instance for the given file using the given input format.
+ *
+ * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
+ * @param operatorInfo The type information for the operator.
+ * @param name The given name for the Pact, used in plans, logs and progress messages.
+ */
+ public GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name) {
+ super(operatorInfo, name);
+
+ if (format == null) {
+ throw new IllegalArgumentException("Input format may not be null.");
+ }
+
+ this.formatWrapper = new UserCodeClassWrapper<T>(format);
+ }
+
+ /**
+ * Creates a new instance for the given file using the given input format, using the default name.
+ *
+ * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
+ * @param operatorInfo The type information for the operator.
+ */
+ public GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo) {
+ super(operatorInfo, DEFAULT_NAME);
+
+ if (format == null) {
+ throw new IllegalArgumentException("Input format may not be null.");
+ }
+
+ this.formatWrapper = new UserCodeClassWrapper<T>(format);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Gets the class describing the input format.
+ *
+ * @return The class describing the input format.
+ */
+ public UserCodeWrapper<? extends T> getFormatWrapper() {
+ return this.formatWrapper;
+ }
+
+ /**
+ * Gets the class describing the input format.
+ * <p>
+ * This method is basically identical to {@link #getFormatWrapper()}.
+ *
+ * @return The class describing the input format.
+ *
+ * @see org.apache.flink.api.common.operators.Operator#getUserCodeWrapper()
+ */
+ @Override
+ public UserCodeWrapper<? extends T> getUserCodeWrapper() {
+ return this.formatWrapper;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Gets the key under which statistics about this data source may be obtained from the
+ * statistics cache.
+ *
+ * @return The statistics cache key.
+ */
+ public String getStatisticsKey() {
+ return this.statisticsKey;
+ }
+
+ /**
+ * Sets the key under which statistics about this data source may be obtained from the
+ * statistics cache. Useful for testing purposes, when providing mock statistics.
+ *
+ * @param statisticsKey The key for the statistics object.
+ */
+ public void setStatisticsKey(String statisticsKey) {
+ this.statisticsKey = statisticsKey;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Accepts the visitor and applies it this instance. Since the data sources have no inputs, no recursive descend
+ * happens. The visitors pre-visit method is called and, if returning <tt>true</tt>, the post-visit method is called.
+ *
+ * @param visitor The visitor.
+ *
+ * @see org.apache.flink.util.Visitable#accept(org.apache.flink.util.Visitor)
+ */
+ @Override
+ public void accept(Visitor<Operator<?>> visitor) {
+ if (visitor.preVisit(this)) {
+ visitor.postVisit(this);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ protected List<OUT> executeOnCollections() throws Exception {
+ @SuppressWarnings("unchecked")
+ InputFormat<OUT, InputSplit> inputFormat = (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
+ inputFormat.configure(this.parameters);
+
+ List<OUT> result = new ArrayList<OUT>();
+
+ // splits
+ InputSplit[] splits = inputFormat.createInputSplits(1);
+ TypeSerializer<OUT> serializer = getOperatorInfo().getOutputType().createSerializer();
+
+ for (InputSplit split : splits) {
+ inputFormat.open(split);
+
+ while (!inputFormat.reachedEnd()) {
+ result.add(inputFormat.nextRecord(serializer.createInstance()));
+ }
+
+ inputFormat.close();
+ }
+
+ return result;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public String toString() {
+ return this.name;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
index 4e87347..765aa73 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
@@ -47,7 +47,7 @@ public abstract class Operator<OUT> implements Visitable<Operator<?>> {
*/
protected final OperatorInformation<OUT> operatorInfo;
- // --------------------------------------------------------------------------------------------
+ // --------------------------------------------------------------------------------------------
/**
* Creates a new contract with the given name. The parameters are empty by default and
@@ -192,6 +192,8 @@ public abstract class Operator<OUT> implements Visitable<Operator<?>> {
return null;
}
+ // --------------------------------------------------------------------------------------------
+
/**
* Takes a list of operators and creates a cascade of unions of this inputs, if needed.
* If not needed (there was only one operator in the list), then that operator is returned.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputOperator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputOperator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputOperator.java
index 6bb93a5..fa923e7 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputOperator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputOperator.java
@@ -22,6 +22,7 @@ package org.apache.flink.api.common.operators;
import java.util.List;
import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.util.Visitor;
@@ -205,4 +206,8 @@ public abstract class SingleInputOperator<IN, OUT, FT extends Function> extends
visitor.postVisit(this);
}
}
+
+ // --------------------------------------------------------------------------------------------
+
+ protected abstract List<OUT> executeOnCollections(List<IN> inputData, RuntimeContext runtimeContext) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/Union.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/Union.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/Union.java
index 58f4b73..f4b3c51 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/Union.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/Union.java
@@ -18,7 +18,11 @@
package org.apache.flink.api.common.operators;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
/**
@@ -42,4 +46,12 @@ public class Union<T> extends DualInputOperator<T, T, T, AbstractRichFunction> {
setFirstInput(input1);
setSecondInput(input2);
}
+
+ @Override
+ protected List<T> executeOnCollections(List<T> inputData1, List<T> inputData2, RuntimeContext runtimeContext) {
+ ArrayList<T> result = new ArrayList<T>(inputData1.size() + inputData2.size());
+ result.addAll(inputData1);
+ result.addAll(inputData2);
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSinkBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSinkBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSinkBase.java
index 558c81c..a36d44b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSinkBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSinkBase.java
@@ -20,6 +20,7 @@
package org.apache.flink.api.common.operators.base;
import org.apache.flink.api.common.io.FileOutputFormat;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.types.Nothing;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSourceBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSourceBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSourceBase.java
index 9b61f95..32ad380 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSourceBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FileDataSourceBase.java
@@ -20,6 +20,7 @@
package org.apache.flink.api.common.operators.base;
import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.OperatorInformation;
import com.google.common.base.Preconditions;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
index a2cdacb..39f8984 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
@@ -16,10 +16,15 @@
* limitations under the License.
*/
-
package org.apache.flink.api.common.operators.base;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.api.common.functions.util.ListCollector;
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;
@@ -43,4 +48,23 @@ public class FilterOperatorBase<T, FT extends FlatMapFunction<T, T>> extends Sin
public FilterOperatorBase(Class<? extends FT> udf, UnaryOperatorInformation<T, T> operatorInfo, String name) {
super(new UserCodeClassWrapper<FT>(udf), operatorInfo, name);
}
+
+ @Override
+ protected List<T> executeOnCollections(List<T> inputData, RuntimeContext ctx) throws Exception {
+ FlatMapFunction<T, T> function = this.userFunction.getUserCodeObject();
+
+ FunctionUtils.openFunction(function, this.parameters);
+ FunctionUtils.setFunctionRuntimeContext(function, ctx);
+
+ ArrayList<T> result = new ArrayList<T>(inputData.size());
+ ListCollector<T> collector = new ListCollector<T>(result);
+
+ for (T element : inputData) {
+ function.flatMap(element, collector);
+ }
+
+ FunctionUtils.closeFunction(function);
+
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSinkBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSinkBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSinkBase.java
deleted file mode 100644
index bda0751..0000000
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSinkBase.java
+++ /dev/null
@@ -1,300 +0,0 @@
-/*
- * 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.api.common.operators.base;
-
-import java.util.List;
-
-import org.apache.flink.api.common.distributions.DataDistribution;
-import org.apache.flink.api.common.io.OutputFormat;
-import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.Ordering;
-import org.apache.flink.api.common.operators.UnaryOperatorInformation;
-import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
-import org.apache.flink.api.common.operators.util.UserCodeWrapper;
-import org.apache.flink.types.Nothing;
-import org.apache.flink.util.Visitor;
-
-import com.google.common.base.Preconditions;
-
-/**
- * Operator for nodes that act as data sinks, storing the data they receive.
- * The way the data is stored is handled by the {@link org.apache.flink.api.common.io.OutputFormat}.
- */
-public class GenericDataSinkBase<IN> extends Operator<Nothing> {
-
- protected final UserCodeWrapper<? extends OutputFormat<IN>> formatWrapper;
-
- protected Operator<IN> input = null;
-
- private Ordering localOrdering;
-
- private Ordering partitionOrdering;
-
- private DataDistribution distribution;
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Creates a GenericDataSink with the provided {@link org.apache.flink.api.common.io.OutputFormat} implementation
- * and the given name.
- *
- * @param f The {@link org.apache.flink.api.common.io.OutputFormat} implementation used to sink the data.
- * @param name The given name for the sink, used in plans, logs and progress messages.
- */
- public GenericDataSinkBase(OutputFormat<IN> f, UnaryOperatorInformation<IN, Nothing> operatorInfo, String name) {
- super(operatorInfo, name);
-
- Preconditions.checkNotNull(f, "The OutputFormat may not be null.");
- this.formatWrapper = new UserCodeObjectWrapper<OutputFormat<IN>>(f);
- }
-
- /**
- * Creates a GenericDataSink with the provided {@link org.apache.flink.api.common.io.OutputFormat} implementation
- * and the given name.
- *
- * @param f The {@link org.apache.flink.api.common.io.OutputFormat} implementation used to sink the data.
- * @param name The given name for the sink, used in plans, logs and progress messages.
- */
- public GenericDataSinkBase(UserCodeWrapper<? extends OutputFormat<IN>> f, UnaryOperatorInformation<IN, Nothing> operatorInfo, String name) {
- super(operatorInfo, name);
- Preconditions.checkNotNull(f, "The OutputFormat class may not be null.");
- this.formatWrapper = f;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Returns this operator's input operator.
- *
- * @return This operator's input.
- */
- public Operator<IN> getInput() {
- return this.input;
- }
-
- /**
- * Sets the given operator as the input to this operator.
- *
- * @param input The operator to use as the input.
- */
- public void setInput(Operator<IN> input) {
- Preconditions.checkNotNull(input, "The input may not be null.");
- this.input = input;
- }
-
- /**
- * Sets the input to the union of the given operators.
- *
- * @param inputs The operator(s) that form the input.
- * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
- */
- @Deprecated
- public void setInputs(Operator<IN>... inputs) {
- Preconditions.checkNotNull(inputs, "The inputs may not be null.");
- this.input = Operator.createUnionCascade(inputs);
- }
-
- /**
- * Sets the input to the union of the given operators.
- *
- * @param inputs The operator(s) that form the input.
- * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
- */
- @Deprecated
- public void setInputs(List<Operator<IN>> inputs) {
- Preconditions.checkNotNull(inputs, "The inputs may not be null.");
- this.input = Operator.createUnionCascade(inputs);
- }
-
- /**
- * Adds to the input the union of the given operators.
- *
- * @param inputs The operator(s) to be unioned with the input.
- * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
- */
- @Deprecated
- public void addInput(Operator<IN>... inputs) {
- Preconditions.checkNotNull(inputs, "The input may not be null.");
- this.input = Operator.createUnionCascade(this.input, inputs);
- }
-
- /**
- * Adds to the input the union of the given operators.
- *
- * @param inputs The operator(s) to be unioned with the input.
- * @deprecated This method will be removed in future versions. Use the {@link org.apache.flink.api.common.operators.Union} operator instead.
- */
- @SuppressWarnings("unchecked")
- @Deprecated
- public void addInputs(List<? extends Operator<IN>> inputs) {
- Preconditions.checkNotNull(inputs, "The inputs may not be null.");
- this.input = createUnionCascade(this.input, (Operator<IN>[]) inputs.toArray(new Operator[inputs.size()]));
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Sets the order in which the sink must write its data. For any value other then <tt>NONE</tt>,
- * this will cause the system to perform a global sort, or try to reuse an order from a
- * previous operation.
- *
- * @param globalOrder The order to write the data in.
- */
- public void setGlobalOrder(Ordering globalOrder) {
- this.localOrdering = globalOrder;
- setRangePartitioned(globalOrder);
- }
-
- /**
- * Sets the order in which the sink must write its data. For any value other then <tt>NONE</tt>,
- * this will cause the system to perform a global sort, or try to reuse an order from a
- * previous operation.
- *
- * @param globalOrder The order to write the data in.
- * @param distribution The distribution to use for the range partitioning.
- */
- public void setGlobalOrder(Ordering globalOrder, DataDistribution distribution) {
- this.localOrdering = globalOrder;
- setRangePartitioned(globalOrder, distribution);
- }
-
- /**
- * Gets the order, in which the data sink writes its data locally. Local order means that
- * with in each fragment of the file inside the distributed file system, the data is ordered,
- * but not across file fragments.
- *
- * @return NONE, if the sink writes data in any order, or ASCENDING (resp. DESCENDING),
- * if the sink writes it data with a local ascending (resp. descending) order.
- */
- public Ordering getLocalOrder() {
- return this.localOrdering;
- }
-
- /**
- * Sets the order in which the sink must write its data within each fragment in the distributed
- * file system. For any value other then <tt>NONE</tt>, this will cause the system to perform a
- * local sort, or try to reuse an order from a previous operation.
- *
- * @param localOrder The local order to write the data in.
- */
- public void setLocalOrder(Ordering localOrder) {
- this.localOrdering = localOrder;
- }
-
- /**
- * Gets the record ordering over which the sink partitions in ranges.
- *
- * @return The record ordering over which to partition in ranges.
- */
- public Ordering getPartitionOrdering() {
- return this.partitionOrdering;
- }
-
- /**
- * Sets the sink to partition the records into ranges over the given ordering.
- *
- * @param partitionOrdering The record ordering over which to partition in ranges.
- */
- public void setRangePartitioned(Ordering partitionOrdering) {
- throw new UnsupportedOperationException(
- "Range partitioning is currently only supported with a user supplied data distribution.");
- }
-
- /**
- * Sets the sink to partition the records into ranges over the given ordering.
- * The bucket boundaries are determined using the given data distribution.
- *
- * @param partitionOrdering The record ordering over which to partition in ranges.
- * @param distribution The distribution to use for the range partitioning.
- */
- public void setRangePartitioned(Ordering partitionOrdering, DataDistribution distribution) {
- if (partitionOrdering.getNumberOfFields() != distribution.getNumberOfFields()) {
- throw new IllegalArgumentException("The number of keys in the distribution must match number of ordered fields.");
- }
-
- // TODO: check compatibility of distribution and ordering (number and order of keys, key types, etc.
- // TODO: adapt partition ordering to data distribution (use prefix of ordering)
-
- this.partitionOrdering = partitionOrdering;
- this.distribution = distribution;
- }
-
- /**
- * Gets the distribution to use for the range partitioning.
- *
- * @return The distribution to use for the range partitioning.
- */
- public DataDistribution getDataDistribution() {
- return this.distribution;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Gets the class describing this sinks output format.
- *
- * @return The output format class.
- */
- public UserCodeWrapper<? extends OutputFormat<IN>> getFormatWrapper() {
- return this.formatWrapper;
- }
-
- /**
- * Gets the class describing the output format.
- * <p>
- * This method is basically identical to {@link #getFormatWrapper()}.
- *
- * @return The class describing the output format.
- *
- * @see org.apache.flink.api.common.operators.Operator#getUserCodeWrapper()
- */
- @Override
- public UserCodeWrapper<? extends OutputFormat<IN>> getUserCodeWrapper() {
- return this.formatWrapper;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Accepts the visitor and applies it this instance. This method applies the visitor in a depth-first traversal.
- * The visitors pre-visit method is called and, if returning
- * <tt>true</tt>, the visitor is recursively applied on the single input. After the recursion returned,
- * the post-visit method is called.
- *
- * @param visitor The visitor.
- *
- * @see org.apache.flink.util.Visitable#accept(org.apache.flink.util.Visitor)
- */
- @Override
- public void accept(Visitor<Operator<?>> visitor) {
- boolean descend = visitor.preVisit(this);
- if (descend) {
- this.input.accept(visitor);
- visitor.postVisit(this);
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- @Override
- public String toString() {
- return this.name;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSourceBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSourceBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSourceBase.java
deleted file mode 100644
index 238a8e3..0000000
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GenericDataSourceBase.java
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * 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.api.common.operators.base;
-
-import org.apache.flink.api.common.io.InputFormat;
-import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.operators.OperatorInformation;
-import org.apache.flink.api.common.operators.util.UserCodeClassWrapper;
-import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
-import org.apache.flink.api.common.operators.util.UserCodeWrapper;
-import org.apache.flink.util.Visitor;
-
-/**
- * Abstract superclass for data sources in a Pact plan.
- *
- * @param <OUT> The output type of the data source
- * @param <T> The type of input format invoked by instances of this data source.
- */
-public class GenericDataSourceBase<OUT, T extends InputFormat<OUT, ?>> extends Operator<OUT> {
-
- private static final String DEFAULT_NAME = "<Unnamed Generic Data Source>";
-
- protected final UserCodeWrapper<? extends T> formatWrapper;
-
- protected String statisticsKey;
-
- /**
- * Creates a new instance for the given file using the given input format.
- *
- * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
- * @param operatorInfo The type information for the operator.
- * @param name The given name for the Pact, used in plans, logs and progress messages.
- */
- public GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo, String name) {
- super(operatorInfo, name);
-
- if (format == null) {
- throw new IllegalArgumentException("Input format may not be null.");
- }
-
- this.formatWrapper = new UserCodeObjectWrapper<T>(format);
- }
-
- /**
- * Creates a new instance for the given file using the given input format, using the default name.
- *
- * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
- * @param operatorInfo The type information for the operator.
- */
- public GenericDataSourceBase(T format, OperatorInformation<OUT> operatorInfo) {
- super(operatorInfo, DEFAULT_NAME);
-
- if (format == null) {
- throw new IllegalArgumentException("Input format may not be null.");
- }
-
- this.formatWrapper = new UserCodeObjectWrapper<T>(format);
- }
-
- /**
- * Creates a new instance for the given file using the given input format.
- *
- * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
- * @param operatorInfo The type information for the operator.
- * @param name The given name for the Pact, used in plans, logs and progress messages.
- */
- public GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo, String name) {
- super(operatorInfo, name);
-
- if (format == null) {
- throw new IllegalArgumentException("Input format may not be null.");
- }
-
- this.formatWrapper = new UserCodeClassWrapper<T>(format);
- }
-
- /**
- * Creates a new instance for the given file using the given input format, using the default name.
- *
- * @param format The {@link org.apache.flink.api.common.io.InputFormat} implementation used to read the data.
- * @param operatorInfo The type information for the operator.
- */
- public GenericDataSourceBase(Class<? extends T> format, OperatorInformation<OUT> operatorInfo) {
- super(operatorInfo, DEFAULT_NAME);
-
- if (format == null) {
- throw new IllegalArgumentException("Input format may not be null.");
- }
-
- this.formatWrapper = new UserCodeClassWrapper<T>(format);
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Gets the class describing the input format.
- *
- * @return The class describing the input format.
- */
- public UserCodeWrapper<? extends T> getFormatWrapper() {
- return this.formatWrapper;
- }
-
- /**
- * Gets the class describing the input format.
- * <p>
- * This method is basically identical to {@link #getFormatWrapper()}.
- *
- * @return The class describing the input format.
- *
- * @see org.apache.flink.api.common.operators.Operator#getUserCodeWrapper()
- */
- @Override
- public UserCodeWrapper<? extends T> getUserCodeWrapper() {
- return this.formatWrapper;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Gets the key under which statistics about this data source may be obtained from the
- * statistics cache.
- *
- * @return The statistics cache key.
- */
- public String getStatisticsKey() {
- return this.statisticsKey;
- }
-
- /**
- * Sets the key under which statistics about this data source may be obtained from the
- * statistics cache. Useful for testing purposes, when providing mock statistics.
- *
- * @param statisticsKey The key for the statistics object.
- */
- public void setStatisticsKey(String statisticsKey) {
- this.statisticsKey = statisticsKey;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Accepts the visitor and applies it this instance. Since the data sources have no inputs, no recursive descend
- * happens. The visitors pre-visit method is called and, if returning <tt>true</tt>, the post-visit method is called.
- *
- * @param visitor The visitor.
- *
- * @see org.apache.flink.util.Visitable#accept(org.apache.flink.util.Visitor)
- */
- @Override
- public void accept(Visitor<Operator<?>> visitor) {
- if (visitor.preVisit(this)) {
- visitor.postVisit(this);
- }
- }
-
-
- public String toString() {
- return this.name;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/base/MapOperatorBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/MapOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/MapOperatorBase.java
index ebb8c19..81d1fed 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/MapOperatorBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/MapOperatorBase.java
@@ -19,7 +19,12 @@
package org.apache.flink.api.common.operators.base;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
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;
@@ -46,4 +51,23 @@ public class MapOperatorBase<IN, OUT, FT extends MapFunction<IN, OUT>> extends S
public MapOperatorBase(Class<? extends FT> udf, UnaryOperatorInformation<IN, OUT> operatorInfo, String name) {
super(new UserCodeClassWrapper<FT>(udf), operatorInfo, name);
}
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ protected List<OUT> executeOnCollections(List<IN> inputData, RuntimeContext ctx) throws Exception {
+ MapFunction<IN, OUT> function = this.userFunction.getUserCodeObject();
+
+ FunctionUtils.openFunction(function, this.parameters);
+ FunctionUtils.setFunctionRuntimeContext(function, ctx);
+
+ ArrayList<OUT> result = new ArrayList<OUT>(inputData.size());
+ for (IN element : inputData) {
+ result.add(function.map(element));
+ }
+
+ FunctionUtils.closeFunction(function);
+
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d10d59eb/flink-core/src/main/java/org/apache/flink/api/common/operators/util/OperatorUtil.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/OperatorUtil.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/OperatorUtil.java
index 83ecac2..106b86a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/OperatorUtil.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/OperatorUtil.java
@@ -35,13 +35,13 @@ import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.api.common.operators.DualInputOperator;
+import org.apache.flink.api.common.operators.GenericDataSinkBase;
+import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.SingleInputOperator;
import org.apache.flink.api.common.operators.base.CoGroupOperatorBase;
import org.apache.flink.api.common.operators.base.CollectorMapOperatorBase;
import org.apache.flink.api.common.operators.base.CrossOperatorBase;
-import org.apache.flink.api.common.operators.base.GenericDataSinkBase;
-import org.apache.flink.api.common.operators.base.GenericDataSourceBase;
import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase;
import org.apache.flink.api.common.operators.base.JoinOperatorBase;