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:04 UTC
[08/23] git commit: [FLINK-1110] Implement broadcast variables for
collection-based execution
[FLINK-1110] Implement broadcast variables 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/fdad5bf8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/fdad5bf8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/fdad5bf8
Branch: refs/heads/master
Commit: fdad5bf8dc713f41c72745d4dad8911c8e819632
Parents: 84bc21e
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Sep 22 19:18:54 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 3 16:22:33 2014 +0200
----------------------------------------------------------------------
.../common/operators/CollectionExecutor.java | 13 ++
.../flink/api/java/CollectionEnvironment.java | 6 +-
.../java/operators/SingleInputUdfOperator.java | 7 ++
.../api/java/operators/TwoInputUdfOperator.java | 7 ++
...ctionExecutionWithBroadcastVariableTest.java | 124 +++++++++++++++++++
5 files changed, 152 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fdad5bf8/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
index 6643703..a6bb5dc 100644
--- 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
@@ -127,9 +127,15 @@ public class CollectionExecutor {
@SuppressWarnings("unchecked")
SingleInputOperator<IN, OUT, ?> typedOp = (SingleInputOperator<IN, OUT, ?>) operator;
+ // build the runtime context and compute broadcast variables, if necessary
RuntimeUDFContext ctx;
if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
ctx = new RuntimeUDFContext(operator.getName(), 1, 0);
+
+ for (Map.Entry<String, Operator<?>> bcInputs : operator.getBroadcastInputs().entrySet()) {
+ List<?> bcData = execute(bcInputs.getValue());
+ ctx.setBroadcastVariable(bcInputs.getKey(), bcData);
+ }
} else {
ctx = null;
}
@@ -153,6 +159,7 @@ public class CollectionExecutor {
throw new InvalidProgramException("The binary operation " + operator.getName() + " has no second input.");
}
+ // compute inputs
@SuppressWarnings("unchecked")
List<IN1> inputData1 = (List<IN1>) execute(inputOp1);
@SuppressWarnings("unchecked")
@@ -161,9 +168,15 @@ public class CollectionExecutor {
@SuppressWarnings("unchecked")
DualInputOperator<IN1, IN2, OUT, ?> typedOp = (DualInputOperator<IN1, IN2, OUT, ?>) operator;
+ // build the runtime context and compute broadcast variables, if necessary
RuntimeUDFContext ctx;
if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
ctx = new RuntimeUDFContext(operator.getName(), 1, 0);
+
+ for (Map.Entry<String, Operator<?>> bcInputs : operator.getBroadcastInputs().entrySet()) {
+ List<?> bcData = execute(bcInputs.getValue());
+ ctx.setBroadcastVariable(bcInputs.getKey(), bcData);
+ }
} else {
ctx = null;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fdad5bf8/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
index 2479e62..07a35f3 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
@@ -29,11 +29,7 @@ public class CollectionEnvironment extends ExecutionEnvironment {
Plan p = createProgramPlan(jobName);
CollectionExecutor exec = new CollectionExecutor();
-
- long start = System.currentTimeMillis();
- exec.execute(p);
- long stop = System.currentTimeMillis();
- return new JobExecutionResult(stop - start, null);
+ return exec.execute(p);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fdad5bf8/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
index d54c20e..0823c15 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/SingleInputUdfOperator.java
@@ -86,6 +86,13 @@ public abstract class SingleInputUdfOperator<IN, OUT, O extends SingleInputUdfOp
@Override
public O withBroadcastSet(DataSet<?> data, String name) {
+ if (data == null) {
+ throw new IllegalArgumentException("Broadcast variable data must not be null.");
+ }
+ if (name == null) {
+ throw new IllegalArgumentException("Broadcast variable name must not be null.");
+ }
+
if (this.broadcastVariables == null) {
this.broadcastVariables = new HashMap<String, DataSet<?>>();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fdad5bf8/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
index 38e9af9..7ca0840 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputUdfOperator.java
@@ -91,6 +91,13 @@ public abstract class TwoInputUdfOperator<IN1, IN2, OUT, O extends TwoInputUdfOp
@Override
public O withBroadcastSet(DataSet<?> data, String name) {
+ if (data == null) {
+ throw new IllegalArgumentException("Broadcast variable data must not be null.");
+ }
+ if (name == null) {
+ throw new IllegalArgumentException("Broadcast variable name must not be null.");
+ }
+
if (this.broadcastVariables == null) {
this.broadcastVariables = new HashMap<String, DataSet<?>>();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/fdad5bf8/flink-java/src/test/java/org/apache/flink/api/common/operators/CollectionExecutionWithBroadcastVariableTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/CollectionExecutionWithBroadcastVariableTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/CollectionExecutionWithBroadcastVariableTest.java
new file mode 100644
index 0000000..adf2d96
--- /dev/null
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/CollectionExecutionWithBroadcastVariableTest.java
@@ -0,0 +1,124 @@
+/**
+ * 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 static org.junit.Assert.*;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.common.functions.RichCrossFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.CollectionEnvironment;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.junit.Test;
+
+@SuppressWarnings("serial")
+public class CollectionExecutionWithBroadcastVariableTest {
+
+ private static final String BC_VAR_NAME = "BC";
+
+ private final String[] TEST_DATA = { "A", "B", "C", "D" };
+ private final String SUFFIX = "-suffixed";
+
+ @Test
+ public void testUnaryOp() {
+ try {
+ ExecutionEnvironment env = new CollectionEnvironment();
+
+ DataSet<String> bcData = env.fromElements(SUFFIX);
+
+ List<String> result = new ArrayList<String>();
+
+ env.fromElements(TEST_DATA)
+ .map(new SuffixAppender()).withBroadcastSet(bcData, BC_VAR_NAME)
+ .output(new LocalCollectionOutputFormat<String>(result));
+
+ env.execute();
+
+ assertEquals(TEST_DATA.length, result.size());
+ for (String s : result) {
+ assertTrue(s.indexOf(SUFFIX) > 0);
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testBinaryOp() {
+ try {
+ ExecutionEnvironment env = new CollectionEnvironment();
+
+ DataSet<String> bcData = env.fromElements(SUFFIX);
+ DataSet<String> inData = env.fromElements(TEST_DATA);
+
+ List<String> result = new ArrayList<String>();
+
+ inData.cross(inData).with(new SuffixCross()).withBroadcastSet(bcData, BC_VAR_NAME)
+ .output(new LocalCollectionOutputFormat<String>(result));
+
+ env.execute();
+
+ assertEquals(TEST_DATA.length * TEST_DATA.length, result.size());
+ for (String s : result) {
+ assertTrue(s.indexOf(SUFFIX) == 2);
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ public static final class SuffixAppender extends RichMapFunction<String, String> {
+
+ private String suffix;
+
+ @Override
+ public void open(Configuration parameters) {
+ suffix = getRuntimeContext().<String>getBroadcastVariable(BC_VAR_NAME).get(0);
+ }
+
+ @Override
+ public String map(String value) {
+ return value + suffix;
+ }
+ }
+
+ public static final class SuffixCross extends RichCrossFunction<String, String, String> {
+
+ private String suffix;
+
+ @Override
+ public void open(Configuration parameters) {
+ suffix = getRuntimeContext().<String>getBroadcastVariable(BC_VAR_NAME).get(0);
+ }
+
+ @Override
+ public String cross(String s1, String s2) {
+ return s1 + s2 + suffix;
+ }
+ }
+}