From e2f4a4acda8c180e69a72709dfd777786ac6650b Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Thu, 26 Jan 2017 09:15:55 -0500 Subject: [PATCH 01/10] [FLINK-5183] [py] Support mulitple jobs per plan file Issues to be resolved: * Execution environments need to be able to get a unique ID that can be used to identify it between languages. At the moment, these IDs are manually assigned. * PythonPlanBinder does not exit when Python process exits, waiting indefinitely for more jobs * Global variables used to run operators in Python (there should be a better solution) --- .../flink/python/api/PythonOperationInfo.java | 9 +- .../flink/python/api/PythonPlanBinder.java | 153 ++++++++++-------- .../python/api/functions/PythonCoGroup.java | 2 +- .../api/functions/PythonMapPartition.java | 2 +- .../api/streaming/data/PythonStreamer.java | 13 +- .../streaming/plan/PythonPlanStreamer.java | 25 ++- .../python/api/flink/plan/Environment.py | 84 +++++++--- 7 files changed, 186 insertions(+), 102 deletions(-) diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java index 5f3f9f16698ee1..d387ce383d7891 100644 --- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java +++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java @@ -44,8 +44,11 @@ public class PythonOperationInfo { public String name; public boolean usesUDF; public int parallelism; + public int envID; + public String uniqueID; + public String uniqueParentID; - public PythonOperationInfo(PythonPlanStreamer streamer) throws IOException { + public PythonOperationInfo(PythonPlanStreamer streamer, int environmentID) throws IOException { identifier = (String) streamer.getRecord(); parentID = (Integer) streamer.getRecord(true); otherID = (Integer) streamer.getRecord(true); @@ -92,6 +95,10 @@ public PythonOperationInfo(PythonPlanStreamer streamer) throws IOException { values[x] = streamer.getRecord(); } parallelism = (Integer) streamer.getRecord(true); + + envID = environmentID; + uniqueID = "" + envID + "." + setID; + uniqueParentID = "" + envID + "." + parentID; } @Override 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 cc63ef4ef20116..2615515c74e245 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 @@ -92,8 +92,9 @@ 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 sets = new HashMap<>(); + private HashMap sets = new HashMap<>(); public ExecutionEnvironment env; + private int environmentCounter = 0; private PythonPlanStreamer streamer; public static final int MAPPED_FILE_SIZE = 1024 * 1024 * 64; @@ -126,8 +127,6 @@ public PythonPlanBinder() throws IOException { } private void runPlan(String[] args) throws Exception { - env = ExecutionEnvironment.getExecutionEnvironment(); - int split = 0; for (int x = 0; x < args.length; x++) { if (args[x].compareTo("-") == 0) { @@ -139,15 +138,27 @@ private void runPlan(String[] args) throws Exception { String tmpPath = FLINK_PYTHON_FILE_PATH + r.nextInt(); prepareFiles(tmpPath, Arrays.copyOfRange(args, 0, split == 0 ? args.length : split)); startPython(tmpPath, Arrays.copyOfRange(args, split == 0 ? args.length : split + 1, args.length)); - receivePlan(); - if (env instanceof LocalEnvironment) { - FLINK_HDFS_PATH = "file:" + System.getProperty("java.io.tmpdir") + File.separator + "flink"; + // Python process should terminate itself when all jobs have been run + while (streamer.isPythonRunning()) { + System.out.println("LOL Python still alive right?"); + env = ExecutionEnvironment.getExecutionEnvironment(); + if (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); + + environmentCounter++; + } else { + break; + } } - distributeFiles(tmpPath, env); - JobExecutionResult jer = env.execute(); - sendResult(jer); + clearPath(tmpPath); close(); } catch (Exception e) { close(); @@ -200,7 +211,6 @@ private static void distributeFiles(String tmpPath, ExecutionEnvironment env) th clearPath(FLINK_HDFS_PATH); FileCache.copy(new Path(tmpPath), new Path(FLINK_HDFS_PATH), true); env.registerCachedFile(FLINK_HDFS_PATH, FLINK_PYTHON_DC_ID); - clearPath(tmpPath); } private void startPython(String tempPath, String[] args) throws IOException { @@ -213,6 +223,7 @@ private void startPython(String tempPath, String[] args) throws IOException { private void sendResult(JobExecutionResult jer) throws IOException { long runtime = jer.getNetRuntime(); + System.out.println("JAVA RUNTIME: " + runtime); streamer.sendRecord(runtime); } @@ -233,9 +244,13 @@ private void close() { } //====Plan========================================================================================================== - private void receivePlan() throws IOException { - receiveParameters(); - receiveOperations(); + private boolean receivePlan() throws IOException { + if ( streamer.startPlanMode() ) { + receiveParameters(); + receiveOperations(); + return true; + } + return false; } //====Environment=================================================================================================== @@ -285,7 +300,7 @@ protected enum Operation { private void receiveOperations() throws IOException { Integer operationCount = (Integer) streamer.getRecord(true); for (int x = 0; x < operationCount; x++) { - PythonOperationInfo info = new PythonOperationInfo(streamer); + PythonOperationInfo info = new PythonOperationInfo(streamer, environmentCounter); Operation op; try { op = Operation.valueOf(info.identifier.toUpperCase()); @@ -394,48 +409,48 @@ private void createCsvSource(PythonOperationInfo info) throws IOException { 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") + sets.put(info.uniqueID, env.createInput(new TupleCsvInputFormat(path, lineD, fieldD, types), info.types).setParallelism(getParallelism(info)).name("CsvSource") .map(new SerializerMap<>()).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") + sets.put(info.uniqueID, env.readTextFile(info.path).setParallelism(getParallelism(info)).name("TextSource") .map(new SerializerMap()).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") + sets.put(info.uniqueID, 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") + sets.put(info.uniqueID, env.generateSequence(info.frm, info.to).setParallelism(getParallelism(info)).name("SequenceSource") .map(new SerializerMap()).setParallelism(getParallelism(info)).name("SequenceSourcePostStep")); } @SuppressWarnings("unchecked") private void createCsvSink(PythonOperationInfo info) throws IOException { - DataSet parent = (DataSet) sets.get(info.parentID); + DataSet parent = (DataSet) sets.get(info.uniqueParentID); 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); + DataSet parent = (DataSet) sets.get(info.uniqueParentID); 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); + DataSet parent = (DataSet) sets.get(info.uniqueParentID); parent.map(new StringDeserializerMap()).setParallelism(getParallelism(info)).name("PrintSinkPreStep") .output(new PrintingOutputFormat(info.toError)).setParallelism(getParallelism(info)); } private void createBroadcastVariable(PythonOperationInfo info) throws IOException { - UdfOperator op1 = (UdfOperator) sets.get(info.parentID); + UdfOperator op1 = (UdfOperator) sets.get(info.uniqueParentID); DataSet op2 = (DataSet) sets.get(info.otherID); op1.withBroadcastSet(op2, info.name); @@ -454,77 +469,77 @@ private void createBroadcastVariable(PythonOperationInfo info) throws IOExceptio @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") + DataSet op = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, 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); + Object op = sets.get(info.uniqueParentID); if (op instanceof DataSet) { - sets.put(info.setID, ((DataSet) op).first(info.count).setParallelism(getParallelism(info)).name("First")); + sets.put(info.uniqueID, ((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") + sets.put(info.uniqueID, ((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") + sets.put(info.uniqueID, ((SortedGrouping) op).first(info.count).setParallelism(getParallelism(info)).name("First") .map(new KeyDiscarder()).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)); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, 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)) + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, op1.partitionByHash(info.keys).setParallelism(getParallelism(info)) .map(new KeyDiscarder()).setParallelism(getParallelism(info)).name("HashPartitionPostStep")); } 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")); + DataSet op = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, op.rebalance().setParallelism(getParallelism(info)).name("Rebalance")); } private void createSortOperation(PythonOperationInfo info) throws IOException { - Grouping op1 = (Grouping) sets.get(info.parentID); + Grouping op1 = (Grouping) sets.get(info.uniqueParentID); if (op1 instanceof UnsortedGrouping) { - sets.put(info.setID, ((UnsortedGrouping) op1).sortGroup(info.field, info.order)); + sets.put(info.uniqueID, ((UnsortedGrouping) op1).sortGroup(info.field, info.order)); return; } if (op1 instanceof SortedGrouping) { - sets.put(info.setID, ((SortedGrouping) op1).sortGroup(info.field, info.order)); + sets.put(info.uniqueID, ((SortedGrouping) op1).sortGroup(info.field, info.order)); } } @SuppressWarnings("unchecked") private void createUnionOperation(PythonOperationInfo info) throws IOException { - DataSet op1 = (DataSet) sets.get(info.parentID); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); DataSet op2 = (DataSet) sets.get(info.otherID); - sets.put(info.setID, op1.union(op2).setParallelism(getParallelism(info)).name("Union")); + sets.put(info.uniqueID, op1.union(op2).setParallelism(getParallelism(info)).name("Union")); } @SuppressWarnings("unchecked") private void createCoGroupOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.parentID); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); 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.setID, info.types); - sets.put(info.setID, new CoGroupRawOperator(op1, op2, key1, key2, pcg, info.types, info.name).setParallelism(getParallelism(info))); + PythonCoGroup pcg = new PythonCoGroup(info.uniqueID, info.types); + sets.put(info.uniqueID, new CoGroupRawOperator(op1, op2, key1, key2, pcg, info.types, info.name).setParallelism(getParallelism(info))); } @SuppressWarnings("unchecked") private void createCrossOperation(DatasizeHint mode, PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.parentID); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); DataSet op2 = (DataSet) sets.get(info.otherID); DefaultCross defaultResult; @@ -544,67 +559,67 @@ private void createCrossOperation(DatasizeHint mode, PythonOperationInfo info) { defaultResult.setParallelism(getParallelism(info)); if (info.usesUDF) { - sets.put(info.setID, defaultResult.mapPartition(new PythonMapPartition(info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + sets.put(info.uniqueID, defaultResult.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); } else { - sets.put(info.setID, defaultResult.name("DefaultCross")); + sets.put(info.uniqueID, 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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); } private void createGroupReduceOperation(PythonOperationInfo info) { - Object op1 = sets.get(info.parentID); + Object op1 = sets.get(info.uniqueParentID); if (op1 instanceof DataSet) { - sets.put(info.setID, applyGroupReduceOperation((DataSet) op1, info)); + sets.put(info.uniqueID, applyGroupReduceOperation((DataSet) op1, info)); return; } if (op1 instanceof UnsortedGrouping) { - sets.put(info.setID, applyGroupReduceOperation((UnsortedGrouping) op1, info)); + sets.put(info.uniqueID, applyGroupReduceOperation((UnsortedGrouping) op1, info)); return; } if (op1 instanceof SortedGrouping) { - sets.put(info.setID, applyGroupReduceOperation((SortedGrouping) op1, info)); + sets.put(info.uniqueID, applyGroupReduceOperation((SortedGrouping) op1, info)); } } @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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); } @SuppressWarnings("unchecked") private void createJoinOperation(DatasizeHint mode, PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.parentID); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); DataSet op2 = (DataSet) sets.get(info.otherID); if (info.usesUDF) { - sets.put(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)) - .mapPartition(new PythonMapPartition(info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + sets.put(info.uniqueID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)) + .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); } else { - sets.put(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))); + sets.put(info.uniqueID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))); } } @@ -627,36 +642,36 @@ private DataSet createDefaultJoin(DataSet op1, DataSet op2, String[] firstKeys, @SuppressWarnings("unchecked") private void createMapOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.parentID); - sets.put(info.setID, op1.mapPartition(new PythonMapPartition(info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + DataSet op1 = (DataSet) sets.get(info.uniqueParentID); + sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); } private void createReduceOperation(PythonOperationInfo info) { - Object op1 = sets.get(info.parentID); + Object op1 = sets.get(info.uniqueParentID); if (op1 instanceof DataSet) { - sets.put(info.setID, applyReduceOperation((DataSet) op1, info)); + sets.put(info.uniqueID, applyReduceOperation((DataSet) op1, info)); return; } if (op1 instanceof UnsortedGrouping) { - sets.put(info.setID, applyReduceOperation((UnsortedGrouping) op1, info)); + sets.put(info.uniqueID, applyReduceOperation((UnsortedGrouping) op1, info)); } } @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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).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.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); } } 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 33d88c3182d76b..c9d33482af649a 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 @@ -31,7 +31,7 @@ public class PythonCoGroup extends RichCoGroupFunction typeInformation; - public PythonCoGroup(int id, TypeInformation typeInformation) { + public PythonCoGroup(String id, TypeInformation typeInformation) { this.typeInformation = typeInformation; streamer = new PythonStreamer(this, id, true); } 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 6282210c4863d2..6a773a76f0145f 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 @@ -32,7 +32,7 @@ public class PythonMapPartition extends RichMapPartitionFunction typeInformation; - public PythonMapPartition(int id, TypeInformation typeInformation) { + public PythonMapPartition(String id, TypeInformation typeInformation) { this.typeInformation = typeInformation; streamer = new PythonStreamer(this, id, typeInformation instanceof PrimitiveArrayTypeInfo); } 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 10aded8f0fff55..93e3bc476be4dd 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 @@ -51,7 +51,7 @@ public class PythonStreamer implements Serializable { private static final int SIGNAL_ERROR = -2; private static final byte SIGNAL_LAST = 32; - private final int id; + private final String id; private final boolean usePython3; private final String planArguments; @@ -73,7 +73,7 @@ public class PythonStreamer implements Serializable { protected final AbstractRichFunction function; - public PythonStreamer(AbstractRichFunction function, int id, boolean usesByteArray) { + public PythonStreamer(AbstractRichFunction function, String id, boolean usesByteArray) { this.id = id; this.usePython3 = PythonPlanBinder.usePython3; planArguments = PythonPlanBinder.arguments.toString(); @@ -126,10 +126,17 @@ public void run() { Runtime.getRuntime().addShutdownHook(shutdownThread); + System.out.println(id); + + String envID = id.split("\\.")[0], + setID = id.split("\\.")[1]; + + System.out.println("JAVA ENTERING OPS MODE"); OutputStream processOutput = process.getOutputStream(); processOutput.write("operator\n".getBytes()); + processOutput.write((envID + "\n").getBytes()); + processOutput.write((setID + "\n").getBytes()); processOutput.write(("" + server.getLocalPort() + "\n").getBytes()); - processOutput.write((id + "\n").getBytes()); processOutput.write((this.function.getRuntimeContext().getIndexOfThisSubtask() + "\n").getBytes()); processOutput.write((inputFilePath + "\n").getBytes()); processOutput.write((outputFilePath + "\n").getBytes()); 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 ecbc7f4365bb77..d34090e86c3202 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 @@ -46,11 +46,7 @@ public void sendRecord(Object record) throws IOException { } public void open(String tmpPath, String args) throws IOException { - server = new ServerSocket(0); startPython(tmpPath, args); - socket = server.accept(); - sender = new PythonPlanSender(socket.getOutputStream()); - receiver = new PythonPlanReceiver(socket.getInputStream()); } private void startPython(String tmpPath, String args) throws IOException { @@ -81,18 +77,37 @@ private void startPython(String tmpPath, String args) throws IOException { } } catch (IllegalThreadStateException ise) {//Process still running } + } + public boolean startPlanMode() throws IOException { + System.out.println("JAVA ENTERING PLAN MODE"); + server = new ServerSocket(0); process.getOutputStream().write("plan\n".getBytes()); process.getOutputStream().write((server.getLocalPort() + "\n").getBytes()); process.getOutputStream().flush(); + if (isPythonRunning()) { + socket = server.accept(); + sender = new PythonPlanSender(socket.getOutputStream()); + receiver = new PythonPlanReceiver(socket.getInputStream()); + return true; + } else { + return false; + } } public void close() { + if (isPythonRunning()) { + process.destroy(); + } + } + + public boolean isPythonRunning() { try { process.exitValue(); } catch (NullPointerException npe) { //exception occurred before process was started } catch (IllegalThreadStateException ise) { //process still active - process.destroy(); + return true; } + return false; } } 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 1e4ba1af4d74f6..b00142dcce5539 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 @@ -27,17 +27,22 @@ import sys from struct import pack -def get_environment(): +_env_counter = 0 +_last_env_id = -1 +_operating = False + +def get_environment(id): """ Creates an execution environment that represents the context in which the program is currently executed. :return:The execution environment of the context in which the program is executed. """ - return Environment() + # TODO: auto-number + return Environment(id) class Environment(object): - def __init__(self): + def __init__(self, env_id = 0): # util self._counter = 0 @@ -46,6 +51,8 @@ def __init__(self): self._local_mode = False self._retry = 0 + self.env_id = env_id + #sets self._sources = [] self._sets = [] @@ -163,13 +170,21 @@ def execute(self, local=False): The environment will execute all parts of the program that have resulted in a "sink" operation. """ + global _operating, _last_env_id self._local_mode = local self._optimize_plan() - plan_mode = sys.stdin.readline().rstrip('\n') == "plan" + if _operating: + plan_mode = False + else: + plan_mode = sys.stdin.readline().rstrip('\n') == "plan" if plan_mode: + print("PYTHON ENTERING PLAN MODE") + sys.stdout.flush() port = int(sys.stdin.readline().rstrip('\n')) + print("pyport: " + str(port)) + sys.stdout.flush() self._connection = Connection.PureTCPConnection(port) self._iterator = Iterator.PlanIterator(self._connection, self) self._collector = Collector.PlanCollector(self._connection, self) @@ -178,27 +193,50 @@ def execute(self, local=False): self._connection.close() return result else: + print("PYTHON ENTERING OPS MODE") + sys.stdout.flush() import struct operator = None try: - port = int(sys.stdin.readline().rstrip('\n')) - - id = int(sys.stdin.readline().rstrip('\n')) - subtask_index = int(sys.stdin.readline().rstrip('\n')) - input_path = sys.stdin.readline().rstrip('\n') - output_path = sys.stdin.readline().rstrip('\n') - - used_set = None - operator = None - for set in self._sets: - if set.id == id: - used_set = set - operator = set.operator - operator._configure(input_path, output_path, port, self, used_set, subtask_index) - operator._go() - operator._close() - sys.stdout.flush() - sys.stderr.flush() + env_id = _last_env_id + if not _operating: + env_id = int(sys.stdin.readline().rstrip('\n')) + + if self.env_id != env_id: + _last_env_id = env_id + _operating = True + else: + _operating = False + _last_env_id = -1 + id = int(sys.stdin.readline().rstrip('\n')) + print('id') + print(id) + sys.stdout.flush() + print('sets') + print(self.env_id) + print(self._sets) + print(len(self._sets)) + for s in self._sets: + print(s.id) + sys.stdout.flush() + + port = int(sys.stdin.readline().rstrip('\n')) + subtask_index = int(sys.stdin.readline().rstrip('\n')) + input_path = sys.stdin.readline().rstrip('\n') + output_path = sys.stdin.readline().rstrip('\n') + + used_set = None + operator = None + + for set in self._sets: + if set.id == id: + used_set = set + operator = set.operator + operator._configure(input_path, output_path, port, self, used_set, subtask_index) + operator._go() + operator._close() + sys.stdout.flush() + sys.stderr.flush() except: sys.stdout.flush() sys.stderr.flush() @@ -319,6 +357,8 @@ def _send_operation(self, set): def _receive_result(self): jer = JobExecutionResult() jer._net_runtime = self._iterator.next() + print("PYTHON RUNTIME: " + str(jer._net_runtime)) + sys.stdout.flush() return jer From ee500403b62ebf9c8b923d5940aefb2f928a5c13 Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Thu, 26 Jan 2017 09:27:47 -0500 Subject: [PATCH 02/10] Remove debug print statements --- .../flink/python/api/PythonPlanBinder.java | 2 -- .../api/streaming/data/PythonStreamer.java | 3 --- .../api/streaming/plan/PythonPlanStreamer.java | 1 - .../flink/python/api/flink/plan/Environment.py | 18 ------------------ 4 files changed, 24 deletions(-) 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 2615515c74e245..58ed86f33b5746 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 @@ -141,7 +141,6 @@ private void runPlan(String[] args) throws Exception { // Python process should terminate itself when all jobs have been run while (streamer.isPythonRunning()) { - System.out.println("LOL Python still alive right?"); env = ExecutionEnvironment.getExecutionEnvironment(); if (receivePlan()) { if (env instanceof LocalEnvironment) { @@ -223,7 +222,6 @@ private void startPython(String tempPath, String[] args) throws IOException { private void sendResult(JobExecutionResult jer) throws IOException { long runtime = jer.getNetRuntime(); - System.out.println("JAVA RUNTIME: " + runtime); streamer.sendRecord(runtime); } 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 93e3bc476be4dd..4b4ee77cb6df5d 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 @@ -126,12 +126,9 @@ public void run() { Runtime.getRuntime().addShutdownHook(shutdownThread); - System.out.println(id); - String envID = id.split("\\.")[0], setID = id.split("\\.")[1]; - System.out.println("JAVA ENTERING OPS MODE"); OutputStream processOutput = process.getOutputStream(); processOutput.write("operator\n".getBytes()); processOutput.write((envID + "\n").getBytes()); 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 d34090e86c3202..2363097dae22ad 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 @@ -80,7 +80,6 @@ private void startPython(String tmpPath, String args) throws IOException { } public boolean startPlanMode() throws IOException { - System.out.println("JAVA ENTERING PLAN MODE"); server = new ServerSocket(0); process.getOutputStream().write("plan\n".getBytes()); process.getOutputStream().write((server.getLocalPort() + "\n").getBytes()); 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 b00142dcce5539..5559c77be1730e 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 @@ -180,11 +180,7 @@ def execute(self, local=False): plan_mode = sys.stdin.readline().rstrip('\n') == "plan" if plan_mode: - print("PYTHON ENTERING PLAN MODE") - sys.stdout.flush() port = int(sys.stdin.readline().rstrip('\n')) - print("pyport: " + str(port)) - sys.stdout.flush() self._connection = Connection.PureTCPConnection(port) self._iterator = Iterator.PlanIterator(self._connection, self) self._collector = Collector.PlanCollector(self._connection, self) @@ -193,8 +189,6 @@ def execute(self, local=False): self._connection.close() return result else: - print("PYTHON ENTERING OPS MODE") - sys.stdout.flush() import struct operator = None try: @@ -209,16 +203,6 @@ def execute(self, local=False): _operating = False _last_env_id = -1 id = int(sys.stdin.readline().rstrip('\n')) - print('id') - print(id) - sys.stdout.flush() - print('sets') - print(self.env_id) - print(self._sets) - print(len(self._sets)) - for s in self._sets: - print(s.id) - sys.stdout.flush() port = int(sys.stdin.readline().rstrip('\n')) subtask_index = int(sys.stdin.readline().rstrip('\n')) @@ -357,8 +341,6 @@ def _send_operation(self, set): def _receive_result(self): jer = JobExecutionResult() jer._net_runtime = self._iterator.next() - print("PYTHON RUNTIME: " + str(jer._net_runtime)) - sys.stdout.flush() return jer From afdeddc03f9817a74927a727733ef5c0025af0c7 Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Thu, 26 Jan 2017 12:04:23 -0500 Subject: [PATCH 03/10] Auto assign environment IDs; rm Python alive checks that didn't help --- .../flink/python/api/PythonPlanBinder.java | 30 ++++++++----------- .../streaming/plan/PythonPlanStreamer.java | 13 +++----- .../python/api/flink/plan/Environment.py | 6 ++-- 3 files changed, 20 insertions(+), 29 deletions(-) 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 58ed86f33b5746..8f786d525d3e5c 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 @@ -142,19 +142,16 @@ private void runPlan(String[] args) throws Exception { // Python process should terminate itself when all jobs have been run while (streamer.isPythonRunning()) { env = ExecutionEnvironment.getExecutionEnvironment(); - if (receivePlan()) { - if (env instanceof LocalEnvironment) { - FLINK_HDFS_PATH = "file:" + System.getProperty("java.io.tmpdir") + File.separator + "flink"; - } + 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); + distributeFiles(tmpPath, env); + JobExecutionResult jer = env.execute(); + sendResult(jer); - environmentCounter++; - } else { - break; - } + environmentCounter++; } clearPath(tmpPath); @@ -242,13 +239,10 @@ private void close() { } //====Plan========================================================================================================== - private boolean receivePlan() throws IOException { - if ( streamer.startPlanMode() ) { - receiveParameters(); - receiveOperations(); - return true; - } - return false; + private void receivePlan() throws IOException { + streamer.startPlanMode(); + receiveParameters(); + receiveOperations(); } //====Environment=================================================================================================== 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 2363097dae22ad..14b1ed42587a5f 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 @@ -79,19 +79,14 @@ private void startPython(String tmpPath, String args) throws IOException { } } - public boolean startPlanMode() throws IOException { + public void startPlanMode() throws IOException { server = new ServerSocket(0); process.getOutputStream().write("plan\n".getBytes()); process.getOutputStream().write((server.getLocalPort() + "\n").getBytes()); process.getOutputStream().flush(); - if (isPythonRunning()) { - socket = server.accept(); - sender = new PythonPlanSender(socket.getOutputStream()); - receiver = new PythonPlanReceiver(socket.getInputStream()); - return true; - } else { - return false; - } + socket = server.accept(); + sender = new PythonPlanSender(socket.getOutputStream()); + receiver = new PythonPlanReceiver(socket.getInputStream()); } public void close() { 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 5559c77be1730e..570eac844a98af 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 @@ -31,13 +31,15 @@ _last_env_id = -1 _operating = False -def get_environment(id): +def get_environment(): """ Creates an execution environment that represents the context in which the program is currently executed. :return:The execution environment of the context in which the program is executed. """ - # TODO: auto-number + global _env_counter + id = _env_counter + _env_counter += 1 return Environment(id) From 5dac565d89a27cc113cd4abce7fbdd57485e4401 Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Thu, 26 Jan 2017 16:15:15 -0500 Subject: [PATCH 04/10] Set 5 second timeout for PythonPlanStreamer socket After the timeout, we check to see if the Python process has exited or not. This allows PythonPlanBinder to exit properly when the Python process has exited. --- .../org/apache/flink/python/api/PythonPlanBinder.java | 10 +++++++++- .../python/api/streaming/plan/PythonPlanStreamer.java | 4 ++++ 2 files changed, 13 insertions(+), 1 deletion(-) 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 8f786d525d3e5c..4e9dea908d65b5 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 @@ -14,6 +14,7 @@ import java.io.File; import java.io.IOException; +import java.net.SocketTimeoutException; import java.net.URI; import java.net.URISyntaxException; import java.util.Arrays; @@ -142,7 +143,14 @@ private void runPlan(String[] args) throws Exception { // Python process should terminate itself when all jobs have been run while (streamer.isPythonRunning()) { env = ExecutionEnvironment.getExecutionEnvironment(); - receivePlan(); + + try { + receivePlan(); + } catch (SocketTimeoutException ste) { + // If the socket times out, check to see if Python process has exited yet + continue; + } + if (env instanceof LocalEnvironment) { FLINK_HDFS_PATH = "file:" + System.getProperty("java.io.tmpdir") + File.separator + "flink"; } 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 14b1ed42587a5f..59b2ff97fa93d3 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 @@ -81,9 +81,13 @@ private void startPython(String tmpPath, String args) throws IOException { public void startPlanMode() throws IOException { server = new ServerSocket(0); + //If after 5 seconds Python doesn't respond, check to see if the Python process has exited + server.setSoTimeout(5000); + process.getOutputStream().write("plan\n".getBytes()); process.getOutputStream().write((server.getLocalPort() + "\n").getBytes()); process.getOutputStream().flush(); + socket = server.accept(); sender = new PythonPlanSender(socket.getOutputStream()); receiver = new PythonPlanReceiver(socket.getInputStream()); From e34522e54921989db6ab37ce3b134672b6980f67 Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Thu, 26 Jan 2017 17:26:03 -0500 Subject: [PATCH 05/10] Multiple jobs per Python plan file unit test --- .../flink/python/api/test_multiple_jobs.py | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_multiple_jobs.py diff --git a/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_multiple_jobs.py b/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_multiple_jobs.py new file mode 100644 index 00000000000000..2e8d1a0fa99b20 --- /dev/null +++ b/flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_multiple_jobs.py @@ -0,0 +1,47 @@ + +# ############################################################################### +# 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. +################################################################################ +from flink.plan.Environment import get_environment +from flink.functions.MapFunction import MapFunction +from flink.functions.CrossFunction import CrossFunction +from flink.functions.JoinFunction import JoinFunction +from flink.functions.CoGroupFunction import CoGroupFunction +from flink.functions.Aggregation import Max, Min, Sum +from utils import Verify, Verify2, Id + +# Test multiple jobs in one Python plan file +if __name__ == "__main__": + env = get_environment() + env.set_parallelism(1) + + d1 = env.from_elements(1, 6, 12) + d1 \ + .first(1) \ + .map_partition(Verify([1], "First with multiple jobs in one Python plan file")).output() + + env.execute(local=True) + + env2 = get_environment() + env2.set_parallelism(1) + + d2 = env2.from_elements(1, 1, 12) + d2 \ + .map(lambda x: x * 2) \ + .map_partition(Verify([2, 2, 24], "Lambda Map with multiple jobs in one Python plan file")).output() + + env2.execute(local=True) From 64c013f8a9942b05f1b3000149a66e34b7b366b1 Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Fri, 27 Jan 2017 09:05:25 -0500 Subject: [PATCH 06/10] Fix support for operators using multiple data sets --- .../apache/flink/python/api/PythonOperationInfo.java | 2 ++ .../org/apache/flink/python/api/PythonPlanBinder.java | 10 +++++----- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java index d387ce383d7891..a16f36a753816d 100644 --- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java +++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java @@ -46,6 +46,7 @@ public class PythonOperationInfo { public int parallelism; public int envID; public String uniqueID; + public String uniqueOtherID; public String uniqueParentID; public PythonOperationInfo(PythonPlanStreamer streamer, int environmentID) throws IOException { @@ -98,6 +99,7 @@ public PythonOperationInfo(PythonPlanStreamer streamer, int environmentID) throw envID = environmentID; uniqueID = "" + envID + "." + setID; + uniqueOtherID = "" + envID + "." + otherID; uniqueParentID = "" + envID + "." + parentID; } 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 4e9dea908d65b5..094e56e01edc4e 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 @@ -451,7 +451,7 @@ private void createPrintSink(PythonOperationInfo info) throws IOException { private void createBroadcastVariable(PythonOperationInfo info) throws IOException { UdfOperator op1 = (UdfOperator) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.otherID); + DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); op1.withBroadcastSet(op2, info.name); Configuration c = op1.getParameters(); @@ -523,14 +523,14 @@ private void createSortOperation(PythonOperationInfo info) throws IOException { @SuppressWarnings("unchecked") private void createUnionOperation(PythonOperationInfo info) throws IOException { DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.otherID); + DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); sets.put(info.uniqueID, op1.union(op2).setParallelism(getParallelism(info)).name("Union")); } @SuppressWarnings("unchecked") private void createCoGroupOperation(PythonOperationInfo info) { DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.otherID); + DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); 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.uniqueID, info.types); @@ -540,7 +540,7 @@ private void createCoGroupOperation(PythonOperationInfo info) { @SuppressWarnings("unchecked") private void createCrossOperation(DatasizeHint mode, PythonOperationInfo info) { DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.otherID); + DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); DefaultCross defaultResult; switch (mode) { @@ -613,7 +613,7 @@ private DataSet applyGroupReduceOperation(SortedGrouping op1, PythonOperationInf @SuppressWarnings("unchecked") private void createJoinOperation(DatasizeHint mode, PythonOperationInfo info) { DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.otherID); + DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); if (info.usesUDF) { sets.put(info.uniqueID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)) From 782f69ac1d17f8ff5812216c69fa4cfd119d5a60 Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Sun, 29 Jan 2017 01:02:36 -0500 Subject: [PATCH 07/10] Create Python ExecutionEnvironment after checking if Python has job This prevents ExecutionEnvironment#getLastJobExecutionResult from being null. --- .../java/org/apache/flink/python/api/PythonPlanBinder.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 094e56e01edc4e..fd9e72776a0e3c 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 @@ -142,8 +142,6 @@ private void runPlan(String[] args) throws Exception { // Python process should terminate itself when all jobs have been run while (streamer.isPythonRunning()) { - env = ExecutionEnvironment.getExecutionEnvironment(); - try { receivePlan(); } catch (SocketTimeoutException ste) { @@ -249,6 +247,7 @@ private void close() { //====Plan========================================================================================================== private void receivePlan() throws IOException { streamer.startPlanMode(); + env = ExecutionEnvironment.getExecutionEnvironment(); receiveParameters(); receiveOperations(); } From de808edc5980fc3d7cc1dfa60ed94fbaf99b740e Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Mon, 30 Jan 2017 15:11:13 -0500 Subject: [PATCH 08/10] [py] Clear sets HashMap between ExecEnvs instead of using unique IDs --- .../flink/python/api/PythonOperationInfo.java | 6 - .../flink/python/api/PythonPlanBinder.java | 124 +++++++++--------- .../python/api/functions/PythonCoGroup.java | 4 +- .../api/functions/PythonMapPartition.java | 4 +- .../api/streaming/data/PythonStreamer.java | 15 +-- 5 files changed, 74 insertions(+), 79 deletions(-) diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java index a16f36a753816d..694c1b42e2cb3e 100644 --- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java +++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/PythonOperationInfo.java @@ -45,9 +45,6 @@ public class PythonOperationInfo { public boolean usesUDF; public int parallelism; public int envID; - public String uniqueID; - public String uniqueOtherID; - public String uniqueParentID; public PythonOperationInfo(PythonPlanStreamer streamer, int environmentID) throws IOException { identifier = (String) streamer.getRecord(); @@ -98,9 +95,6 @@ public PythonOperationInfo(PythonPlanStreamer streamer, int environmentID) throw parallelism = (Integer) streamer.getRecord(true); envID = environmentID; - uniqueID = "" + envID + "." + setID; - uniqueOtherID = "" + envID + "." + otherID; - uniqueParentID = "" + envID + "." + parentID; } @Override 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 fd9e72776a0e3c..e03eb3ca94c826 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 @@ -93,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 sets = new HashMap<>(); + private HashMap sets = new HashMap<>(); public ExecutionEnvironment env; private int environmentCounter = 0; private PythonPlanStreamer streamer; @@ -248,6 +248,8 @@ private void close() { private void receivePlan() throws IOException { streamer.startPlanMode(); env = ExecutionEnvironment.getExecutionEnvironment(); + //IDs used in HashMap of sets are only unique for each environment + sets.clear(); receiveParameters(); receiveOperations(); } @@ -408,49 +410,49 @@ private void createCsvSource(PythonOperationInfo info) throws IOException { String lineD = info.lineDelimiter; String fieldD = info.fieldDelimiter; TupleTypeInfo types = (TupleTypeInfo) info.types; - sets.put(info.uniqueID, env.createInput(new TupleCsvInputFormat(path, lineD, fieldD, types), info.types).setParallelism(getParallelism(info)).name("CsvSource") + 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")); } private void createTextSource(PythonOperationInfo info) throws IOException { - sets.put(info.uniqueID, env.readTextFile(info.path).setParallelism(getParallelism(info)).name("TextSource") + sets.put(info.setID, env.readTextFile(info.path).setParallelism(getParallelism(info)).name("TextSource") .map(new SerializerMap()).setParallelism(getParallelism(info)).name("TextSourcePostStep")); } private void createValueSource(PythonOperationInfo info) throws IOException { - sets.put(info.uniqueID, env.fromElements(info.values).setParallelism(getParallelism(info)).name("ValueSource") + sets.put(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.uniqueID, env.generateSequence(info.frm, info.to).setParallelism(getParallelism(info)).name("SequenceSource") + sets.put(info.setID, env.generateSequence(info.frm, info.to).setParallelism(getParallelism(info)).name("SequenceSource") .map(new SerializerMap()).setParallelism(getParallelism(info)).name("SequenceSourcePostStep")); } @SuppressWarnings("unchecked") private void createCsvSink(PythonOperationInfo info) throws IOException { - DataSet parent = (DataSet) sets.get(info.uniqueParentID); + DataSet parent = (DataSet) sets.get(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.uniqueParentID); + DataSet parent = (DataSet) sets.get(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.uniqueParentID); + DataSet parent = (DataSet) sets.get(info.parentID); parent.map(new StringDeserializerMap()).setParallelism(getParallelism(info)).name("PrintSinkPreStep") .output(new PrintingOutputFormat(info.toError)).setParallelism(getParallelism(info)); } private void createBroadcastVariable(PythonOperationInfo info) throws IOException { - UdfOperator op1 = (UdfOperator) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); + UdfOperator op1 = (UdfOperator) sets.get(info.parentID); + DataSet op2 = (DataSet) sets.get(info.otherID); op1.withBroadcastSet(op2, info.name); Configuration c = op1.getParameters(); @@ -468,78 +470,78 @@ private void createBroadcastVariable(PythonOperationInfo info) throws IOExceptio @SuppressWarnings("unchecked") private void createDistinctOperation(PythonOperationInfo info) throws IOException { - DataSet op = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op.distinct(info.keys).setParallelism(getParallelism(info)).name("Distinct") + 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.uniqueParentID); + Object op = sets.get(info.parentID); if (op instanceof DataSet) { - sets.put(info.uniqueID, ((DataSet) op).first(info.count).setParallelism(getParallelism(info)).name("First")); + sets.put(info.setID, ((DataSet) op).first(info.count).setParallelism(getParallelism(info)).name("First")); return; } if (op instanceof UnsortedGrouping) { - sets.put(info.uniqueID, ((UnsortedGrouping) op).first(info.count).setParallelism(getParallelism(info)).name("First") + 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.uniqueID, ((SortedGrouping) op).first(info.count).setParallelism(getParallelism(info)).name("First") + sets.put(info.setID, ((SortedGrouping) op).first(info.count).setParallelism(getParallelism(info)).name("First") .map(new KeyDiscarder()).setParallelism(getParallelism(info)).name("FirstPostStep")); } } private void createGroupOperation(PythonOperationInfo info) throws IOException { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op1.groupBy(info.keys)); + DataSet op1 = (DataSet) sets.get(info.parentID); + sets.put(info.setID, op1.groupBy(info.keys)); } @SuppressWarnings("unchecked") private void createHashPartitionOperation(PythonOperationInfo info) throws IOException { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op1.partitionByHash(info.keys).setParallelism(getParallelism(info)) + 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 void createRebalanceOperation(PythonOperationInfo info) throws IOException { - DataSet op = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op.rebalance().setParallelism(getParallelism(info)).name("Rebalance")); + DataSet op = (DataSet) sets.get(info.parentID); + sets.put(info.setID, op.rebalance().setParallelism(getParallelism(info)).name("Rebalance")); } private void createSortOperation(PythonOperationInfo info) throws IOException { - Grouping op1 = (Grouping) sets.get(info.uniqueParentID); + Grouping op1 = (Grouping) sets.get(info.parentID); if (op1 instanceof UnsortedGrouping) { - sets.put(info.uniqueID, ((UnsortedGrouping) op1).sortGroup(info.field, info.order)); + sets.put(info.setID, ((UnsortedGrouping) op1).sortGroup(info.field, info.order)); return; } if (op1 instanceof SortedGrouping) { - sets.put(info.uniqueID, ((SortedGrouping) op1).sortGroup(info.field, info.order)); + sets.put(info.setID, ((SortedGrouping) op1).sortGroup(info.field, info.order)); } } @SuppressWarnings("unchecked") private void createUnionOperation(PythonOperationInfo info) throws IOException { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); - sets.put(info.uniqueID, op1.union(op2).setParallelism(getParallelism(info)).name("Union")); + 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")); } @SuppressWarnings("unchecked") private void createCoGroupOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); + 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.uniqueID, info.types); - sets.put(info.uniqueID, new CoGroupRawOperator(op1, op2, key1, key2, pcg, info.types, info.name).setParallelism(getParallelism(info))); + 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))); } @SuppressWarnings("unchecked") private void createCrossOperation(DatasizeHint mode, PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); + DataSet op1 = (DataSet) sets.get(info.parentID); + DataSet op2 = (DataSet) sets.get(info.otherID); DefaultCross defaultResult; switch (mode) { @@ -558,67 +560,67 @@ private void createCrossOperation(DatasizeHint mode, PythonOperationInfo info) { defaultResult.setParallelism(getParallelism(info)); if (info.usesUDF) { - sets.put(info.uniqueID, defaultResult.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + sets.put(info.setID, defaultResult.mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name)); } else { - sets.put(info.uniqueID, defaultResult.name("DefaultCross")); + sets.put(info.setID, defaultResult.name("DefaultCross")); } } @SuppressWarnings("unchecked") private void createFilterOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + 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)); } @SuppressWarnings("unchecked") private void createFlatMapOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + 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 void createGroupReduceOperation(PythonOperationInfo info) { - Object op1 = sets.get(info.uniqueParentID); + Object op1 = sets.get(info.parentID); if (op1 instanceof DataSet) { - sets.put(info.uniqueID, applyGroupReduceOperation((DataSet) op1, info)); + sets.put(info.setID, applyGroupReduceOperation((DataSet) op1, info)); return; } if (op1 instanceof UnsortedGrouping) { - sets.put(info.uniqueID, applyGroupReduceOperation((UnsortedGrouping) op1, info)); + sets.put(info.setID, applyGroupReduceOperation((UnsortedGrouping) op1, info)); return; } if (op1 instanceof SortedGrouping) { - sets.put(info.uniqueID, applyGroupReduceOperation((SortedGrouping) op1, info)); + sets.put(info.setID, applyGroupReduceOperation((SortedGrouping) op1, info)); } } @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.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).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.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).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.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); } @SuppressWarnings("unchecked") private void createJoinOperation(DatasizeHint mode, PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - DataSet op2 = (DataSet) sets.get(info.uniqueOtherID); + DataSet op1 = (DataSet) sets.get(info.parentID); + DataSet op2 = (DataSet) sets.get(info.otherID); if (info.usesUDF) { - sets.put(info.uniqueID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info)) - .mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + 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)); } else { - sets.put(info.uniqueID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))); + sets.put(info.setID, createDefaultJoin(op1, op2, info.keys1, info.keys2, mode, getParallelism(info))); } } @@ -641,36 +643,36 @@ private DataSet createDefaultJoin(DataSet op1, DataSet op2, String[] firstKeys, @SuppressWarnings("unchecked") private void createMapOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + 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)); } @SuppressWarnings("unchecked") private void createMapPartitionOperation(PythonOperationInfo info) { - DataSet op1 = (DataSet) sets.get(info.uniqueParentID); - sets.put(info.uniqueID, op1.mapPartition(new PythonMapPartition(info.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name)); + 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 void createReduceOperation(PythonOperationInfo info) { - Object op1 = sets.get(info.uniqueParentID); + Object op1 = sets.get(info.parentID); if (op1 instanceof DataSet) { - sets.put(info.uniqueID, applyReduceOperation((DataSet) op1, info)); + sets.put(info.setID, applyReduceOperation((DataSet) op1, info)); return; } if (op1 instanceof UnsortedGrouping) { - sets.put(info.uniqueID, applyReduceOperation((UnsortedGrouping) op1, info)); + sets.put(info.setID, applyReduceOperation((UnsortedGrouping) op1, info)); } } @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.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).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.uniqueID, info.types)).setParallelism(getParallelism(info)).name(info.name); + .mapPartition(new PythonMapPartition(info.envID, info.setID, info.types)).setParallelism(getParallelism(info)).name(info.name); } } 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 c9d33482af649a..1775c92385c995 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 @@ -31,9 +31,9 @@ public class PythonCoGroup extends RichCoGroupFunction typeInformation; - public PythonCoGroup(String id, TypeInformation typeInformation) { + public PythonCoGroup(int envID, int setID, TypeInformation typeInformation) { this.typeInformation = typeInformation; - streamer = new PythonStreamer(this, id, true); + streamer = new PythonStreamer(this, envID, setID, true); } /** 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 6a773a76f0145f..3c54af9d0093d6 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 @@ -32,9 +32,9 @@ public class PythonMapPartition extends RichMapPartitionFunction typeInformation; - public PythonMapPartition(String id, TypeInformation typeInformation) { + public PythonMapPartition(int envId, int setId, TypeInformation typeInformation) { this.typeInformation = typeInformation; - streamer = new PythonStreamer(this, id, typeInformation instanceof PrimitiveArrayTypeInfo); + streamer = new PythonStreamer(this, envId, setId, typeInformation instanceof PrimitiveArrayTypeInfo); } /** 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 4b4ee77cb6df5d..2fe08e04c00971 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 @@ -51,7 +51,8 @@ public class PythonStreamer implements Serializable { private static final int SIGNAL_ERROR = -2; private static final byte SIGNAL_LAST = 32; - private final String id; + private final int envID; + private final int setID; private final boolean usePython3; private final String planArguments; @@ -73,8 +74,9 @@ public class PythonStreamer implements Serializable { protected final AbstractRichFunction function; - public PythonStreamer(AbstractRichFunction function, String id, boolean usesByteArray) { - this.id = id; + public PythonStreamer(AbstractRichFunction function, int envID, int setID, boolean usesByteArray) { + this.envID = envID; + this.setID = setID; this.usePython3 = PythonPlanBinder.usePython3; planArguments = PythonPlanBinder.arguments.toString(); sender = new PythonSender(); @@ -93,8 +95,8 @@ public void open() throws IOException { } private void startPython() throws IOException { - this.outputFilePath = FLINK_TMP_DATA_DIR + "/" + id + this.function.getRuntimeContext().getIndexOfThisSubtask() + "output"; - this.inputFilePath = FLINK_TMP_DATA_DIR + "/" + id + this.function.getRuntimeContext().getIndexOfThisSubtask() + "input"; + this.outputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "." + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "output"; + this.inputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "." + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "input"; sender.open(inputFilePath); receiver.open(outputFilePath); @@ -126,9 +128,6 @@ public void run() { Runtime.getRuntime().addShutdownHook(shutdownThread); - String envID = id.split("\\.")[0], - setID = id.split("\\.")[1]; - OutputStream processOutput = process.getOutputStream(); processOutput.write("operator\n".getBytes()); processOutput.write((envID + "\n").getBytes()); From 4370a1b848feae283072f093c53e857e371d6fac Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Mon, 30 Jan 2017 23:30:18 -0500 Subject: [PATCH 09/10] EnvironmentContainer class in Python API to simplify execution logic This new class keeps track of all of the execution environments currently created in the plan file, and handles the logic for fetching the next environment to be executed and for checking if a certain execution environment is the one that should be executed. --- .../python/api/flink/plan/Environment.py | 56 ++++++++++++------- 1 file changed, 36 insertions(+), 20 deletions(-) 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 570eac844a98af..89722ffc4b17f9 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 @@ -27,24 +27,48 @@ import sys from struct import pack -_env_counter = 0 -_last_env_id = -1 -_operating = False + +class EnvironmentContainer(object): + """Keeps track of which ExecutionEnvironment is being run.""" + + environment_counter = 0 + environment_id_to_execute = None + + def create_environment(self): + env = Environment(self, self.environment_counter) + self.environment_counter += 1 + return env + + def is_executing(self): + """Checks if we are waiting for a certain environment to be executed.""" + return not self.environment_id_to_execute is None + + def fetch_next_environment(self, calling_environment_id): + """Checks (and if necessary, fetches) the next environment to be executed.""" + if not self.is_executing(): + self.environment_id_to_execute = int(sys.stdin.readline().rstrip('\n')) + + if self.environment_id_to_execute == calling_environment_id: + self.environment_id_to_execute = None + return True + + return False + + +container = EnvironmentContainer() def get_environment(): """ Creates an execution environment that represents the context in which the program is currently executed. - + :return:The execution environment of the context in which the program is executed. """ - global _env_counter - id = _env_counter - _env_counter += 1 - return Environment(id) + global container + return container.create_environment() class Environment(object): - def __init__(self, env_id = 0): + def __init__(self, container, env_id): # util self._counter = 0 @@ -53,6 +77,7 @@ def __init__(self, env_id = 0): self._local_mode = False self._retry = 0 + self.container = container self.env_id = env_id #sets @@ -176,7 +201,7 @@ def execute(self, local=False): self._local_mode = local self._optimize_plan() - if _operating: + if self.container.is_executing(): plan_mode = False else: plan_mode = sys.stdin.readline().rstrip('\n') == "plan" @@ -194,16 +219,7 @@ def execute(self, local=False): import struct operator = None try: - env_id = _last_env_id - if not _operating: - env_id = int(sys.stdin.readline().rstrip('\n')) - - if self.env_id != env_id: - _last_env_id = env_id - _operating = True - else: - _operating = False - _last_env_id = -1 + if self.container.fetch_next_environment(self.env_id): id = int(sys.stdin.readline().rstrip('\n')) port = int(sys.stdin.readline().rstrip('\n')) From f8024829cdc180ce0e463de67b587a79377d250c Mon Sep 17 00:00:00 2001 From: Geoffrey Mon Date: Tue, 31 Jan 2017 08:12:53 -0500 Subject: [PATCH 10/10] Improvements and tweaks * Attribute and file name changes * Move plan/execute mode logic into ExecutionContainer * Send environment ID with execution parameters * Other minor changes --- .../flink/python/api/PythonPlanBinder.java | 14 +++++---- .../api/streaming/data/PythonStreamer.java | 4 +-- .../python/api/flink/plan/Environment.py | 30 +++++++++---------- 3 files changed, 24 insertions(+), 24 deletions(-) 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 e03eb3ca94c826..a3ede99486ee45 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 @@ -95,7 +95,7 @@ public class PythonPlanBinder { private HashMap sets = new HashMap<>(); public ExecutionEnvironment env; - private int environmentCounter = 0; + private int currentEnvironmentID = 0; private PythonPlanStreamer streamer; public static final int MAPPED_FILE_SIZE = 1024 * 1024 * 64; @@ -156,8 +156,6 @@ private void runPlan(String[] args) throws Exception { distributeFiles(tmpPath, env); JobExecutionResult jer = env.execute(); sendResult(jer); - - environmentCounter++; } clearPath(tmpPath); @@ -261,11 +259,12 @@ private void receivePlan() throws IOException { private enum Parameters { DOP, MODE, - RETRY + RETRY, + ID } private void receiveParameters() throws IOException { - for (int x = 0; x < 3; x++) { + for (int x = 0; x < 4; x++) { Tuple value = (Tuple) streamer.getRecord(true); switch (Parameters.valueOf(((String) value.getField(0)).toUpperCase())) { case DOP: @@ -279,6 +278,9 @@ private void receiveParameters() throws IOException { int retry = (Integer) value.getField(1); env.setRestartStrategy(RestartStrategies.fixedDelayRestart(retry, 10000L)); break; + case ID: + currentEnvironmentID = (Integer) value.getField(1); + break; } } if (env.getParallelism() < 0) { @@ -301,7 +303,7 @@ protected enum Operation { private void receiveOperations() throws IOException { Integer operationCount = (Integer) streamer.getRecord(true); for (int x = 0; x < operationCount; x++) { - PythonOperationInfo info = new PythonOperationInfo(streamer, environmentCounter); + PythonOperationInfo info = new PythonOperationInfo(streamer, currentEnvironmentID); Operation op; try { op = Operation.valueOf(info.identifier.toUpperCase()); 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 2fe08e04c00971..765f5c13aeae0a 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 @@ -95,8 +95,8 @@ public void open() throws IOException { } private void startPython() throws IOException { - this.outputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "." + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "output"; - this.inputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "." + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "input"; + this.outputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "_" + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "output"; + this.inputFilePath = FLINK_TMP_DATA_DIR + "/" + envID + "_" + setID + this.function.getRuntimeContext().getIndexOfThisSubtask() + "input"; sender.open(inputFilePath); receiver.open(outputFilePath); 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 89722ffc4b17f9..3cff79d41494fd 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 @@ -29,23 +29,27 @@ class EnvironmentContainer(object): - """Keeps track of which ExecutionEnvironment is being run.""" + """Keeps track of which ExecutionEnvironment is active.""" environment_counter = 0 environment_id_to_execute = None def create_environment(self): + """Assign each new environment a unique ID.""" env = Environment(self, self.environment_counter) self.environment_counter += 1 return env - def is_executing(self): - """Checks if we are waiting for a certain environment to be executed.""" - return not self.environment_id_to_execute is None + def is_planning(self): + """ + Checks if we are waiting for a certain environment to be executed. + If not, grabs the next mode (plan or operation execution). + """ + return self.environment_id_to_execute is None and sys.stdin.readline().rstrip('\n') == "plan" def fetch_next_environment(self, calling_environment_id): """Checks (and if necessary, fetches) the next environment to be executed.""" - if not self.is_executing(): + if self.environment_id_to_execute is None: self.environment_id_to_execute = int(sys.stdin.readline().rstrip('\n')) if self.environment_id_to_execute == calling_environment_id: @@ -63,7 +67,6 @@ def get_environment(): :return:The execution environment of the context in which the program is executed. """ - global container return container.create_environment() @@ -77,8 +80,8 @@ def __init__(self, container, env_id): self._local_mode = False self._retry = 0 - self.container = container - self.env_id = env_id + self._container = container + self._env_id = env_id #sets self._sources = [] @@ -197,16 +200,10 @@ def execute(self, local=False): The environment will execute all parts of the program that have resulted in a "sink" operation. """ - global _operating, _last_env_id self._local_mode = local self._optimize_plan() - if self.container.is_executing(): - plan_mode = False - else: - plan_mode = sys.stdin.readline().rstrip('\n') == "plan" - - if plan_mode: + if self._container.is_planning(): port = int(sys.stdin.readline().rstrip('\n')) self._connection = Connection.PureTCPConnection(port) self._iterator = Iterator.PlanIterator(self._connection, self) @@ -219,7 +216,7 @@ def execute(self, local=False): import struct operator = None try: - if self.container.fetch_next_environment(self.env_id): + if self._container.fetch_next_environment(self._env_id): id = int(sys.stdin.readline().rstrip('\n')) port = int(sys.stdin.readline().rstrip('\n')) @@ -317,6 +314,7 @@ def _send_parameters(self): collect(("dop", self._dop)) collect(("mode", self._local_mode)) collect(("retry", self._retry)) + collect(("id", self._env_id)) def _send_operations(self): self._collector.collect(len(self._sources) + len(self._sets) + len(self._sinks) + len(self._broadcast))