You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2017/04/06 08:58:07 UTC
[1/5] flink git commit: [hotfix] [py] Code cleanup - PythonPlanBinder
Repository: flink
Updated Branches:
refs/heads/master 89d9dec38 -> 5ff9c99ff
[hotfix] [py] Code cleanup - PythonPlanBinder
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/940d16c7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/940d16c7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/940d16c7
Branch: refs/heads/master
Commit: 940d16c77731a8558b13cd9b3406862229e4bd4f
Parents: bba49d6
Author: zentol <ch...@apache.org>
Authored: Thu Mar 30 23:46:44 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Thu Apr 6 10:57:10 2017 +0200
----------------------------------------------------------------------
.../flink/python/api/PythonPlanBinder.java | 45 ++++++++++----------
1 file changed, 22 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/940d16c7/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
index 7c228e1..733a6fb 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
@@ -12,13 +12,6 @@
*/
package org.apache.flink.python.api;
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Arrays;
-import java.util.Random;
-
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.operators.Keys;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
@@ -40,24 +33,32 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.python.api.functions.util.NestedKeyDiscarder;
-import org.apache.flink.python.api.functions.util.StringTupleDeserializerMap;
import org.apache.flink.python.api.PythonOperationInfo.DatasizeHint;
-import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.HUGE;
-import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.NONE;
-import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.TINY;
import org.apache.flink.python.api.functions.PythonCoGroup;
-import org.apache.flink.python.api.functions.util.IdentityGroupReduce;
import org.apache.flink.python.api.functions.PythonMapPartition;
+import org.apache.flink.python.api.functions.util.IdentityGroupReduce;
import org.apache.flink.python.api.functions.util.KeyDiscarder;
+import org.apache.flink.python.api.functions.util.NestedKeyDiscarder;
import org.apache.flink.python.api.functions.util.SerializerMap;
import org.apache.flink.python.api.functions.util.StringDeserializerMap;
+import org.apache.flink.python.api.functions.util.StringTupleDeserializerMap;
import org.apache.flink.python.api.streaming.plan.PythonPlanStreamer;
import org.apache.flink.python.api.util.SetCache;
import org.apache.flink.runtime.filecache.FileCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Arrays;
+import java.util.Random;
+
+import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.HUGE;
+import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.NONE;
+import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.TINY;
+
/**
* This class allows the execution of a Flink plan written in python.
*/
@@ -116,7 +117,7 @@ public class PythonPlanBinder {
binder.runPlan(Arrays.copyOfRange(args, 1, args.length));
}
- public PythonPlanBinder() throws IOException {
+ public PythonPlanBinder() {
Configuration conf = GlobalConfiguration.loadConfiguration();
FLINK_PYTHON2_BINARY_PATH = conf.getString(FLINK_PYTHON2_BINARY_KEY, "python");
FLINK_PYTHON3_BINARY_PATH = conf.getString(FLINK_PYTHON3_BINARY_KEY, "python3");
@@ -164,6 +165,7 @@ public class PythonPlanBinder {
}
//=====Setup========================================================================================================
+
/**
* Copies all files to a common directory (FLINK_PYTHON_FILE_PATH). This allows us to distribute it as one big
* package, and resolves PYTHONPATH issues.
@@ -186,7 +188,7 @@ public class PythonPlanBinder {
}
}
- private static void clearPath(String path) throws IOException, URISyntaxException {
+ private static void clearPath(String path) throws IOException {
FileSystem fs = FileSystem.get(new Path(path).toUri());
if (fs.exists(new Path(path))) {
fs.delete(new Path(path), true);
@@ -232,9 +234,9 @@ public class PythonPlanBinder {
local.delete(new Path(FLINK_PYTHON_FILE_PATH), true);
local.delete(new Path(FLINK_TMP_DATA_DIR), true);
streamer.close();
- } catch (NullPointerException npe) {
+ } catch (NullPointerException ignored) {
} catch (IOException ioe) {
- LOG.error("PythonAPI file cleanup failed. " + ioe.getMessage());
+ LOG.error("PythonAPI file cleanup failed. {}", ioe.getMessage());
} catch (URISyntaxException use) { // can't occur
}
}
@@ -249,6 +251,7 @@ public class PythonPlanBinder {
}
//====Environment===================================================================================================
+
/**
* This enum contains the identifiers for all supported environment parameters.
*/
@@ -285,6 +288,7 @@ public class PythonPlanBinder {
}
//====Operations====================================================================================================
+
/**
* This enum contains the identifiers for all supported DataSet operations.
*/
@@ -300,12 +304,7 @@ public class PythonPlanBinder {
Integer operationCount = (Integer) streamer.getRecord(true);
for (int x = 0; x < operationCount; x++) {
PythonOperationInfo info = new PythonOperationInfo(streamer, currentEnvironmentID);
- Operation op;
- try {
- op = Operation.valueOf(info.identifier.toUpperCase());
- } catch (IllegalArgumentException iae) {
- throw new IllegalArgumentException("Invalid operation specified: " + info.identifier);
- }
+ Operation op = Operation.valueOf(info.identifier.toUpperCase());
switch (op) {
case SOURCE_CSV:
createCsvSource(info);
[2/5] flink git commit: [FLINK-6223] [py] Rework PythonPlanBinder
generics
Posted by ch...@apache.org.
[FLINK-6223] [py] Rework PythonPlanBinder generics
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bba49d67
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bba49d67
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bba49d67
Branch: refs/heads/master
Commit: bba49d673f9a36d4291bc74f07895e957de75bce
Parents: 8f78e9d
Author: zentol <ch...@apache.org>
Authored: Thu Mar 30 23:45:14 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Thu Apr 6 10:57:10 2017 +0200
----------------------------------------------------------------------
.../flink/python/api/PythonPlanBinder.java | 339 +++++++++----------
.../apache/flink/python/api/util/SetCache.java | 204 +++++++++++
2 files changed, 367 insertions(+), 176 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bba49d67/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
index a3cae4a..7c228e1 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
@@ -17,11 +17,12 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.Random;
import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.operators.Keys;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.LocalEnvironment;
@@ -29,12 +30,11 @@ import org.apache.flink.api.java.io.PrintingOutputFormat;
import org.apache.flink.api.java.io.TupleCsvInputFormat;
import org.apache.flink.api.java.operators.CoGroupRawOperator;
import org.apache.flink.api.java.operators.CrossOperator.DefaultCross;
-import org.apache.flink.api.java.operators.Grouping;
-import org.apache.flink.api.common.operators.Keys;
import org.apache.flink.api.java.operators.SortedGrouping;
import org.apache.flink.api.java.operators.UdfOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
@@ -53,6 +53,7 @@ import org.apache.flink.python.api.functions.util.KeyDiscarder;
import org.apache.flink.python.api.functions.util.SerializerMap;
import org.apache.flink.python.api.functions.util.StringDeserializerMap;
import org.apache.flink.python.api.streaming.plan.PythonPlanStreamer;
+import org.apache.flink.python.api.util.SetCache;
import org.apache.flink.runtime.filecache.FileCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -92,7 +93,7 @@ public class PythonPlanBinder {
private static String FLINK_HDFS_PATH = "hdfs:/tmp";
public static final String FLINK_TMP_DATA_DIR = System.getProperty("java.io.tmpdir") + File.separator + "flink_data";
- private HashMap<Integer, Object> sets = new HashMap<>();
+ private final SetCache sets = new SetCache();
public ExecutionEnvironment env;
private int currentEnvironmentID = 0;
private PythonPlanStreamer streamer;
@@ -242,7 +243,7 @@ public class PythonPlanBinder {
private void receivePlan() throws IOException {
env = ExecutionEnvironment.getExecutionEnvironment();
//IDs used in HashMap of sets are only unique for each environment
- sets.clear();
+ sets.reset();
receiveParameters();
receiveOperations();
}
@@ -263,18 +264,18 @@ public class PythonPlanBinder {
Tuple value = (Tuple) streamer.getRecord(true);
switch (Parameters.valueOf(((String) value.getField(0)).toUpperCase())) {
case DOP:
- Integer dop = (Integer) value.getField(1);
+ Integer dop = value.<Integer>getField(1);
env.setParallelism(dop);
break;
case MODE:
- FLINK_HDFS_PATH = (Boolean) value.getField(1) ? "file:/tmp/flink" : "hdfs:/tmp/flink";
+ FLINK_HDFS_PATH = value.<Boolean>getField(1) ? "file:/tmp/flink" : "hdfs:/tmp/flink";
break;
case RETRY:
- int retry = (Integer) value.getField(1);
+ int retry = value.<Integer>getField(1);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(retry, 10000L));
break;
case ID:
- currentEnvironmentID = (Integer) value.getField(1);
+ currentEnvironmentID = value.<Integer>getField(1);
break;
}
}
@@ -352,40 +353,40 @@ public class PythonPlanBinder {
createUnionOperation(info);
break;
case COGROUP:
- createCoGroupOperation(info);
+ createCoGroupOperation(info, info.types);
break;
case CROSS:
- createCrossOperation(NONE, info);
+ createCrossOperation(NONE, info, info.types);
break;
case CROSS_H:
- createCrossOperation(HUGE, info);
+ createCrossOperation(HUGE, info, info.types);
break;
case CROSS_T:
- createCrossOperation(TINY, info);
+ createCrossOperation(TINY, info, info.types);
break;
case FILTER:
- createFilterOperation(info);
+ createFilterOperation(info, info.types);
break;
case FLATMAP:
- createFlatMapOperation(info);
+ createFlatMapOperation(info, info.types);
break;
case GROUPREDUCE:
createGroupReduceOperation(info);
break;
case JOIN:
- createJoinOperation(NONE, info);
+ createJoinOperation(NONE, info, info.types);
break;
case JOIN_H:
- createJoinOperation(HUGE, info);
+ createJoinOperation(HUGE, info, info.types);
break;
case JOIN_T:
- createJoinOperation(TINY, info);
+ createJoinOperation(TINY, info, info.types);
break;
case MAP:
- createMapOperation(info);
+ createMapOperation(info, info.types);
break;
case MAPPARTITION:
- createMapPartitionOperation(info);
+ createMapPartitionOperation(info, info.types);
break;
case REDUCE:
createReduceOperation(info);
@@ -399,57 +400,54 @@ public class PythonPlanBinder {
}
@SuppressWarnings("unchecked")
- private void createCsvSource(PythonOperationInfo info) throws IOException {
+ private <T extends Tuple> void createCsvSource(PythonOperationInfo info) {
if (!(info.types instanceof TupleTypeInfo)) {
throw new RuntimeException("The output type of a csv source has to be a tuple. The derived type is " + info);
}
Path path = new Path(info.path);
String lineD = info.lineDelimiter;
String fieldD = info.fieldDelimiter;
- TupleTypeInfo<?> types = (TupleTypeInfo) info.types;
- sets.put(info.setID, env.createInput(new TupleCsvInputFormat(path, lineD, fieldD, types), info.types).setParallelism(getParallelism(info)).name("CsvSource")
- .map(new SerializerMap<>()).setParallelism(getParallelism(info)).name("CsvSourcePostStep"));
+ TupleTypeInfo<T> types = (TupleTypeInfo<T>) info.types;
+ sets.add(info.setID, env.createInput(new TupleCsvInputFormat<>(path, lineD, fieldD, types), types).setParallelism(getParallelism(info)).name("CsvSource")
+ .map(new SerializerMap<T>()).setParallelism(getParallelism(info)).name("CsvSourcePostStep"));
}
- private void createTextSource(PythonOperationInfo info) throws IOException {
- sets.put(info.setID, env.readTextFile(info.path).setParallelism(getParallelism(info)).name("TextSource")
- .map(new SerializerMap<String>()).setParallelism(getParallelism(info)).name("TextSourcePostStep"));
+ private void createTextSource(PythonOperationInfo info) {
+ sets.add(info.setID, env.readTextFile(info.path).setParallelism(getParallelism(info)).name("TextSource")
+ .map(new SerializerMap<String>()).setParallelism(getParallelism(info)).name("TextSourcePostStep"));
}
- private void createValueSource(PythonOperationInfo info) throws IOException {
- sets.put(info.setID, env.fromElements(info.values).setParallelism(getParallelism(info)).name("ValueSource")
- .map(new SerializerMap<>()).setParallelism(getParallelism(info)).name("ValueSourcePostStep"));
+ private void createValueSource(PythonOperationInfo info) {
+ sets.add(info.setID, env.fromElements(info.values).setParallelism(getParallelism(info)).name("ValueSource")
+ .map(new SerializerMap<>()).setParallelism(getParallelism(info)).name("ValueSourcePostStep"));
}
- private void createSequenceSource(PythonOperationInfo info) throws IOException {
- sets.put(info.setID, env.generateSequence(info.frm, info.to).setParallelism(getParallelism(info)).name("SequenceSource")
- .map(new SerializerMap<Long>()).setParallelism(getParallelism(info)).name("SequenceSourcePostStep"));
+ private void createSequenceSource(PythonOperationInfo info) {
+ sets.add(info.setID, env.generateSequence(info.frm, info.to).setParallelism(getParallelism(info)).name("SequenceSource")
+ .map(new SerializerMap<Long>()).setParallelism(getParallelism(info)).name("SequenceSourcePostStep"));
}
- @SuppressWarnings("unchecked")
- private void createCsvSink(PythonOperationInfo info) throws IOException {
- DataSet parent = (DataSet) sets.get(info.parentID);
+ private void createCsvSink(PythonOperationInfo info) {
+ DataSet<byte[]> parent = sets.getDataSet(info.parentID);
parent.map(new StringTupleDeserializerMap()).setParallelism(getParallelism(info)).name("CsvSinkPreStep")
.writeAsCsv(info.path, info.lineDelimiter, info.fieldDelimiter, info.writeMode).setParallelism(getParallelism(info)).name("CsvSink");
}
- @SuppressWarnings("unchecked")
- private void createTextSink(PythonOperationInfo info) throws IOException {
- DataSet parent = (DataSet) sets.get(info.parentID);
+ private void createTextSink(PythonOperationInfo info) {
+ DataSet<byte[]> parent = sets.getDataSet(info.parentID);
parent.map(new StringDeserializerMap()).setParallelism(getParallelism(info))
.writeAsText(info.path, info.writeMode).setParallelism(getParallelism(info)).name("TextSink");
}
- @SuppressWarnings("unchecked")
- private void createPrintSink(PythonOperationInfo info) throws IOException {
- DataSet parent = (DataSet) sets.get(info.parentID);
+ private void createPrintSink(PythonOperationInfo info) {
+ DataSet<byte[]> parent = sets.getDataSet(info.parentID);
parent.map(new StringDeserializerMap()).setParallelism(getParallelism(info)).name("PrintSinkPreStep")
- .output(new PrintingOutputFormat(info.toError)).setParallelism(getParallelism(info));
+ .output(new PrintingOutputFormat<String>(info.toError)).setParallelism(getParallelism(info));
}
- private void createBroadcastVariable(PythonOperationInfo info) throws IOException {
- UdfOperator<?> op1 = (UdfOperator) sets.get(info.parentID);
- DataSet<?> op2 = (DataSet) sets.get(info.otherID);
+ private void createBroadcastVariable(PythonOperationInfo info) {
+ UdfOperator<?> op1 = (UdfOperator<?>) sets.getDataSet(info.parentID);
+ DataSet<?> op2 = sets.getDataSet(info.otherID);
op1.withBroadcastSet(op2, info.name);
Configuration c = op1.getParameters();
@@ -465,82 +463,80 @@ public class PythonPlanBinder {
op1.withParameters(c);
}
- @SuppressWarnings("unchecked")
- private void createDistinctOperation(PythonOperationInfo info) throws IOException {
- DataSet op = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op.distinct(info.keys).setParallelism(getParallelism(info)).name("Distinct")
- .map(new KeyDiscarder()).setParallelism(getParallelism(info)).name("DistinctPostStep"));
- }
-
- @SuppressWarnings("unchecked")
- private void createFirstOperation(PythonOperationInfo info) throws IOException {
- Object op = sets.get(info.parentID);
- if (op instanceof DataSet) {
- sets.put(info.setID, ((DataSet) op).first(info.count).setParallelism(getParallelism(info)).name("First"));
- return;
- }
- if (op instanceof UnsortedGrouping) {
- sets.put(info.setID, ((UnsortedGrouping) op).first(info.count).setParallelism(getParallelism(info)).name("First")
- .map(new KeyDiscarder()).setParallelism(getParallelism(info)).name("FirstPostStep"));
- return;
- }
- if (op instanceof SortedGrouping) {
- sets.put(info.setID, ((SortedGrouping) op).first(info.count).setParallelism(getParallelism(info)).name("First")
- .map(new KeyDiscarder()).setParallelism(getParallelism(info)).name("FirstPostStep"));
+ private <K extends Tuple> void createDistinctOperation(PythonOperationInfo info) {
+ DataSet<Tuple2<K, byte[]>> op = sets.getDataSet(info.parentID);
+ DataSet<byte[]> result = op
+ .distinct(info.keys).setParallelism(getParallelism(info)).name("Distinct")
+ .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("DistinctPostStep");
+ sets.add(info.setID, result);
+ }
+
+ private <K extends Tuple> void createFirstOperation(PythonOperationInfo info) {
+ if (sets.isDataSet(info.parentID)) {
+ DataSet<byte[]> op = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op
+ .first(info.count).setParallelism(getParallelism(info)).name("First"));
+ } else if (sets.isUnsortedGrouping(info.parentID)) {
+ UnsortedGrouping<Tuple2<K, byte[]>> op = sets.getUnsortedGrouping(info.parentID);
+ sets.add(info.setID, op
+ .first(info.count).setParallelism(getParallelism(info)).name("First")
+ .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("FirstPostStep"));
+ } else if (sets.isSortedGrouping(info.parentID)) {
+ SortedGrouping<Tuple2<K, byte[]>> op = sets.getSortedGrouping(info.parentID);
+ sets.add(info.setID, op
+ .first(info.count).setParallelism(getParallelism(info)).name("First")
+ .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("FirstPostStep"));
}
}
- private void createGroupOperation(PythonOperationInfo info) throws IOException {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op1.groupBy(info.keys));
+ private void createGroupOperation(PythonOperationInfo info) {
+ DataSet<?> op1 = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op1.groupBy(info.keys));
}
- @SuppressWarnings("unchecked")
- private void createHashPartitionOperation(PythonOperationInfo info) throws IOException {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op1.partitionByHash(info.keys).setParallelism(getParallelism(info))
- .map(new KeyDiscarder()).setParallelism(getParallelism(info)).name("HashPartitionPostStep"));
+ private <K extends Tuple> void createHashPartitionOperation(PythonOperationInfo info) {
+ DataSet<Tuple2<K, byte[]>> op1 = sets.getDataSet(info.parentID);
+ DataSet<byte[]> result = op1
+ .partitionByHash(info.keys).setParallelism(getParallelism(info))
+ .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("HashPartitionPostStep");
+ sets.add(info.setID, result);
}
- private void createRebalanceOperation(PythonOperationInfo info) throws IOException {
- DataSet op = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op.rebalance().setParallelism(getParallelism(info)).name("Rebalance"));
+ private void createRebalanceOperation(PythonOperationInfo info) {
+ DataSet<?> op = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op.rebalance().setParallelism(getParallelism(info)).name("Rebalance"));
}
- private void createSortOperation(PythonOperationInfo info) throws IOException {
- Grouping op1 = (Grouping) sets.get(info.parentID);
- if (op1 instanceof UnsortedGrouping) {
- sets.put(info.setID, ((UnsortedGrouping) op1).sortGroup(info.field, info.order));
- return;
- }
- if (op1 instanceof SortedGrouping) {
- sets.put(info.setID, ((SortedGrouping) op1).sortGroup(info.field, info.order));
+ private void createSortOperation(PythonOperationInfo info) {
+ if (sets.isDataSet(info.parentID)) {
+ throw new IllegalArgumentException("sort() can not be applied on a DataSet.");
+ } else if (sets.isUnsortedGrouping(info.parentID)) {
+ sets.add(info.setID, sets.getUnsortedGrouping(info.parentID).sortGroup(info.field, info.order));
+ } else if (sets.isSortedGrouping(info.parentID)) {
+ sets.add(info.setID, sets.getSortedGrouping(info.parentID).sortGroup(info.field, info.order));
}
}
- @SuppressWarnings("unchecked")
- private void createUnionOperation(PythonOperationInfo info) throws IOException {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- DataSet op2 = (DataSet) sets.get(info.otherID);
- sets.put(info.setID, op1.union(op2).setParallelism(getParallelism(info)).name("Union"));
+ private <IN> void createUnionOperation(PythonOperationInfo info) {
+ DataSet<IN> op1 = sets.getDataSet(info.parentID);
+ DataSet<IN> op2 = sets.getDataSet(info.otherID);
+ sets.add(info.setID, op1.union(op2).setParallelism(getParallelism(info)).name("Union"));
}
- @SuppressWarnings("unchecked")
- private void createCoGroupOperation(PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- DataSet op2 = (DataSet) sets.get(info.otherID);
- Keys.ExpressionKeys<?> key1 = new Keys.ExpressionKeys(info.keys1, op1.getType());
- Keys.ExpressionKeys<?> key2 = new Keys.ExpressionKeys(info.keys2, op2.getType());
- PythonCoGroup pcg = new PythonCoGroup(info.envID, info.setID, info.types);
- sets.put(info.setID, new CoGroupRawOperator(op1, op2, key1, key2, pcg, info.types, info.name).setParallelism(getParallelism(info)));
+ private <IN1, IN2, OUT> void createCoGroupOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN1> op1 = sets.getDataSet(info.parentID);
+ DataSet<IN2> op2 = sets.getDataSet(info.otherID);
+ Keys.ExpressionKeys<IN1> key1 = new Keys.ExpressionKeys<>(info.keys1, op1.getType());
+ Keys.ExpressionKeys<IN2> key2 = new Keys.ExpressionKeys<>(info.keys2, op2.getType());
+ PythonCoGroup<IN1, IN2, OUT> pcg = new PythonCoGroup<>(info.envID, info.setID, type);
+ sets.add(info.setID, new CoGroupRawOperator<>(op1, op2, key1, key2, pcg, type, info.name).setParallelism(getParallelism(info)));
}
- @SuppressWarnings("unchecked")
- private void createCrossOperation(DatasizeHint mode, PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- DataSet op2 = (DataSet) sets.get(info.otherID);
+ private <IN1, IN2, OUT> void createCrossOperation(DatasizeHint mode, PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN1> op1 = sets.getDataSet(info.parentID);
+ DataSet<IN2> op2 = sets.getDataSet(info.otherID);
- DefaultCross defaultResult;
+ DefaultCross<IN1, IN2> defaultResult;
switch (mode) {
case NONE:
defaultResult = op1.cross(op2);
@@ -557,119 +553,110 @@ public class PythonPlanBinder {
defaultResult.setParallelism(getParallelism(info));
if (info.usesUDF) {
- sets.put(info.setID, defaultResult.mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, defaultResult.mapPartition(new PythonMapPartition<Tuple2<IN1, IN2>, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
} else {
- sets.put(info.setID, defaultResult.name("DefaultCross"));
+ sets.add(info.setID, defaultResult.name("DefaultCross"));
}
}
- @SuppressWarnings("unchecked")
- private void createFilterOperation(PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op1.mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name));
+ private <IN, OUT> void createFilterOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN> op1 = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
}
- @SuppressWarnings("unchecked")
- private void createFlatMapOperation(PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op1.mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name));
+ private <IN, OUT> void createFlatMapOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN> op1 = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
}
private void createGroupReduceOperation(PythonOperationInfo info) {
- Object op1 = sets.get(info.parentID);
- if (op1 instanceof DataSet) {
- sets.put(info.setID, applyGroupReduceOperation((DataSet) op1, info));
- return;
- }
- if (op1 instanceof UnsortedGrouping) {
- sets.put(info.setID, applyGroupReduceOperation((UnsortedGrouping) op1, info));
- return;
- }
- if (op1 instanceof SortedGrouping) {
- sets.put(info.setID, applyGroupReduceOperation((SortedGrouping) op1, info));
+ if (sets.isDataSet(info.parentID)) {
+ sets.add(info.setID, applyGroupReduceOperation(sets.getDataSet(info.parentID), info, info.types));
+ } else if (sets.isUnsortedGrouping(info.parentID)) {
+ sets.add(info.setID, applyGroupReduceOperation(sets.getUnsortedGrouping(info.parentID), info, info.types));
+ } else if (sets.isSortedGrouping(info.parentID)) {
+ sets.add(info.setID, applyGroupReduceOperation(sets.getSortedGrouping(info.parentID), info, info.types));
}
}
- @SuppressWarnings("unchecked")
- private DataSet applyGroupReduceOperation(DataSet op1, PythonOperationInfo info) {
- return op1.reduceGroup(new IdentityGroupReduce()).setCombinable(false).name("PythonGroupReducePreStep").setParallelism(getParallelism(info))
- .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name);
+ private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(DataSet<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
+ return op1
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).name("PythonGroupReducePreStep").setParallelism(getParallelism(info))
+ .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
}
- @SuppressWarnings("unchecked")
- private DataSet applyGroupReduceOperation(UnsortedGrouping op1, PythonOperationInfo info) {
- return op1.reduceGroup(new IdentityGroupReduce()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name);
+ private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(UnsortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
+ return op1
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
+ .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
}
- @SuppressWarnings("unchecked")
- private DataSet applyGroupReduceOperation(SortedGrouping op1, PythonOperationInfo info) {
- return op1.reduceGroup(new IdentityGroupReduce()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name);
+ private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(SortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
+ return op1
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
+ .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
}
- @SuppressWarnings("unchecked")
- private void createJoinOperation(DatasizeHint mode, PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- DataSet op2 = (DataSet) sets.get(info.otherID);
+ private <IN1, IN2, OUT> void createJoinOperation(DatasizeHint mode, PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN1> op1 = sets.getDataSet(info.parentID);
+ DataSet<IN2> op2 = sets.getDataSet(info.otherID);
if (info.usesUDF) {
- sets.put(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))
- .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))
+ .mapPartition(new PythonMapPartition<Tuple2<byte[], byte[]>, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
} else {
- sets.put(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)));
+ sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)));
}
}
- @SuppressWarnings("unchecked")
- private DataSet createDefaultJoin(DataSet op1, DataSet op2, String[] firstKeys, String[] secondKeys, DatasizeHint mode, int parallelism) {
+ private <IN1, IN2> DataSet<Tuple2<byte[], byte[]>> createDefaultJoin(DataSet<IN1> op1, DataSet<IN2> op2, String[] firstKeys, String[] secondKeys, DatasizeHint mode, int parallelism) {
switch (mode) {
case NONE:
- return op1.join(op2).where(firstKeys).equalTo(secondKeys).setParallelism(parallelism)
- .map(new NestedKeyDiscarder()).setParallelism(parallelism).name("DefaultJoinPostStep");
+ return op1
+ .join(op2).where(firstKeys).equalTo(secondKeys).setParallelism(parallelism)
+ .map(new NestedKeyDiscarder<Tuple2<IN1, IN2>>()).setParallelism(parallelism).name("DefaultJoinPostStep");
case HUGE:
- return op1.joinWithHuge(op2).where(firstKeys).equalTo(secondKeys).setParallelism(parallelism)
- .map(new NestedKeyDiscarder()).setParallelism(parallelism).name("DefaultJoinPostStep");
+ return op1
+ .joinWithHuge(op2).where(firstKeys).equalTo(secondKeys).setParallelism(parallelism)
+ .map(new NestedKeyDiscarder<Tuple2<IN1, IN2>>()).setParallelism(parallelism).name("DefaultJoinPostStep");
case TINY:
- return op1.joinWithTiny(op2).where(firstKeys).equalTo(secondKeys).setParallelism(parallelism)
- .map(new NestedKeyDiscarder()).setParallelism(parallelism).name("DefaultJoinPostStep");
+ return op1
+ .joinWithTiny(op2).where(firstKeys).equalTo(secondKeys).setParallelism(parallelism)
+ .map(new NestedKeyDiscarder<Tuple2<IN1, IN2>>()).setParallelism(parallelism).name("DefaultJoinPostStep");
default:
throw new IllegalArgumentException("Invalid join mode specified.");
}
}
- @SuppressWarnings("unchecked")
- private void createMapOperation(PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op1.mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name));
+ private <IN, OUT> void createMapOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN> op1 = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
}
- @SuppressWarnings("unchecked")
- private void createMapPartitionOperation(PythonOperationInfo info) {
- DataSet op1 = (DataSet) sets.get(info.parentID);
- sets.put(info.setID, op1.mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name));
+ private <IN, OUT> void createMapPartitionOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
+ DataSet<IN> op1 = sets.getDataSet(info.parentID);
+ sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
}
private void createReduceOperation(PythonOperationInfo info) {
- Object op1 = sets.get(info.parentID);
- if (op1 instanceof DataSet) {
- sets.put(info.setID, applyReduceOperation((DataSet) op1, info));
- return;
- }
- if (op1 instanceof UnsortedGrouping) {
- sets.put(info.setID, applyReduceOperation((UnsortedGrouping) op1, info));
+ if (sets.isDataSet(info.parentID)) {
+ sets.add(info.setID, applyReduceOperation(sets.getDataSet(info.parentID), info, info.types));
+ } else if (sets.isUnsortedGrouping(info.parentID)) {
+ sets.add(info.setID, applyReduceOperation(sets.getUnsortedGrouping(info.parentID), info, info.types));
+ } else if (sets.isSortedGrouping(info.parentID)) {
+ throw new IllegalArgumentException("Reduce cannot be applied on a SortedGrouping.");
}
}
- @SuppressWarnings("unchecked")
- private DataSet applyReduceOperation(DataSet op1, PythonOperationInfo info) {
- return op1.reduceGroup(new IdentityGroupReduce()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
- .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name);
+ private <IN, OUT> DataSet<OUT> applyReduceOperation(DataSet<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
+ return op1
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
+ .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
}
- @SuppressWarnings("unchecked")
- private DataSet applyReduceOperation(UnsortedGrouping op1, PythonOperationInfo info) {
- return op1.reduceGroup(new IdentityGroupReduce()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
- .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name);
+ private <IN, OUT> DataSet<OUT> applyReduceOperation(UnsortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
+ return op1
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
+ .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bba49d67/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/util/SetCache.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/util/SetCache.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/util/SetCache.java
new file mode 100644
index 0000000..750ba63
--- /dev/null
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/util/SetCache.java
@@ -0,0 +1,204 @@
+/*
+ * 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.python.api.util;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.operators.SortedGrouping;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A container holding {@link DataSet DataSets}, {@link SortedGrouping sorted} and{@link UnsortedGrouping unsorted}
+ * groupings.
+ */
+public class SetCache {
+
+ private enum SetType {
+
+ DATA_SET(DataSet.class.getName()),
+ UNSORTED_GROUPING(UnsortedGrouping.class.getName()),
+ SORTED_GROUPING(SortedGrouping.class.getName());
+
+ private final String className;
+
+ SetType(String className) {
+ this.className = className;
+ }
+
+ @Override
+ public String toString() {
+ return className;
+ }
+ }
+
+ private final Map<Integer, SetType> setTypes = new HashMap<>();
+
+ @SuppressWarnings("rawtypes")
+ private final Map<Integer, DataSet> dataSets = new HashMap<>();
+ @SuppressWarnings("rawtypes")
+ private final Map<Integer, UnsortedGrouping> unsortedGroupings = new HashMap<>();
+ @SuppressWarnings("rawtypes")
+ private final Map<Integer, SortedGrouping> sortedGroupings = new HashMap<>();
+
+ private int cachedID = -1;
+ private SetType cachedType = null;
+
+ /**
+ * Adds the given {@link DataSet} to this cache for the given ID.
+ *
+ * @param id Set ID
+ * @param set DataSet to add
+ * @param <D> DataSet class
+ */
+ public <D extends DataSet<?>> void add(int id, D set) {
+ cacheSetType(id, SetType.DATA_SET);
+ dataSets.put(id, set);
+ }
+
+ /**
+ * Adds the given {@link UnsortedGrouping} to this cache for the given ID.
+ *
+ * @param id Set ID
+ * @param set UnsortedGrouping to add
+ * @param <U> UnsortedGrouping class
+ */
+ public <U extends UnsortedGrouping<?>> void add(int id, U set) {
+ cacheSetType(id, SetType.UNSORTED_GROUPING);
+ unsortedGroupings.put(id, set);
+ }
+
+ /**
+ * Adds the given {@link SortedGrouping} to this cache for the given ID.
+ *
+ * @param id Set ID
+ * @param set SortedGrouping to add
+ * @param <S> SortedGrouping class
+ */
+ public <S extends SortedGrouping<?>> void add(int id, S set) {
+ cacheSetType(id, SetType.SORTED_GROUPING);
+ sortedGroupings.put(id, set);
+ }
+
+ private <T> void cacheSetType(int id, SetType type) {
+ SetType prior = setTypes.put(id, type);
+ if (prior != null) {
+ throw new IllegalStateException("Set ID " + id + " used to denote multiple sets.");
+ }
+ }
+
+ /**
+ * Checks whether the cached set for the given ID is a {@link DataSet}.
+ *
+ * @param id id of set to check
+ * @return true, if the cached set is a DataSet, false otherwise
+ */
+ public boolean isDataSet(int id) {
+ return isType(id, SetType.DATA_SET);
+ }
+
+ /**
+ * Checks whether the cached set for the given ID is an {@link UnsortedGrouping}.
+ *
+ * @param id id of set to check
+ * @return true, if the cached set is an UnsortedGrouping, false otherwise
+ */
+ public boolean isUnsortedGrouping(int id) {
+ return isType(id, SetType.UNSORTED_GROUPING);
+ }
+
+ /**
+ * Checks whether the cached set for the given ID is a {@link SortedGrouping}.
+ *
+ * @param id Set ID
+ * @return true, if the cached set is a SortedGrouping, false otherwise
+ */
+ public boolean isSortedGrouping(int id) {
+ return isType(id, SetType.SORTED_GROUPING);
+ }
+
+ private boolean isType(int id, SetType type) {
+ if (cachedID != id) {
+ cachedID = id;
+ cachedType = setTypes.get(id);
+ if (cachedType == null) {
+ throw new IllegalStateException("No set exists for the given ID " + id);
+ }
+ }
+ return cachedType == type;
+ }
+
+ /**
+ * Returns the cached {@link DataSet} for the given ID.
+ *
+ * @param id Set ID
+ * @param <T> DataSet type
+ * @return Cached DataSet
+ * @throws IllegalStateException if the cached set is not a DataSet
+ */
+ @SuppressWarnings("unchecked")
+ public <T> DataSet<T> getDataSet(int id) {
+ return verifyType(id, dataSets.get(id), SetType.DATA_SET);
+ }
+
+ /**
+ * Returns the cached {@link UnsortedGrouping} for the given ID.
+ *
+ * @param id Set ID
+ * @param <T> UnsortedGrouping type
+ * @return Cached UnsortedGrouping
+ * @throws IllegalStateException if the cached set is not an UnsortedGrouping
+ */
+ @SuppressWarnings("unchecked")
+ public <T> UnsortedGrouping<T> getUnsortedGrouping(int id) {
+ return verifyType(id, unsortedGroupings.get(id), SetType.UNSORTED_GROUPING);
+ }
+
+ /**
+ * Returns the cached {@link SortedGrouping} for the given ID.
+ *
+ * @param id Set ID
+ * @param <T> SortedGrouping type
+ * @return Cached SortedGrouping
+ * @throws IllegalStateException if the cached set is not a SortedGrouping
+ */
+ @SuppressWarnings("unchecked")
+ public <T> SortedGrouping<T> getSortedGrouping(int id) {
+ return verifyType(id, sortedGroupings.get(id), SetType.SORTED_GROUPING);
+ }
+
+ private <X> X verifyType(int id, X set, SetType type) {
+ if (set == null) {
+ SetType actualType = setTypes.get(id);
+ throw new IllegalStateException("Set ID " + id + " did not denote a " + type + ", but a " + actualType + " instead.");
+ }
+ return set;
+ }
+
+ /**
+ * Resets this SetCache, removing any cached sets.
+ */
+ public void reset() {
+ setTypes.clear();
+
+ dataSets.clear();
+ unsortedGroupings.clear();
+ sortedGroupings.clear();
+ }
+}
[5/5] flink git commit: [FLINK-6229] [py] Rework setup of
PythonPlanBinder
Posted by ch...@apache.org.
[FLINK-6229] [py] Rework setup of PythonPlanBinder
- make file/argument split more readable
- pass on Paths where applicable instead of recreating them every time
- rename PPB#clearPath to more appropriate deleteIfExists
- simplify PPB#copyFile
- simplify PPB#startPython
- use UUID#randomUUID() instead of Random.nextInt()
- remove several invalid exception declarations
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5ff9c99f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5ff9c99f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5ff9c99f
Branch: refs/heads/master
Commit: 5ff9c99ff193725c125f2b4c450411db97b6f3b4
Parents: bdcebfd
Author: zentol <ch...@apache.org>
Authored: Fri Mar 31 12:11:40 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Thu Apr 6 10:57:11 2017 +0200
----------------------------------------------------------------------
.../flink/python/api/PythonPlanBinder.java | 283 ++++++++-----------
1 file changed, 124 insertions(+), 159 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5ff9c99f/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
index b6181b4..2378d60 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
@@ -49,9 +49,8 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
-import java.net.URISyntaxException;
import java.util.Arrays;
-import java.util.Random;
+import java.util.UUID;
import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.HUGE;
import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.NONE;
@@ -68,9 +67,6 @@ public class PythonPlanBinder {
public static final String PLANBINDER_CONFIG_BCVAR_COUNT = "PLANBINDER_BCVAR_COUNT";
public static final String PLANBINDER_CONFIG_BCVAR_NAME_PREFIX = "PLANBINDER_BCVAR_";
-
- private static final Random r = new Random();
-
public static final String PLAN_ARGUMENTS_KEY = "python.plan.arguments";
private static final String FLINK_PYTHON_REL_LOCAL_PATH = File.separator + "resources" + File.separator + "python";
@@ -80,10 +76,9 @@ public class PythonPlanBinder {
private final String pythonLibraryPath;
private final String tmpPlanFilesDir;
- private String tmpDistributedDir;
+ private Path tmpDistributedDir;
private final SetCache sets = new SetCache();
- public ExecutionEnvironment env;
private int currentEnvironmentID = 0;
private PythonPlanStreamer streamer;
@@ -108,9 +103,9 @@ public class PythonPlanBinder {
String configuredPlanTmpPath = globalConfig.getString(PythonOptions.PLAN_TMP_DIR);
tmpPlanFilesDir = configuredPlanTmpPath != null
? configuredPlanTmpPath
- : System.getProperty("java.io.tmpdir") + File.separator + "flink_plan_" + r.nextInt();
+ : System.getProperty("java.io.tmpdir") + File.separator + "flink_plan_" + UUID.randomUUID();
- tmpDistributedDir = globalConfig.getString(PythonOptions.DC_TMP_DIR);
+ tmpDistributedDir = new Path(globalConfig.getString(PythonOptions.DC_TMP_DIR));
String flinkRootDir = System.getenv("FLINK_ROOT_DIR");
pythonLibraryPath = flinkRootDir != null
@@ -131,121 +126,95 @@ public class PythonPlanBinder {
void runPlan(String[] args) throws Exception {
int split = 0;
for (int x = 0; x < args.length; x++) {
- if (args[x].compareTo("-") == 0) {
+ if (args[x].equals("-")) {
split = x;
+ break;
}
}
try {
- String tmpPath = tmpPlanFilesDir;
- prepareFiles(tmpPath, Arrays.copyOfRange(args, 0, split == 0 ? args.length : split));
- startPython(tmpPath, Arrays.copyOfRange(args, split == 0 ? args.length : split + 1, args.length));
+ String planFile = args[0];
+ String[] filesToCopy = Arrays.copyOfRange(args, 1, split == 0 ? args.length : split);
+ String[] planArgumentsArray = Arrays.copyOfRange(args, split == 0 ? args.length : split + 1, args.length);
+
+ StringBuilder planArgumentsBuilder = new StringBuilder();
+ for (String arg : planArgumentsArray) {
+ planArgumentsBuilder.append(" ").append(arg);
+ }
+ String planArguments = planArgumentsBuilder.toString();
+
+ operatorConfig.setString(PLAN_ARGUMENTS_KEY, planArguments);
+
+ // copy flink library, plan file and additional files to temporary location
+ Path tmpPlanFilesPath = new Path(tmpPlanFilesDir);
+ deleteIfExists(tmpPlanFilesPath);
+ FileCache.copy(new Path(pythonLibraryPath), tmpPlanFilesPath, false);
+ copyFile(new Path(planFile), tmpPlanFilesPath, FLINK_PYTHON_PLAN_NAME);
+ for (String file : filesToCopy) {
+ Path source = new Path(file);
+ copyFile(source, tmpPlanFilesPath, source.getName());
+ }
+
+ // start python process
+ streamer = new PythonPlanStreamer(operatorConfig);
+ streamer.open(tmpPlanFilesDir, planArguments);
// Python process should terminate itself when all jobs have been run
while (streamer.preparePlanMode()) {
- receivePlan();
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ receivePlan(env);
+
+ // upload files to remote FS and register on Distributed Cache
+ deleteIfExists(tmpDistributedDir);
+ FileCache.copy(tmpPlanFilesPath, tmpDistributedDir, true);
+ env.registerCachedFile(tmpDistributedDir.toUri().toString(), FLINK_PYTHON_DC_ID);
- distributeFiles(tmpPath, env);
JobExecutionResult jer = env.execute();
- sendResult(jer);
+ long runtime = jer.getNetRuntime();
+ streamer.sendRecord(runtime);
streamer.finishPlanMode();
+ sets.reset();
+ }
+ } finally {
+ try {
+ // clean up created files
+ FileSystem distributedFS = tmpDistributedDir.getFileSystem();
+ distributedFS.delete(tmpDistributedDir, true);
+
+ FileSystem local = FileSystem.getLocalFileSystem();
+ local.delete(new Path(tmpPlanFilesDir), true);
+ } catch (IOException ioe) {
+ LOG.error("PythonAPI file cleanup failed. {}", ioe.getMessage());
+ } finally {
+ if (streamer != null) {
+ streamer.close();
+ }
}
-
- clearPath(tmpPath);
- close();
- } catch (Exception e) {
- close();
- throw e;
}
}
//=====Setup========================================================================================================
- /**
- * Copies all files to a common directory {@link PythonOptions#PLAN_TMP_DIR}). This allows us to distribute it as
- * one big package which resolves PYTHONPATH issues.
- *
- * @param filePaths
- * @throws IOException
- * @throws URISyntaxException
- */
- private void prepareFiles(String tempFilePath, String... filePaths) throws IOException, URISyntaxException {
- //Flink python package
- clearPath(tempFilePath);
- FileCache.copy(new Path(pythonLibraryPath), new Path(tmpPlanFilesDir), false);
-
- //plan file
- copyFile(filePaths[0], tempFilePath, FLINK_PYTHON_PLAN_NAME);
-
- //additional files/folders
- for (int x = 1; x < filePaths.length; x++) {
- copyFile(filePaths[x], tempFilePath, null);
+ private static void deleteIfExists(Path path) throws IOException {
+ FileSystem fs = path.getFileSystem();
+ if (fs.exists(path)) {
+ fs.delete(path, true);
}
}
- private static void clearPath(String path) throws IOException {
- FileSystem fs = FileSystem.get(new Path(path).toUri());
- if (fs.exists(new Path(path))) {
- fs.delete(new Path(path), true);
- }
- }
-
- private static void copyFile(String path, String target, String name) throws IOException, URISyntaxException {
- if (path.endsWith("/")) {
- path = path.substring(0, path.length() - 1);
- }
- String identifier = name == null ? path.substring(path.lastIndexOf("/")) : name;
- String tmpFilePath = target + "/" + identifier;
- clearPath(tmpFilePath);
- Path p = new Path(path);
- FileCache.copy(p.makeQualified(FileSystem.get(p.toUri())), new Path(tmpFilePath), true);
- }
-
- private void distributeFiles(String tmpPath, ExecutionEnvironment env) throws IOException {
- clearPath(tmpDistributedDir);
- FileCache.copy(new Path(tmpPath), new Path(tmpDistributedDir), true);
- env.registerCachedFile(new Path(tmpDistributedDir).toUri().toString(), FLINK_PYTHON_DC_ID);
- }
-
- private void startPython(String tempPath, String[] args) throws IOException {
- StringBuilder arguments = new StringBuilder();
- for (String arg : args) {
- arguments.append(" ").append(arg);
- }
-
- operatorConfig.setString(PLAN_ARGUMENTS_KEY, arguments.toString());
-
- streamer = new PythonPlanStreamer(operatorConfig);
- streamer.open(tempPath, arguments.toString());
- }
-
- private void sendResult(JobExecutionResult jer) throws IOException {
- long runtime = jer.getNetRuntime();
- streamer.sendRecord(runtime);
- }
-
- private void close() {
- try { //prevent throwing exception so that previous exceptions aren't hidden.
- FileSystem hdfs = new Path(tmpDistributedDir).getFileSystem();
- hdfs.delete(new Path(tmpDistributedDir), true);
-
- FileSystem local = FileSystem.getLocalFileSystem();
- local.delete(new Path(tmpPlanFilesDir), true);
- streamer.close();
- } catch (NullPointerException ignored) {
- } catch (IOException ioe) {
- LOG.error("PythonAPI file cleanup failed. {}", ioe.getMessage());
- }
+ private static void copyFile(Path source, Path targetDirectory, String name) throws IOException {
+ Path targetFilePath = new Path(targetDirectory, name);
+ deleteIfExists(targetFilePath);
+ FileCache.copy(source, targetFilePath, true);
}
//====Plan==========================================================================================================
- private void receivePlan() throws IOException {
- env = ExecutionEnvironment.getExecutionEnvironment();
+ private void receivePlan(ExecutionEnvironment env) throws IOException {
//IDs used in HashMap of sets are only unique for each environment
- sets.reset();
- receiveParameters();
- receiveOperations();
+ receiveParameters(env);
+ receiveOperations(env);
}
//====Environment===================================================================================================
@@ -260,7 +229,7 @@ public class PythonPlanBinder {
ID
}
- private void receiveParameters() throws IOException {
+ private void receiveParameters(ExecutionEnvironment env) throws IOException {
for (int x = 0; x < 4; x++) {
Tuple value = (Tuple) streamer.getRecord(true);
switch (Parameters.valueOf(((String) value.getField(0)).toUpperCase())) {
@@ -271,7 +240,7 @@ public class PythonPlanBinder {
case MODE:
if (value.<Boolean>getField(1)) {
LOG.info("Local execution specified, using default for {}.", PythonOptions.DC_TMP_DIR);
- tmpDistributedDir = PythonOptions.DC_TMP_DIR.defaultValue();
+ tmpDistributedDir = new Path(PythonOptions.DC_TMP_DIR.defaultValue());
}
break;
case RETRY:
@@ -301,23 +270,23 @@ public class PythonPlanBinder {
COGROUP, CROSS, CROSS_H, CROSS_T, FILTER, FLATMAP, GROUPREDUCE, JOIN, JOIN_H, JOIN_T, MAP, REDUCE, MAPPARTITION
}
- private void receiveOperations() throws IOException {
+ private void receiveOperations(ExecutionEnvironment env) throws IOException {
Integer operationCount = (Integer) streamer.getRecord(true);
for (int x = 0; x < operationCount; x++) {
PythonOperationInfo info = new PythonOperationInfo(streamer, currentEnvironmentID);
Operation op = Operation.valueOf(info.identifier.toUpperCase());
switch (op) {
case SOURCE_CSV:
- createCsvSource(info);
+ createCsvSource(env, info);
break;
case SOURCE_TEXT:
- createTextSource(info);
+ createTextSource(env, info);
break;
case SOURCE_VALUE:
- createValueSource(info);
+ createValueSource(env, info);
break;
case SOURCE_SEQ:
- createSequenceSource(info);
+ createSequenceSource(env, info);
break;
case SINK_CSV:
createCsvSink(info);
@@ -395,12 +364,8 @@ public class PythonPlanBinder {
}
}
- private int getParallelism(PythonOperationInfo info) {
- return info.parallelism == -1 ? env.getParallelism() : info.parallelism;
- }
-
@SuppressWarnings("unchecked")
- private <T extends Tuple> void createCsvSource(PythonOperationInfo info) {
+ private <T extends Tuple> void createCsvSource(ExecutionEnvironment env, PythonOperationInfo info) {
if (!(info.types instanceof TupleTypeInfo)) {
throw new RuntimeException("The output type of a csv source has to be a tuple. The derived type is " + info);
}
@@ -408,41 +373,41 @@ public class PythonPlanBinder {
String lineD = info.lineDelimiter;
String fieldD = info.fieldDelimiter;
TupleTypeInfo<T> types = (TupleTypeInfo<T>) info.types;
- sets.add(info.setID, env.createInput(new TupleCsvInputFormat<>(path, lineD, fieldD, types), types).setParallelism(getParallelism(info)).name("CsvSource")
- .map(new SerializerMap<T>()).setParallelism(getParallelism(info)).name("CsvSourcePostStep"));
+ sets.add(info.setID, env.createInput(new TupleCsvInputFormat<>(path, lineD, fieldD, types), types).setParallelism(info.parallelism).name("CsvSource")
+ .map(new SerializerMap<T>()).setParallelism(info.parallelism).name("CsvSourcePostStep"));
}
- private void createTextSource(PythonOperationInfo info) {
- sets.add(info.setID, env.readTextFile(info.path).setParallelism(getParallelism(info)).name("TextSource")
- .map(new SerializerMap<String>()).setParallelism(getParallelism(info)).name("TextSourcePostStep"));
+ private void createTextSource(ExecutionEnvironment env, PythonOperationInfo info) {
+ sets.add(info.setID, env.readTextFile(info.path).setParallelism(info.parallelism).name("TextSource")
+ .map(new SerializerMap<String>()).setParallelism(info.parallelism).name("TextSourcePostStep"));
}
- private void createValueSource(PythonOperationInfo info) {
- sets.add(info.setID, env.fromElements(info.values).setParallelism(getParallelism(info)).name("ValueSource")
- .map(new SerializerMap<>()).setParallelism(getParallelism(info)).name("ValueSourcePostStep"));
+ private void createValueSource(ExecutionEnvironment env, PythonOperationInfo info) {
+ sets.add(info.setID, env.fromElements(info.values).setParallelism(info.parallelism).name("ValueSource")
+ .map(new SerializerMap<>()).setParallelism(info.parallelism).name("ValueSourcePostStep"));
}
- private void createSequenceSource(PythonOperationInfo info) {
- sets.add(info.setID, env.generateSequence(info.frm, info.to).setParallelism(getParallelism(info)).name("SequenceSource")
- .map(new SerializerMap<Long>()).setParallelism(getParallelism(info)).name("SequenceSourcePostStep"));
+ private void createSequenceSource(ExecutionEnvironment env, PythonOperationInfo info) {
+ sets.add(info.setID, env.generateSequence(info.frm, info.to).setParallelism(info.parallelism).name("SequenceSource")
+ .map(new SerializerMap<Long>()).setParallelism(info.parallelism).name("SequenceSourcePostStep"));
}
private void createCsvSink(PythonOperationInfo info) {
DataSet<byte[]> parent = sets.getDataSet(info.parentID);
- parent.map(new StringTupleDeserializerMap()).setParallelism(getParallelism(info)).name("CsvSinkPreStep")
- .writeAsCsv(info.path, info.lineDelimiter, info.fieldDelimiter, info.writeMode).setParallelism(getParallelism(info)).name("CsvSink");
+ parent.map(new StringTupleDeserializerMap()).setParallelism(info.parallelism).name("CsvSinkPreStep")
+ .writeAsCsv(info.path, info.lineDelimiter, info.fieldDelimiter, info.writeMode).setParallelism(info.parallelism).name("CsvSink");
}
private void createTextSink(PythonOperationInfo info) {
DataSet<byte[]> parent = sets.getDataSet(info.parentID);
- parent.map(new StringDeserializerMap()).setParallelism(getParallelism(info))
- .writeAsText(info.path, info.writeMode).setParallelism(getParallelism(info)).name("TextSink");
+ parent.map(new StringDeserializerMap()).setParallelism(info.parallelism)
+ .writeAsText(info.path, info.writeMode).setParallelism(info.parallelism).name("TextSink");
}
private void createPrintSink(PythonOperationInfo info) {
DataSet<byte[]> parent = sets.getDataSet(info.parentID);
- parent.map(new StringDeserializerMap()).setParallelism(getParallelism(info)).name("PrintSinkPreStep")
- .output(new PrintingOutputFormat<String>(info.toError)).setParallelism(getParallelism(info));
+ parent.map(new StringDeserializerMap()).setParallelism(info.parallelism).name("PrintSinkPreStep")
+ .output(new PrintingOutputFormat<String>(info.toError)).setParallelism(info.parallelism);
}
private void createBroadcastVariable(PythonOperationInfo info) {
@@ -466,8 +431,8 @@ public class PythonPlanBinder {
private <K extends Tuple> void createDistinctOperation(PythonOperationInfo info) {
DataSet<Tuple2<K, byte[]>> op = sets.getDataSet(info.parentID);
DataSet<byte[]> result = op
- .distinct(info.keys).setParallelism(getParallelism(info)).name("Distinct")
- .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("DistinctPostStep");
+ .distinct(info.keys).setParallelism(info.parallelism).name("Distinct")
+ .map(new KeyDiscarder<K>()).setParallelism(info.parallelism).name("DistinctPostStep");
sets.add(info.setID, result);
}
@@ -475,17 +440,17 @@ public class PythonPlanBinder {
if (sets.isDataSet(info.parentID)) {
DataSet<byte[]> op = sets.getDataSet(info.parentID);
sets.add(info.setID, op
- .first(info.count).setParallelism(getParallelism(info)).name("First"));
+ .first(info.count).setParallelism(info.parallelism).name("First"));
} else if (sets.isUnsortedGrouping(info.parentID)) {
UnsortedGrouping<Tuple2<K, byte[]>> op = sets.getUnsortedGrouping(info.parentID);
sets.add(info.setID, op
- .first(info.count).setParallelism(getParallelism(info)).name("First")
- .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("FirstPostStep"));
+ .first(info.count).setParallelism(info.parallelism).name("First")
+ .map(new KeyDiscarder<K>()).setParallelism(info.parallelism).name("FirstPostStep"));
} else if (sets.isSortedGrouping(info.parentID)) {
SortedGrouping<Tuple2<K, byte[]>> op = sets.getSortedGrouping(info.parentID);
sets.add(info.setID, op
- .first(info.count).setParallelism(getParallelism(info)).name("First")
- .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("FirstPostStep"));
+ .first(info.count).setParallelism(info.parallelism).name("First")
+ .map(new KeyDiscarder<K>()).setParallelism(info.parallelism).name("FirstPostStep"));
}
}
@@ -497,14 +462,14 @@ public class PythonPlanBinder {
private <K extends Tuple> void createHashPartitionOperation(PythonOperationInfo info) {
DataSet<Tuple2<K, byte[]>> op1 = sets.getDataSet(info.parentID);
DataSet<byte[]> result = op1
- .partitionByHash(info.keys).setParallelism(getParallelism(info))
- .map(new KeyDiscarder<K>()).setParallelism(getParallelism(info)).name("HashPartitionPostStep");
+ .partitionByHash(info.keys).setParallelism(info.parallelism)
+ .map(new KeyDiscarder<K>()).setParallelism(info.parallelism).name("HashPartitionPostStep");
sets.add(info.setID, result);
}
private void createRebalanceOperation(PythonOperationInfo info) {
DataSet<?> op = sets.getDataSet(info.parentID);
- sets.add(info.setID, op.rebalance().setParallelism(getParallelism(info)).name("Rebalance"));
+ sets.add(info.setID, op.rebalance().setParallelism(info.parallelism).name("Rebalance"));
}
private void createSortOperation(PythonOperationInfo info) {
@@ -520,7 +485,7 @@ public class PythonPlanBinder {
private <IN> void createUnionOperation(PythonOperationInfo info) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
DataSet<IN> op2 = sets.getDataSet(info.otherID);
- sets.add(info.setID, op1.union(op2).setParallelism(getParallelism(info)).name("Union"));
+ sets.add(info.setID, op1.union(op2).setParallelism(info.parallelism).name("Union"));
}
private <IN1, IN2, OUT> void createCoGroupOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
@@ -529,7 +494,7 @@ public class PythonPlanBinder {
Keys.ExpressionKeys<IN1> key1 = new Keys.ExpressionKeys<>(info.keys1, op1.getType());
Keys.ExpressionKeys<IN2> key2 = new Keys.ExpressionKeys<>(info.keys2, op2.getType());
PythonCoGroup<IN1, IN2, OUT> pcg = new PythonCoGroup<>(operatorConfig, info.envID, info.setID, type);
- sets.add(info.setID, new CoGroupRawOperator<>(op1, op2, key1, key2, pcg, type, info.name).setParallelism(getParallelism(info)));
+ sets.add(info.setID, new CoGroupRawOperator<>(op1, op2, key1, key2, pcg, type, info.name).setParallelism(info.parallelism));
}
private <IN1, IN2, OUT> void createCrossOperation(DatasizeHint mode, PythonOperationInfo info, TypeInformation<OUT> type) {
@@ -551,11 +516,11 @@ public class PythonPlanBinder {
throw new IllegalArgumentException("Invalid Cross mode specified: " + mode);
}
- defaultResult.setParallelism(getParallelism(info));
+ defaultResult.setParallelism(info.parallelism);
if (info.usesUDF) {
sets.add(info.setID, defaultResult
.mapPartition(new PythonMapPartition<Tuple2<IN1, IN2>, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name));
+ .setParallelism(info.parallelism).name(info.name));
} else {
sets.add(info.setID, defaultResult.name("DefaultCross"));
}
@@ -565,14 +530,14 @@ public class PythonPlanBinder {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
sets.add(info.setID, op1
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name));
+ .setParallelism(info.parallelism).name(info.name));
}
private <IN, OUT> void createFlatMapOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
sets.add(info.setID, op1
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name));
+ .setParallelism(info.parallelism).name(info.name));
}
private void createGroupReduceOperation(PythonOperationInfo info) {
@@ -587,23 +552,23 @@ public class PythonPlanBinder {
private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(DataSet<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
- .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).name("PythonGroupReducePreStep").setParallelism(getParallelism(info))
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).name("PythonGroupReducePreStep").setParallelism(info.parallelism)
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name);
+ .setParallelism(info.parallelism).name(info.name);
}
private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(UnsortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
- .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(info.parallelism).name("PythonGroupReducePreStep")
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name);
+ .setParallelism(info.parallelism).name(info.name);
}
private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(SortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
- .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(info.parallelism).name("PythonGroupReducePreStep")
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name);
+ .setParallelism(info.parallelism).name(info.name);
}
private <IN1, IN2, OUT> void createJoinOperation(DatasizeHint mode, PythonOperationInfo info, TypeInformation<OUT> type) {
@@ -611,11 +576,11 @@ public class PythonPlanBinder {
DataSet<IN2> op2 = sets.getDataSet(info.otherID);
if (info.usesUDF) {
- sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))
+ sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, info.parallelism)
.mapPartition(new PythonMapPartition<Tuple2<byte[], byte[]>, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name));
+ .setParallelism(info.parallelism).name(info.name));
} else {
- sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)));
+ sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, info.parallelism));
}
}
@@ -642,14 +607,14 @@ public class PythonPlanBinder {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
sets.add(info.setID, op1
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name));
+ .setParallelism(info.parallelism).name(info.name));
}
private <IN, OUT> void createMapPartitionOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
sets.add(info.setID, op1
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name));
+ .setParallelism(info.parallelism).name(info.name));
}
private void createReduceOperation(PythonOperationInfo info) {
@@ -664,15 +629,15 @@ public class PythonPlanBinder {
private <IN, OUT> DataSet<OUT> applyReduceOperation(DataSet<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
- .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(info.parallelism).name("PythonReducePreStep")
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name);
+ .setParallelism(info.parallelism).name(info.name);
}
private <IN, OUT> DataSet<OUT> applyReduceOperation(UnsortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
- .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
+ .reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(info.parallelism).name("PythonReducePreStep")
.mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
- .setParallelism(getParallelism(info)).name(info.name);
+ .setParallelism(info.parallelism).name(info.name);
}
}
[4/5] flink git commit: [FLINK-6229] [py] Rework configuration of
PythonPlanBinder/Operators
Posted by ch...@apache.org.
[FLINK-6229] [py] Rework configuration of PythonPlanBinder/Operators
- unify python2/python3 configuration
- explicitly pass on a configuration to each operator
- port all configuration options to ConfigOptions
- [FLINK-5516] Make all paths explicitly configurable
- [FLINK-6230] Make mmap size configurable
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bdcebfda
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bdcebfda
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bdcebfda
Branch: refs/heads/master
Commit: bdcebfda06846a1e21bb6a4678909d503ebc6333
Parents: 940d16c
Author: zentol <ch...@apache.org>
Authored: Fri Mar 31 11:55:18 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Thu Apr 6 10:57:11 2017 +0200
----------------------------------------------------------------------
docs/dev/batch/python.md | 10 +-
flink-dist/src/main/flink-bin/bin/pyflink.bat | 25 ++++
flink-dist/src/main/flink-bin/bin/pyflink.sh | 25 ++++
flink-dist/src/main/flink-bin/bin/pyflink2.bat | 25 ----
flink-dist/src/main/flink-bin/bin/pyflink2.sh | 25 ----
flink-dist/src/main/flink-bin/bin/pyflink3.bat | 25 ----
flink-dist/src/main/flink-bin/bin/pyflink3.sh | 26 ----
.../apache/flink/python/api/PythonOptions.java | 74 ++++++++++
.../flink/python/api/PythonPlanBinder.java | 143 +++++++++++--------
.../python/api/functions/PythonCoGroup.java | 4 +-
.../api/functions/PythonMapPartition.java | 4 +-
.../streaming/data/PythonDualInputSender.java | 6 +
.../streaming/data/PythonDualInputStreamer.java | 5 +-
.../api/streaming/data/PythonReceiver.java | 29 ++--
.../python/api/streaming/data/PythonSender.java | 36 ++---
.../streaming/data/PythonSingleInputSender.java | 6 +
.../data/PythonSingleInputStreamer.java | 9 +-
.../api/streaming/data/PythonStreamer.java | 39 ++---
.../api/streaming/plan/PythonPlanStreamer.java | 12 +-
.../python/api/flink/connection/Connection.py | 21 ++-
.../api/flink/functions/CoGroupFunction.py | 4 +-
.../python/api/flink/functions/Function.py | 4 +-
.../api/flink/functions/GroupReduceFunction.py | 4 +-
.../api/flink/functions/ReduceFunction.py | 4 +-
.../flink/python/api/flink/plan/Environment.py | 5 +-
.../flink/python/api/PythonPlanBinderTest.java | 12 +-
26 files changed, 322 insertions(+), 260 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/docs/dev/batch/python.md
----------------------------------------------------------------------
diff --git a/docs/dev/batch/python.md b/docs/dev/batch/python.md
index 09a4fa8..c4c2671 100644
--- a/docs/dev/batch/python.md
+++ b/docs/dev/batch/python.md
@@ -149,8 +149,7 @@ Apart from setting up Flink, no additional work is required. The python package
The Python API was tested on Linux/Windows systems that have Python 2.7 or 3.4 installed.
-By default Flink will start python processes by calling "python" or "python3", depending on which start-script
-was used. By setting the "python.binary.python[2/3]" key in the flink-conf.yaml you can modify this behaviour to use a binary of your choice.
+By default Flink will start python processes by calling "python". By setting the "python.binary.path" key in the flink-conf.yaml you can modify this behaviour to use a binary of your choice.
{% top %}
@@ -624,12 +623,11 @@ Executing Plans
---------------
To run the plan with Flink, go to your Flink distribution, and run the pyflink.sh script from the /bin folder.
-use pyflink2.sh for python 2.7, and pyflink3.sh for python 3.4. The script containing the plan has to be passed
-as the first argument, followed by a number of additional python packages, and finally, separated by - additional
-arguments that will be fed to the script.
+The script containing the plan has to be passed as the first argument, followed by a number of additional python
+packages, and finally, separated by - additional arguments that will be fed to the script.
{% highlight python %}
-./bin/pyflink<2/3>.sh <Script>[ <pathToPackage1>[ <pathToPackageX]][ - <param1>[ <paramX>]]
+./bin/pyflink.sh <Script>[ <pathToPackage1>[ <pathToPackageX]][ - <param1>[ <paramX>]]
{% endhighlight %}
{% top %}
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-dist/src/main/flink-bin/bin/pyflink.bat
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink.bat b/flink-dist/src/main/flink-bin/bin/pyflink.bat
new file mode 100644
index 0000000..da180f3
--- /dev/null
+++ b/flink-dist/src/main/flink-bin/bin/pyflink.bat
@@ -0,0 +1,25 @@
+::###############################################################################
+:: 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.
+::###############################################################################
+
+@echo off
+setlocal EnableDelayedExpansion
+
+SET bin=%~dp0
+SET FLINK_ROOT_DIR=%bin%..
+
+"%FLINK_ROOT_DIR%\bin\flink" run -v "%FLINK_ROOT_DIR%"\lib\flink-python*.jar %*
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-dist/src/main/flink-bin/bin/pyflink.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink.sh b/flink-dist/src/main/flink-bin/bin/pyflink.sh
new file mode 100644
index 0000000..37679d3
--- /dev/null
+++ b/flink-dist/src/main/flink-bin/bin/pyflink.sh
@@ -0,0 +1,25 @@
+#!/bin/bash
+################################################################################
+# 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.
+################################################################################
+
+bin=`dirname "$0"`
+bin=`cd "$bin"; pwd`
+
+. "$bin"/config.sh
+
+"$FLINK_BIN_DIR"/flink run -v "$FLINK_ROOT_DIR"/lib/flink-python*.jar "$@"
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-dist/src/main/flink-bin/bin/pyflink2.bat
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink2.bat b/flink-dist/src/main/flink-bin/bin/pyflink2.bat
deleted file mode 100644
index 1122ed4..0000000
--- a/flink-dist/src/main/flink-bin/bin/pyflink2.bat
+++ /dev/null
@@ -1,25 +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.
-::###############################################################################
-
-@echo off
-setlocal EnableDelayedExpansion
-
-SET bin=%~dp0
-SET FLINK_ROOT_DIR=%bin%..
-
-"%FLINK_ROOT_DIR%\bin\flink" run -v "%FLINK_ROOT_DIR%"\lib\flink-python*.jar 2 %*
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-dist/src/main/flink-bin/bin/pyflink2.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink2.sh b/flink-dist/src/main/flink-bin/bin/pyflink2.sh
deleted file mode 100644
index 8a326e9..0000000
--- a/flink-dist/src/main/flink-bin/bin/pyflink2.sh
+++ /dev/null
@@ -1,25 +0,0 @@
-#!/bin/bash
-################################################################################
-# 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.
-################################################################################
-
-bin=`dirname "$0"`
-bin=`cd "$bin"; pwd`
-
-. "$bin"/config.sh
-
-"$FLINK_BIN_DIR"/flink run -v "$FLINK_ROOT_DIR"/lib/flink-python*.jar "2" "$@"
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-dist/src/main/flink-bin/bin/pyflink3.bat
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink3.bat b/flink-dist/src/main/flink-bin/bin/pyflink3.bat
deleted file mode 100644
index 0294e37..0000000
--- a/flink-dist/src/main/flink-bin/bin/pyflink3.bat
+++ /dev/null
@@ -1,25 +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.
-::###############################################################################
-
-@echo off
-setlocal EnableDelayedExpansion
-
-SET bin=%~dp0
-SET FLINK_ROOT_DIR=%bin%..
-
-"%FLINK_ROOT_DIR%\bin\flink" run -v "%FLINK_ROOT_DIR%"\lib\flink-python*.jar 3 %*
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-dist/src/main/flink-bin/bin/pyflink3.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink3.sh b/flink-dist/src/main/flink-bin/bin/pyflink3.sh
deleted file mode 100644
index d2d1991..0000000
--- a/flink-dist/src/main/flink-bin/bin/pyflink3.sh
+++ /dev/null
@@ -1,26 +0,0 @@
-#!/bin/bash
-################################################################################
-# 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.
-################################################################################
-
-bin=`dirname "$0"`
-bin=`cd "$bin"; pwd`
-
-. "$bin"/config.sh
-
-
-"$FLINK_BIN_DIR"/flink run -v "$FLINK_ROOT_DIR"/lib/flink-python*.jar "3" "$@"
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOptions.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOptions.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOptions.java
new file mode 100644
index 0000000..de053a0
--- /dev/null
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOptions.java
@@ -0,0 +1,74 @@
+/*
+ * 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.python.api;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.io.File;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * Configuration options for the Python API.
+ */
+public class PythonOptions {
+
+ /**
+ * The config parameter defining the path to the python binary to use.
+ */
+ public static final ConfigOption<String> PYTHON_BINARY_PATH =
+ key("python.binary.path")
+ .defaultValue("python")
+ .withDeprecatedKeys("python.binary.python2", "python.binary.python3");
+
+ /**
+ * The config parameter defining the size of the memory-mapped files, in kb.
+ * This value must be large enough to ensure that the largest serialized record can be written completely into
+ * the file.
+ *
+ * Every task will allocate 2 memory-files, each with this size.
+ */
+ public static final ConfigOption<Long> MMAP_FILE_SIZE =
+ key("python.mmap.size.kb")
+ .defaultValue(4L);
+
+ /**
+ * The config parameter defining where temporary plan-related files are stored on the client.
+ */
+ public static final ConfigOption<String> PLAN_TMP_DIR =
+ key("python.plan.tmp.dir")
+ .noDefaultValue();
+
+ /**
+ * The config parameter defining where the memory-mapped files will be created.
+ */
+ public static final ConfigOption<String> DATA_TMP_DIR =
+ key("python.mmap.tmp.dir")
+ .noDefaultValue();
+
+ /**
+ * The config parameter defining where the flink python library and user supplied files will be uploaded to before
+ * registering them with the Distributed Cache. This directory must be accessible from all worker nodes.
+ */
+ public static final ConfigOption<String> DC_TMP_DIR =
+ key("python.dc.tmp.dir")
+ .defaultValue(System.getProperty("java.io.tmpdir") + File.separator + "flink_dc");
+
+ private PythonOptions() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
index 733a6fb..b6181b4 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonPlanBinder.java
@@ -18,7 +18,6 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.LocalEnvironment;
import org.apache.flink.api.java.io.PrintingOutputFormat;
import org.apache.flink.api.java.io.TupleCsvInputFormat;
import org.apache.flink.api.java.operators.CoGroupRawOperator;
@@ -50,7 +49,6 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
-import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.Random;
@@ -65,42 +63,30 @@ import static org.apache.flink.python.api.PythonOperationInfo.DatasizeHint.TINY;
public class PythonPlanBinder {
static final Logger LOG = LoggerFactory.getLogger(PythonPlanBinder.class);
- public static final String ARGUMENT_PYTHON_2 = "2";
- public static final String ARGUMENT_PYTHON_3 = "3";
-
public static final String FLINK_PYTHON_DC_ID = "flink";
public static final String FLINK_PYTHON_PLAN_NAME = File.separator + "plan.py";
- public static final String FLINK_PYTHON2_BINARY_KEY = "python.binary.python2";
- public static final String FLINK_PYTHON3_BINARY_KEY = "python.binary.python3";
public static final String PLANBINDER_CONFIG_BCVAR_COUNT = "PLANBINDER_BCVAR_COUNT";
public static final String PLANBINDER_CONFIG_BCVAR_NAME_PREFIX = "PLANBINDER_BCVAR_";
- public static String FLINK_PYTHON2_BINARY_PATH =
- GlobalConfiguration.loadConfiguration().getString(FLINK_PYTHON2_BINARY_KEY, "python");
- public static String FLINK_PYTHON3_BINARY_PATH =
- GlobalConfiguration.loadConfiguration().getString(FLINK_PYTHON3_BINARY_KEY, "python3");
private static final Random r = new Random();
- public static final String FLINK_PYTHON_FILE_PATH = System.getProperty("java.io.tmpdir") + File.separator + "flink_plan";
+ public static final String PLAN_ARGUMENTS_KEY = "python.plan.arguments";
+
private static final String FLINK_PYTHON_REL_LOCAL_PATH = File.separator + "resources" + File.separator + "python";
- private static final String FLINK_DIR = System.getenv("FLINK_ROOT_DIR");
- private static String FULL_PATH;
- public static StringBuilder arguments = new StringBuilder();
+ private final Configuration operatorConfig;
- public static boolean usePython3 = false;
+ private final String pythonLibraryPath;
- private static String FLINK_HDFS_PATH = "hdfs:/tmp";
- public static final String FLINK_TMP_DATA_DIR = System.getProperty("java.io.tmpdir") + File.separator + "flink_data";
+ private final String tmpPlanFilesDir;
+ private String tmpDistributedDir;
private final SetCache sets = new SetCache();
public ExecutionEnvironment env;
private int currentEnvironmentID = 0;
private PythonPlanStreamer streamer;
- public static final int MAPPED_FILE_SIZE = 1024 * 1024 * 64;
-
/**
* Entry point for the execution of a python plan.
*
@@ -112,23 +98,37 @@ public class PythonPlanBinder {
System.out.println("Usage: ./bin/pyflink<2/3>.[sh/bat] <pathToScript>[ <pathToPackage1>[ <pathToPackageX]][ - <parameter1>[ <parameterX>]]");
return;
}
- usePython3 = args[0].equals(ARGUMENT_PYTHON_3);
- PythonPlanBinder binder = new PythonPlanBinder();
- binder.runPlan(Arrays.copyOfRange(args, 1, args.length));
- }
- public PythonPlanBinder() {
- Configuration conf = GlobalConfiguration.loadConfiguration();
- FLINK_PYTHON2_BINARY_PATH = conf.getString(FLINK_PYTHON2_BINARY_KEY, "python");
- FLINK_PYTHON3_BINARY_PATH = conf.getString(FLINK_PYTHON3_BINARY_KEY, "python3");
- FULL_PATH = FLINK_DIR != null
+ Configuration globalConfig = GlobalConfiguration.loadConfiguration();
+ PythonPlanBinder binder = new PythonPlanBinder(globalConfig);
+ binder.runPlan(args);
+ }
+
+ public PythonPlanBinder(Configuration globalConfig) {
+ String configuredPlanTmpPath = globalConfig.getString(PythonOptions.PLAN_TMP_DIR);
+ tmpPlanFilesDir = configuredPlanTmpPath != null
+ ? configuredPlanTmpPath
+ : System.getProperty("java.io.tmpdir") + File.separator + "flink_plan_" + r.nextInt();
+
+ tmpDistributedDir = globalConfig.getString(PythonOptions.DC_TMP_DIR);
+
+ String flinkRootDir = System.getenv("FLINK_ROOT_DIR");
+ pythonLibraryPath = flinkRootDir != null
//command-line
- ? FLINK_DIR + FLINK_PYTHON_REL_LOCAL_PATH
+ ? flinkRootDir + FLINK_PYTHON_REL_LOCAL_PATH
//testing
- : new Path(FileSystem.getLocalFileSystem().getWorkingDirectory(), "src/main/python/org/apache/flink/python/api").toString();
+ : new File(System.getProperty("user.dir"), "src/main/python/org/apache/flink/python/api").getAbsolutePath();
+
+ operatorConfig = new Configuration();
+ operatorConfig.setString(PythonOptions.PYTHON_BINARY_PATH, globalConfig.getString(PythonOptions.PYTHON_BINARY_PATH));
+ String configuredTmpDataDir = globalConfig.getString(PythonOptions.DATA_TMP_DIR);
+ if (configuredTmpDataDir != null) {
+ operatorConfig.setString(PythonOptions.DATA_TMP_DIR, configuredTmpDataDir);
+ }
+ operatorConfig.setLong(PythonOptions.MMAP_FILE_SIZE, globalConfig.getLong(PythonOptions.MMAP_FILE_SIZE));
}
- private void runPlan(String[] args) throws Exception {
+ void runPlan(String[] args) throws Exception {
int split = 0;
for (int x = 0; x < args.length; x++) {
if (args[x].compareTo("-") == 0) {
@@ -137,7 +137,7 @@ public class PythonPlanBinder {
}
try {
- String tmpPath = FLINK_PYTHON_FILE_PATH + r.nextInt();
+ String tmpPath = tmpPlanFilesDir;
prepareFiles(tmpPath, Arrays.copyOfRange(args, 0, split == 0 ? args.length : split));
startPython(tmpPath, Arrays.copyOfRange(args, split == 0 ? args.length : split + 1, args.length));
@@ -145,10 +145,6 @@ public class PythonPlanBinder {
while (streamer.preparePlanMode()) {
receivePlan();
- if (env instanceof LocalEnvironment) {
- FLINK_HDFS_PATH = "file:" + System.getProperty("java.io.tmpdir") + File.separator + "flink";
- }
-
distributeFiles(tmpPath, env);
JobExecutionResult jer = env.execute();
sendResult(jer);
@@ -167,8 +163,8 @@ public class PythonPlanBinder {
//=====Setup========================================================================================================
/**
- * Copies all files to a common directory (FLINK_PYTHON_FILE_PATH). This allows us to distribute it as one big
- * package, and resolves PYTHONPATH issues.
+ * Copies all files to a common directory {@link PythonOptions#PLAN_TMP_DIR}). This allows us to distribute it as
+ * one big package which resolves PYTHONPATH issues.
*
* @param filePaths
* @throws IOException
@@ -177,7 +173,7 @@ public class PythonPlanBinder {
private void prepareFiles(String tempFilePath, String... filePaths) throws IOException, URISyntaxException {
//Flink python package
clearPath(tempFilePath);
- FileCache.copy(new Path(FULL_PATH), new Path(tempFilePath), false);
+ FileCache.copy(new Path(pythonLibraryPath), new Path(tmpPlanFilesDir), false);
//plan file
copyFile(filePaths[0], tempFilePath, FLINK_PYTHON_PLAN_NAME);
@@ -206,17 +202,21 @@ public class PythonPlanBinder {
FileCache.copy(p.makeQualified(FileSystem.get(p.toUri())), new Path(tmpFilePath), true);
}
- private static void distributeFiles(String tmpPath, ExecutionEnvironment env) throws IOException, URISyntaxException {
- clearPath(FLINK_HDFS_PATH);
- FileCache.copy(new Path(tmpPath), new Path(FLINK_HDFS_PATH), true);
- env.registerCachedFile(FLINK_HDFS_PATH, FLINK_PYTHON_DC_ID);
+ private void distributeFiles(String tmpPath, ExecutionEnvironment env) throws IOException {
+ clearPath(tmpDistributedDir);
+ FileCache.copy(new Path(tmpPath), new Path(tmpDistributedDir), true);
+ env.registerCachedFile(new Path(tmpDistributedDir).toUri().toString(), FLINK_PYTHON_DC_ID);
}
private void startPython(String tempPath, String[] args) throws IOException {
+ StringBuilder arguments = new StringBuilder();
for (String arg : args) {
arguments.append(" ").append(arg);
}
- streamer = new PythonPlanStreamer();
+
+ operatorConfig.setString(PLAN_ARGUMENTS_KEY, arguments.toString());
+
+ streamer = new PythonPlanStreamer(operatorConfig);
streamer.open(tempPath, arguments.toString());
}
@@ -227,17 +227,15 @@ public class PythonPlanBinder {
private void close() {
try { //prevent throwing exception so that previous exceptions aren't hidden.
- FileSystem hdfs = FileSystem.get(new URI(FLINK_HDFS_PATH));
- hdfs.delete(new Path(FLINK_HDFS_PATH), true);
+ FileSystem hdfs = new Path(tmpDistributedDir).getFileSystem();
+ hdfs.delete(new Path(tmpDistributedDir), true);
FileSystem local = FileSystem.getLocalFileSystem();
- local.delete(new Path(FLINK_PYTHON_FILE_PATH), true);
- local.delete(new Path(FLINK_TMP_DATA_DIR), true);
+ local.delete(new Path(tmpPlanFilesDir), true);
streamer.close();
} catch (NullPointerException ignored) {
} catch (IOException ioe) {
LOG.error("PythonAPI file cleanup failed. {}", ioe.getMessage());
- } catch (URISyntaxException use) { // can't occur
}
}
@@ -271,7 +269,10 @@ public class PythonPlanBinder {
env.setParallelism(dop);
break;
case MODE:
- FLINK_HDFS_PATH = value.<Boolean>getField(1) ? "file:/tmp/flink" : "hdfs:/tmp/flink";
+ if (value.<Boolean>getField(1)) {
+ LOG.info("Local execution specified, using default for {}.", PythonOptions.DC_TMP_DIR);
+ tmpDistributedDir = PythonOptions.DC_TMP_DIR.defaultValue();
+ }
break;
case RETRY:
int retry = value.<Integer>getField(1);
@@ -527,7 +528,7 @@ public class PythonPlanBinder {
DataSet<IN2> op2 = sets.getDataSet(info.otherID);
Keys.ExpressionKeys<IN1> key1 = new Keys.ExpressionKeys<>(info.keys1, op1.getType());
Keys.ExpressionKeys<IN2> key2 = new Keys.ExpressionKeys<>(info.keys2, op2.getType());
- PythonCoGroup<IN1, IN2, OUT> pcg = new PythonCoGroup<>(info.envID, info.setID, type);
+ PythonCoGroup<IN1, IN2, OUT> pcg = new PythonCoGroup<>(operatorConfig, info.envID, info.setID, type);
sets.add(info.setID, new CoGroupRawOperator<>(op1, op2, key1, key2, pcg, type, info.name).setParallelism(getParallelism(info)));
}
@@ -552,7 +553,9 @@ public class PythonPlanBinder {
defaultResult.setParallelism(getParallelism(info));
if (info.usesUDF) {
- sets.add(info.setID, defaultResult.mapPartition(new PythonMapPartition<Tuple2<IN1, IN2>, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, defaultResult
+ .mapPartition(new PythonMapPartition<Tuple2<IN1, IN2>, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name));
} else {
sets.add(info.setID, defaultResult.name("DefaultCross"));
}
@@ -560,12 +563,16 @@ public class PythonPlanBinder {
private <IN, OUT> void createFilterOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
- sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, op1
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name));
}
private <IN, OUT> void createFlatMapOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
- sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, op1
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name));
}
private void createGroupReduceOperation(PythonOperationInfo info) {
@@ -581,19 +588,22 @@ public class PythonPlanBinder {
private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(DataSet<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
.reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).name("PythonGroupReducePreStep").setParallelism(getParallelism(info))
- .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name);
}
private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(UnsortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
.reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name);
}
private <IN, OUT> DataSet<OUT> applyGroupReduceOperation(SortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
.reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name);
}
private <IN1, IN2, OUT> void createJoinOperation(DatasizeHint mode, PythonOperationInfo info, TypeInformation<OUT> type) {
@@ -602,7 +612,8 @@ public class PythonPlanBinder {
if (info.usesUDF) {
sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))
- .mapPartition(new PythonMapPartition<Tuple2<byte[], byte[]>, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
+ .mapPartition(new PythonMapPartition<Tuple2<byte[], byte[]>, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name));
} else {
sets.add(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)));
}
@@ -629,12 +640,16 @@ public class PythonPlanBinder {
private <IN, OUT> void createMapOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
- sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, op1
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name));
}
private <IN, OUT> void createMapPartitionOperation(PythonOperationInfo info, TypeInformation<OUT> type) {
DataSet<IN> op1 = sets.getDataSet(info.parentID);
- sets.add(info.setID, op1.mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name));
+ sets.add(info.setID, op1
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name));
}
private void createReduceOperation(PythonOperationInfo info) {
@@ -650,12 +665,14 @@ public class PythonPlanBinder {
private <IN, OUT> DataSet<OUT> applyReduceOperation(DataSet<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
.reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
- .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name);
}
private <IN, OUT> DataSet<OUT> applyReduceOperation(UnsortedGrouping<IN> op1, PythonOperationInfo info, TypeInformation<OUT> type) {
return op1
.reduceGroup(new IdentityGroupReduce<IN>()).setCombinable(false).setParallelism(getParallelism(info)).name("PythonReducePreStep")
- .mapPartition(new PythonMapPartition<IN, OUT>(info.envID, info.setID, type)).setParallelism(getParallelism(info)).name(info.name);
+ .mapPartition(new PythonMapPartition<IN, OUT>(operatorConfig, info.envID, info.setID, type))
+ .setParallelism(getParallelism(info)).name(info.name);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
index ff5a8d4..a5e3e75 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
@@ -35,9 +35,9 @@ public class PythonCoGroup<IN1, IN2, OUT> extends RichCoGroupFunction<IN1, IN2,
private final PythonDualInputStreamer<IN1, IN2, OUT> streamer;
private final transient TypeInformation<OUT> typeInformation;
- public PythonCoGroup(int envID, int setID, TypeInformation<OUT> typeInformation) {
+ public PythonCoGroup(Configuration config, int envID, int setID, TypeInformation<OUT> typeInformation) {
this.typeInformation = typeInformation;
- streamer = new PythonDualInputStreamer<>(this, envID, setID, typeInformation instanceof PrimitiveArrayTypeInfo);
+ streamer = new PythonDualInputStreamer<>(this, config, envID, setID, typeInformation instanceof PrimitiveArrayTypeInfo);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonMapPartition.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonMapPartition.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonMapPartition.java
index 9142581..207ead9 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonMapPartition.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonMapPartition.java
@@ -35,9 +35,9 @@ public class PythonMapPartition<IN, OUT> extends RichMapPartitionFunction<IN, OU
private final PythonSingleInputStreamer<IN, OUT> streamer;
private final transient TypeInformation<OUT> typeInformation;
- public PythonMapPartition(int envId, int setId, TypeInformation<OUT> typeInformation) {
+ public PythonMapPartition(Configuration config, int envId, int setId, TypeInformation<OUT> typeInformation) {
this.typeInformation = typeInformation;
- streamer = new PythonSingleInputStreamer<>(this, envId, setId, typeInformation instanceof PrimitiveArrayTypeInfo);
+ streamer = new PythonSingleInputStreamer<>(this, config, envId, setId, typeInformation instanceof PrimitiveArrayTypeInfo);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputSender.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputSender.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputSender.java
index 3b8e423..a16f522 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputSender.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputSender.java
@@ -17,6 +17,8 @@
*/
package org.apache.flink.python.api.streaming.data;
+import org.apache.flink.configuration.Configuration;
+
import java.io.IOException;
/**
@@ -32,6 +34,10 @@ public class PythonDualInputSender<IN1, IN2> extends PythonSender {
private transient Serializer<IN1> serializer1;
private transient Serializer<IN2> serializer2;
+ protected PythonDualInputSender(Configuration config) {
+ super(config);
+ }
+
/**
* Extracts records from an iterator and writes them to the memory-mapped file. This method assumes that all values
* in the iterator are of the same type. This method does NOT take care of synchronization. The caller must
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputStreamer.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputStreamer.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputStreamer.java
index 2e9ba2c..b7e8a25 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputStreamer.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonDualInputStreamer.java
@@ -18,6 +18,7 @@
package org.apache.flink.python.api.streaming.data;
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.io.IOException;
@@ -35,8 +36,8 @@ public class PythonDualInputStreamer<IN1, IN2, OUT> extends PythonStreamer<Pytho
private static final long serialVersionUID = -607175070491761873L;
- public PythonDualInputStreamer(AbstractRichFunction function, int envID, int setID, boolean usesByteArray) {
- super(function, envID, setID, usesByteArray, new PythonDualInputSender<IN1, IN2>());
+ public PythonDualInputStreamer(AbstractRichFunction function, Configuration config, int envID, int setID, boolean usesByteArray) {
+ super(function, config, envID, setID, usesByteArray, new PythonDualInputSender<IN1, IN2>(config));
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonReceiver.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonReceiver.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonReceiver.java
index 838a261..c7c1f7a 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonReceiver.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonReceiver.java
@@ -20,8 +20,9 @@ import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_TMP_DATA_DIR;
-import static org.apache.flink.python.api.PythonPlanBinder.MAPPED_FILE_SIZE;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.python.api.PythonOptions;
import org.apache.flink.util.Collector;
/**
@@ -30,43 +31,40 @@ import org.apache.flink.util.Collector;
public class PythonReceiver<OUT> implements Serializable {
private static final long serialVersionUID = -2474088929850009968L;
- private transient File inputFile;
private transient RandomAccessFile inputRAF;
private transient FileChannel inputChannel;
private transient MappedByteBuffer fileBuffer;
+ private final long mappedFileSizeBytes;
+
private final boolean readAsByteArray;
private transient Deserializer<OUT> deserializer;
- public PythonReceiver(boolean usesByteArray) {
+ public PythonReceiver(Configuration config, boolean usesByteArray) {
readAsByteArray = usesByteArray;
+ mappedFileSizeBytes = config.getLong(PythonOptions.MMAP_FILE_SIZE) << 10;
}
//=====Setup========================================================================================================
@SuppressWarnings("unchecked")
- public void open(String path) throws IOException {
- setupMappedFile(path);
+ public void open(File inputFile) throws IOException {
deserializer = (Deserializer<OUT>) (readAsByteArray ? new ByteArrayDeserializer() : new TupleDeserializer());
- }
- private void setupMappedFile(String inputFilePath) throws IOException {
- File x = new File(FLINK_TMP_DATA_DIR);
- x.mkdirs();
+ inputFile.getParentFile().mkdirs();
- inputFile = new File(inputFilePath);
if (inputFile.exists()) {
inputFile.delete();
}
inputFile.createNewFile();
- inputRAF = new RandomAccessFile(inputFilePath, "rw");
- inputRAF.setLength(MAPPED_FILE_SIZE);
- inputRAF.seek(MAPPED_FILE_SIZE - 1);
+ inputRAF = new RandomAccessFile(inputFile, "rw");
+ inputRAF.setLength(mappedFileSizeBytes);
+ inputRAF.seek(mappedFileSizeBytes - 1);
inputRAF.writeByte(0);
inputRAF.seek(0);
inputChannel = inputRAF.getChannel();
- fileBuffer = inputChannel.map(FileChannel.MapMode.READ_WRITE, 0, MAPPED_FILE_SIZE);
+ fileBuffer = inputChannel.map(FileChannel.MapMode.READ_WRITE, 0, mappedFileSizeBytes);
}
public void close() throws IOException {
@@ -76,7 +74,6 @@ public class PythonReceiver<OUT> implements Serializable {
private void closeMappedFile() throws IOException {
inputChannel.close();
inputRAF.close();
- inputFile.delete();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSender.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSender.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSender.java
index 9ada758..3d13271 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSender.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSender.java
@@ -14,6 +14,8 @@ package org.apache.flink.python.api.streaming.data;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.python.api.PythonOptions;
import java.io.File;
import java.io.IOException;
@@ -23,9 +25,6 @@ import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_TMP_DATA_DIR;
-import static org.apache.flink.python.api.PythonPlanBinder.MAPPED_FILE_SIZE;
-
/**
* General-purpose class to write data to memory-mapped files.
*/
@@ -37,32 +36,36 @@ public abstract class PythonSender implements Serializable {
public static final byte TYPE_KEY_VALUE = 62;
public static final byte TYPE_VALUE_VALUE = 61;
- private transient File outputFile;
private transient RandomAccessFile outputRAF;
private transient FileChannel outputChannel;
private transient MappedByteBuffer fileBuffer;
- //=====Setup========================================================================================================
- public void open(String path) throws IOException {
- setupMappedFile(path);
+ private final long mappedFileSizeBytes;
+
+ private final Configuration config;
+
+ protected PythonSender(Configuration config) {
+ this.config = config;
+ mappedFileSizeBytes = config.getLong(PythonOptions.MMAP_FILE_SIZE) << 10;
}
- private void setupMappedFile(String outputFilePath) throws IOException {
- File x = new File(FLINK_TMP_DATA_DIR);
- x.mkdirs();
+ //=====Setup========================================================================================================
+ public void open(File outputFile) throws IOException {
+ outputFile.mkdirs();
- outputFile = new File(outputFilePath);
if (outputFile.exists()) {
outputFile.delete();
}
outputFile.createNewFile();
- outputRAF = new RandomAccessFile(outputFilePath, "rw");
- outputRAF.setLength(MAPPED_FILE_SIZE);
- outputRAF.seek(MAPPED_FILE_SIZE - 1);
+ outputRAF = new RandomAccessFile(outputFile, "rw");
+
+
+ outputRAF.setLength(mappedFileSizeBytes);
+ outputRAF.seek(mappedFileSizeBytes - 1);
outputRAF.writeByte(0);
outputRAF.seek(0);
outputChannel = outputRAF.getChannel();
- fileBuffer = outputChannel.map(FileChannel.MapMode.READ_WRITE, 0, MAPPED_FILE_SIZE);
+ fileBuffer = outputChannel.map(FileChannel.MapMode.READ_WRITE, 0, mappedFileSizeBytes);
}
public void close() throws IOException {
@@ -72,7 +75,6 @@ public abstract class PythonSender implements Serializable {
private void closeMappedFile() throws IOException {
outputChannel.close();
outputRAF.close();
- outputFile.delete();
}
//=====IO===========================================================================================================
@@ -92,7 +94,7 @@ public abstract class PythonSender implements Serializable {
while (input.hasNext()) {
IN value = input.next();
ByteBuffer bb = serializer.serialize(value);
- if (bb.remaining() > MAPPED_FILE_SIZE) {
+ if (bb.remaining() > mappedFileSizeBytes) {
throw new RuntimeException("Serialized object does not fit into a single buffer.");
}
if (bb.remaining() <= fileBuffer.remaining()) {
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputSender.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputSender.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputSender.java
index 42a1799..74d0604 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputSender.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputSender.java
@@ -17,6 +17,8 @@
*/
package org.apache.flink.python.api.streaming.data;
+import org.apache.flink.configuration.Configuration;
+
import java.io.IOException;
/**
@@ -30,6 +32,10 @@ public class PythonSingleInputSender<IN> extends PythonSender {
private transient Serializer<IN> serializer;
+ protected PythonSingleInputSender(Configuration config) {
+ super(config);
+ }
+
/**
* Extracts records from an iterator and writes them to the memory-mapped file. This method assumes that all values
* in the iterator are of the same type. This method does NOT take care of synchronization. The caller must
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputStreamer.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputStreamer.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputStreamer.java
index d013111..e7f018c 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputStreamer.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonSingleInputStreamer.java
@@ -18,6 +18,7 @@
package org.apache.flink.python.api.streaming.data;
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.io.IOException;
@@ -33,8 +34,8 @@ public class PythonSingleInputStreamer<IN, OUT> extends PythonStreamer<PythonSin
private static final long serialVersionUID = -5149905918522069034L;
- public PythonSingleInputStreamer(AbstractRichFunction function, int envID, int setID, boolean usesByteArray) {
- super(function, envID, setID, usesByteArray, new PythonSingleInputSender<IN>());
+ public PythonSingleInputStreamer(AbstractRichFunction function, Configuration config, int envID, int setID, boolean usesByteArray) {
+ super(function, config, envID, setID, usesByteArray, new PythonSingleInputSender<IN>(config));
}
/**
@@ -83,7 +84,9 @@ public class PythonSingleInputStreamer<IN, OUT> extends PythonStreamer<PythonSin
}
}
} catch (SocketTimeoutException ignored) {
- throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg);
+ throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg.get());
+ } catch (Exception e) {
+ throw new RuntimeException("Critical failure. " + msg.get(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java
index 006a1b2..97d5780 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java
@@ -15,7 +15,7 @@ package org.apache.flink.python.api.streaming.data;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.python.api.PythonPlanBinder;
+import org.apache.flink.python.api.PythonOptions;
import org.apache.flink.python.api.streaming.util.SerializationUtils.IntSerializer;
import org.apache.flink.python.api.streaming.util.SerializationUtils.StringSerializer;
import org.apache.flink.python.api.streaming.util.StreamPrinter;
@@ -24,6 +24,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataInputStream;
import java.io.DataOutputStream;
+import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
@@ -34,13 +35,11 @@ import java.net.SocketTimeoutException;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicReference;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON2_BINARY_PATH;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON3_BINARY_PATH;
import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON_DC_ID;
import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON_PLAN_NAME;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_TMP_DATA_DIR;
import static org.apache.flink.python.api.PythonPlanBinder.PLANBINDER_CONFIG_BCVAR_COUNT;
import static org.apache.flink.python.api.PythonPlanBinder.PLANBINDER_CONFIG_BCVAR_NAME_PREFIX;
+import static org.apache.flink.python.api.PythonPlanBinder.PLAN_ARGUMENTS_KEY;
/**
* This streamer is used by functions to send/receive data to/from an external python process.
@@ -56,10 +55,9 @@ public class PythonStreamer<S extends PythonSender, OUT> implements Serializable
protected static final int SIGNAL_ERROR = -2;
protected static final byte SIGNAL_LAST = 32;
+ private final Configuration config;
private final int envID;
private final int setID;
- private final boolean usePython3;
- private final String planArguments;
private transient Process process;
private transient Thread shutdownThread;
@@ -79,12 +77,11 @@ public class PythonStreamer<S extends PythonSender, OUT> implements Serializable
protected transient Thread outPrinter;
protected transient Thread errorPrinter;
- public PythonStreamer(AbstractRichFunction function, int envID, int setID, boolean usesByteArray, S sender) {
+ public PythonStreamer(AbstractRichFunction function, Configuration config, int envID, int setID, boolean usesByteArray, S sender) {
+ this.config = config;
this.envID = envID;
this.setID = setID;
- this.usePython3 = PythonPlanBinder.usePython3;
- planArguments = PythonPlanBinder.arguments.toString();
- receiver = new PythonReceiver<>(usesByteArray);
+ this.receiver = new PythonReceiver<>(config, usesByteArray);
this.function = function;
this.sender = sender;
}
@@ -101,16 +98,21 @@ public class PythonStreamer<S extends PythonSender, OUT> implements Serializable
}
private void startPython() throws IOException {
- String outputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "_" + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "output";
- String inputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "_" + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "input";
+ String tmpDir = config.getString(PythonOptions.DATA_TMP_DIR);
+ if (tmpDir == null) {
+ tmpDir = System.getProperty("java.io.tmpdir");
+ }
+ File outputFile = new File(tmpDir, envID + "_" + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "_output");
+ File inputFile = new File(tmpDir, envID + "_" + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "_input)");
- sender.open(inputFilePath);
- receiver.open(outputFilePath);
+ sender.open(inputFile);
+ receiver.open(outputFile);
String path = function.getRuntimeContext().getDistributedCache().getFile(FLINK_PYTHON_DC_ID).getAbsolutePath();
+
String planPath = path + FLINK_PYTHON_PLAN_NAME;
- String pythonBinaryPath = usePython3 ? FLINK_PYTHON3_BINARY_PATH : FLINK_PYTHON2_BINARY_PATH;
+ String pythonBinaryPath = config.getString(PythonOptions.PYTHON_BINARY_PATH);
try {
Runtime.getRuntime().exec(pythonBinaryPath);
@@ -118,7 +120,7 @@ public class PythonStreamer<S extends PythonSender, OUT> implements Serializable
throw new RuntimeException(pythonBinaryPath + " does not point to a valid python binary.");
}
- process = Runtime.getRuntime().exec(pythonBinaryPath + " -O -B " + planPath + planArguments);
+ process = Runtime.getRuntime().exec(pythonBinaryPath + " -O -B " + planPath + config.getString(PLAN_ARGUMENTS_KEY, ""));
outPrinter = new Thread(new StreamPrinter(process.getInputStream()));
outPrinter.start();
errorPrinter = new Thread(new StreamPrinter(process.getErrorStream(), msg));
@@ -143,8 +145,9 @@ public class PythonStreamer<S extends PythonSender, OUT> implements Serializable
processOutput.write(("" + server.getLocalPort() + "\n").getBytes(ConfigConstants.DEFAULT_CHARSET));
processOutput.write((this.function.getRuntimeContext().getIndexOfThisSubtask() + "\n")
.getBytes(ConfigConstants.DEFAULT_CHARSET));
- processOutput.write((inputFilePath + "\n").getBytes(ConfigConstants.DEFAULT_CHARSET));
- processOutput.write((outputFilePath + "\n").getBytes(ConfigConstants.DEFAULT_CHARSET));
+ processOutput.write(((config.getLong(PythonOptions.MMAP_FILE_SIZE) << 10) + "\n").getBytes(ConfigConstants.DEFAULT_CHARSET));
+ processOutput.write((inputFile + "\n").getBytes(ConfigConstants.DEFAULT_CHARSET));
+ processOutput.write((outputFile + "\n").getBytes(ConfigConstants.DEFAULT_CHARSET));
processOutput.flush();
while (true) {
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/plan/PythonPlanStreamer.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/plan/PythonPlanStreamer.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/plan/PythonPlanStreamer.java
index 9b62563..9e93dda 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/plan/PythonPlanStreamer.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/plan/PythonPlanStreamer.java
@@ -13,6 +13,8 @@
package org.apache.flink.python.api.streaming.plan;
import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.python.api.PythonOptions;
import org.apache.flink.python.api.streaming.util.StreamPrinter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -22,10 +24,7 @@ import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON2_BINARY_PATH;
-import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON3_BINARY_PATH;
import static org.apache.flink.python.api.PythonPlanBinder.FLINK_PYTHON_PLAN_NAME;
-import static org.apache.flink.python.api.PythonPlanBinder.usePython3;
/**
* Generic class to exchange data during the plan phase.
@@ -33,6 +32,7 @@ import static org.apache.flink.python.api.PythonPlanBinder.usePython3;
public class PythonPlanStreamer {
protected static final Logger LOG = LoggerFactory.getLogger(PythonPlanStreamer.class);
+ private final Configuration config;
protected PythonPlanSender sender;
protected PythonPlanReceiver receiver;
@@ -40,6 +40,10 @@ public class PythonPlanStreamer {
private Process process;
private ServerSocket server;
private Socket socket;
+
+ public PythonPlanStreamer(Configuration config) {
+ this.config = config;
+ }
public Object getRecord() throws IOException {
return getRecord(false);
@@ -58,7 +62,7 @@ public class PythonPlanStreamer {
}
private void startPython(String tmpPath, String args) throws IOException {
- String pythonBinaryPath = usePython3 ? FLINK_PYTHON3_BINARY_PATH : FLINK_PYTHON2_BINARY_PATH;
+ String pythonBinaryPath = config.getString(PythonOptions.PYTHON_BINARY_PATH);
try {
Runtime.getRuntime().exec(pythonBinaryPath);
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/connection/Connection.py
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/connection/Connection.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/connection/Connection.py
index 293f5e9..09e6b36 100644
--- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/connection/Connection.py
+++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/connection/Connection.py
@@ -23,8 +23,6 @@ import sys
PY2 = sys.version_info[0] == 2
PY3 = sys.version_info[0] == 3
-MAPPED_FILE_SIZE = 1024 * 1024 * 64
-
SIGNAL_REQUEST_BUFFER = b"\x00\x00\x00\x00"
SIGNAL_REQUEST_BUFFER_G0 = b"\xFF\xFF\xFF\xFD"
SIGNAL_REQUEST_BUFFER_G1 = b"\xFF\xFF\xFF\xFC"
@@ -67,15 +65,16 @@ class PureTCPConnection(object):
class BufferingTCPMappedFileConnection(object):
- def __init__(self, input_file, output_file, port):
+ def __init__(self, input_file, output_file, mmap_size, port):
self._input_file = open(input_file, "rb+")
self._output_file = open(output_file, "rb+")
+ self._mmap_size = mmap_size
if hasattr(mmap, 'MAP_SHARED'):
- self._file_input_buffer = mmap.mmap(self._input_file.fileno(), MAPPED_FILE_SIZE, mmap.MAP_SHARED, mmap.ACCESS_READ)
- self._file_output_buffer = mmap.mmap(self._output_file.fileno(), MAPPED_FILE_SIZE, mmap.MAP_SHARED, mmap.ACCESS_WRITE)
+ self._file_input_buffer = mmap.mmap(self._input_file.fileno(), mmap_size, mmap.MAP_SHARED, mmap.ACCESS_READ)
+ self._file_output_buffer = mmap.mmap(self._output_file.fileno(), mmap_size, mmap.MAP_SHARED, mmap.ACCESS_WRITE)
else:
- self._file_input_buffer = mmap.mmap(self._input_file.fileno(), MAPPED_FILE_SIZE, None, mmap.ACCESS_READ)
- self._file_output_buffer = mmap.mmap(self._output_file.fileno(), MAPPED_FILE_SIZE, None, mmap.ACCESS_WRITE)
+ self._file_input_buffer = mmap.mmap(self._input_file.fileno(), mmap_size, None, mmap.ACCESS_READ)
+ self._file_output_buffer = mmap.mmap(self._output_file.fileno(), mmap_size, None, mmap.ACCESS_WRITE)
self._socket = SOCKET.socket(family=SOCKET.AF_INET, type=SOCKET.SOCK_STREAM)
self._socket.connect((SOCKET.gethostbyname("localhost"), port))
@@ -92,10 +91,10 @@ class BufferingTCPMappedFileConnection(object):
def write(self, msg):
length = len(msg)
- if length > MAPPED_FILE_SIZE:
+ if length > self._mmap_size:
raise Exception("Serialized object does not fit into a single buffer.")
tmp = self._out_size + length
- if tmp > MAPPED_FILE_SIZE:
+ if tmp > self._mmap_size:
self._write_buffer()
self.write(msg)
else:
@@ -150,8 +149,8 @@ class BufferingTCPMappedFileConnection(object):
class TwinBufferingTCPMappedFileConnection(BufferingTCPMappedFileConnection):
- def __init__(self, input_file, output_file, port):
- super(TwinBufferingTCPMappedFileConnection, self).__init__(input_file, output_file, port)
+ def __init__(self, input_file, output_file, mmap_size, port):
+ super(TwinBufferingTCPMappedFileConnection, self).__init__(input_file, output_file, mmap_size, port)
self._input = [b"", b""]
self._input_offset = [0, 0]
self._input_size = [0, 0]
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/CoGroupFunction.py
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/CoGroupFunction.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/CoGroupFunction.py
index 83f563b..edd2c61 100644
--- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/CoGroupFunction.py
+++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/CoGroupFunction.py
@@ -25,8 +25,8 @@ class CoGroupFunction(Function.Function):
self._keys1 = None
self._keys2 = None
- def _configure(self, input_file, output_file, port, env, info, subtask_index):
- self._connection = Connection.TwinBufferingTCPMappedFileConnection(input_file, output_file, port)
+ def _configure(self, input_file, output_file, mmap_size, port, env, info, subtask_index):
+ self._connection = Connection.TwinBufferingTCPMappedFileConnection(input_file, output_file, mmap_size, port)
self._iterator = Iterator.Iterator(self._connection, env, 0)
self._iterator2 = Iterator.Iterator(self._connection, env, 1)
self._cgiter = Iterator.CoGroupIterator(self._iterator, self._iterator2, self._keys1, self._keys2)
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/Function.py
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/Function.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/Function.py
index 45a0f2e..a70a359 100644
--- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/Function.py
+++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/Function.py
@@ -32,8 +32,8 @@ class Function(object):
self.context = None
self._env = None
- def _configure(self, input_file, output_file, port, env, info, subtask_index):
- self._connection = Connection.BufferingTCPMappedFileConnection(input_file, output_file, port)
+ def _configure(self, input_file, output_file, mmap_size, port, env, info, subtask_index):
+ self._connection = Connection.BufferingTCPMappedFileConnection(input_file, output_file, mmap_size, port)
self._iterator = Iterator.Iterator(self._connection, env)
self._collector = Collector.Collector(self._connection, env, info)
self.context = RuntimeContext.RuntimeContext(self._iterator, self._collector, subtask_index)
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/GroupReduceFunction.py
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/GroupReduceFunction.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/GroupReduceFunction.py
index 77b53a2..5a353a1 100644
--- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/GroupReduceFunction.py
+++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/GroupReduceFunction.py
@@ -25,8 +25,8 @@ class GroupReduceFunction(Function.Function):
def __init__(self):
super(GroupReduceFunction, self).__init__()
- def _configure(self, input_file, output_file, port, env, info, subtask_index):
- super(GroupReduceFunction, self)._configure(input_file, output_file, port, env, info, subtask_index)
+ def _configure(self, input_file, output_file, mmap_size, port, env, info, subtask_index):
+ super(GroupReduceFunction, self)._configure(input_file, output_file, mmap_size, port, env, info, subtask_index)
if len(info.key1) == 0:
self._run = self._run_all_group_reduce
else:
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/ReduceFunction.py
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/ReduceFunction.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/ReduceFunction.py
index 08af276..5acabe0 100644
--- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/ReduceFunction.py
+++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/functions/ReduceFunction.py
@@ -24,8 +24,8 @@ class ReduceFunction(Function.Function):
def __init__(self):
super(ReduceFunction, self).__init__()
- def _configure(self, input_file, output_file, port, env, info, subtask_index):
- super(ReduceFunction, self)._configure(input_file, output_file, port, env, info, subtask_index)
+ def _configure(self, input_file, output_file, mmap_size, port, env, info, subtask_index):
+ super(ReduceFunction, self)._configure(input_file, output_file, mmap_size, port, env, info, subtask_index)
if len(info.key1) == 0:
self._run = self._run_all_reduce
else:
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/Environment.py
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/Environment.py b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/Environment.py
index 6e496de..797ae96 100644
--- a/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/Environment.py
+++ b/flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/plan/Environment.py
@@ -234,6 +234,7 @@ class Environment(object):
port = int(sys.stdin.readline().rstrip('\n'))
subtask_index = int(sys.stdin.readline().rstrip('\n'))
+ mmap_size = int(sys.stdin.readline().rstrip('\n'))
input_path = sys.stdin.readline().rstrip('\n')
output_path = sys.stdin.readline().rstrip('\n')
@@ -244,7 +245,7 @@ class Environment(object):
if set.id == id:
used_set = set
operator = set.operator
- operator._configure(input_path, output_path, port, self, used_set, subtask_index)
+ operator._configure(input_path, output_path, mmap_size, port, self, used_set, subtask_index)
operator._go()
operator._close()
sys.stdout.flush()
@@ -252,7 +253,7 @@ class Environment(object):
except:
sys.stdout.flush()
sys.stderr.flush()
- if operator is not None:
+ if operator is not None and operator._connection is not None:
operator._connection._socket.send(struct.pack(">i", -2))
elif port is not None:
socket = SOCKET.socket(family=SOCKET.AF_INET, type=SOCKET.SOCK_STREAM)
http://git-wip-us.apache.org/repos/asf/flink/blob/bdcebfda/flink-libraries/flink-python/src/test/java/org/apache/flink/python/api/PythonPlanBinderTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/test/java/org/apache/flink/python/api/PythonPlanBinderTest.java b/flink-libraries/flink-python/src/test/java/org/apache/flink/python/api/PythonPlanBinderTest.java
index d144298..ba8ea78 100644
--- a/flink-libraries/flink-python/src/test/java/org/apache/flink/python/api/PythonPlanBinderTest.java
+++ b/flink-libraries/flink-python/src/test/java/org/apache/flink/python/api/PythonPlanBinderTest.java
@@ -12,6 +12,7 @@
*/
package org.apache.flink.python.api;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
@@ -21,9 +22,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import static org.apache.flink.python.api.PythonPlanBinder.ARGUMENT_PYTHON_2;
-import static org.apache.flink.python.api.PythonPlanBinder.ARGUMENT_PYTHON_3;
-
public class PythonPlanBinderTest extends JavaProgramTestBase {
@Override
@@ -75,12 +73,16 @@ public class PythonPlanBinderTest extends JavaProgramTestBase {
String utils = findUtilsFile();
if (isPython2Supported()) {
for (String file : findTestFiles()) {
- PythonPlanBinder.main(new String[]{ARGUMENT_PYTHON_2, file, utils});
+ Configuration configuration = new Configuration();
+ config.setString(PythonOptions.PYTHON_BINARY_PATH, "python");
+ new PythonPlanBinder(configuration).runPlan(new String[]{file, utils});
}
}
if (isPython3Supported()) {
for (String file : findTestFiles()) {
- PythonPlanBinder.main(new String[]{ARGUMENT_PYTHON_3, file, utils});
+ Configuration configuration = new Configuration();
+ config.setString(PythonOptions.PYTHON_BINARY_PATH, "python3");
+ new PythonPlanBinder(configuration).runPlan(new String[]{file, utils});
}
}
}
[3/5] flink git commit: [hotfix] [py] Fix PythonCoGroup useByteArray
check
Posted by ch...@apache.org.
[hotfix] [py] Fix PythonCoGroup useByteArray check
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8f78e9d7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8f78e9d7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8f78e9d7
Branch: refs/heads/master
Commit: 8f78e9d72f494c4eb773211beb30a7116d048109
Parents: 89d9dec
Author: zentol <ch...@apache.org>
Authored: Thu Mar 30 23:13:50 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Thu Apr 6 10:57:10 2017 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/python/api/functions/PythonCoGroup.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8f78e9d7/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
index 72d3361..ff5a8d4 100644
--- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
+++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/functions/PythonCoGroup.java
@@ -12,6 +12,7 @@
*/
package org.apache.flink.python.api.functions;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.python.api.streaming.data.PythonDualInputStreamer;
@@ -36,7 +37,7 @@ public class PythonCoGroup<IN1, IN2, OUT> extends RichCoGroupFunction<IN1, IN2,
public PythonCoGroup(int envID, int setID, TypeInformation<OUT> typeInformation) {
this.typeInformation = typeInformation;
- streamer = new PythonDualInputStreamer<>(this, envID, setID, true);
+ streamer = new PythonDualInputStreamer<>(this, envID, setID, typeInformation instanceof PrimitiveArrayTypeInfo);
}
/**