You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2015/07/30 19:11:20 UTC
[3/3] flink git commit: [FLINK-1927][FLINK-2173][py] Operator
distribution rework, fix file paths
[FLINK-1927][FLINK-2173][py] Operator distribution rework, fix file paths
Python operators are no longer serialized and shipped across the
cluster. Instead the plan file is executed on each node, followed by
usage of the respective operator object.
- removed dill library
- filepaths are always explicitly passed to python
- fix error reporting
This closes #931.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/68b15593
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/68b15593
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/68b15593
Branch: refs/heads/master
Commit: 68b1559319a870c3120263c5d10967ff6b690682
Parents: 1919ae7
Author: zentol <s....@web.de>
Authored: Tue Jul 21 21:22:19 2015 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Thu Jul 30 19:09:12 2015 +0200
----------------------------------------------------------------------
.../api/java/common/streaming/Receiver.java | 8 +-
.../api/java/common/streaming/Sender.java | 6 +-
.../api/java/python/PythonPlanBinder.java | 82 +-
.../java/python/functions/PythonCoGroup.java | 4 +-
.../python/functions/PythonCombineIdentity.java | 4 +-
.../python/functions/PythonMapPartition.java | 4 +-
.../java/python/streaming/PythonStreamer.java | 66 +-
.../languagebinding/api/python/dill/__diff.py | 229 ----
.../languagebinding/api/python/dill/__init__.py | 73 --
.../languagebinding/api/python/dill/_objects.py | 530 ---------
.../languagebinding/api/python/dill/detect.py | 222 ----
.../languagebinding/api/python/dill/dill.py | 1034 ------------------
.../languagebinding/api/python/dill/objtypes.py | 27 -
.../languagebinding/api/python/dill/pointers.py | 122 ---
.../languagebinding/api/python/dill/source.py | 1010 -----------------
.../languagebinding/api/python/dill/temp.py | 236 ----
.../languagebinding/api/python/executor.py | 54 -
.../api/python/flink/connection/Collector.py | 4 -
.../api/python/flink/connection/Connection.py | 14 +-
.../api/python/flink/functions/Function.py | 13 +-
.../api/python/flink/plan/Environment.py | 70 +-
pom.xml | 2 -
22 files changed, 110 insertions(+), 3704 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java b/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
index 2741714..23720d7 100644
--- a/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
+++ b/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
@@ -50,11 +50,7 @@ public class Receiver implements Serializable {
setupMappedFile(path);
}
- private void setupMappedFile(String path) throws FileNotFoundException, IOException {
- String inputFilePath = function == null
- ? FLINK_TMP_DATA_DIR + "/" + "output"
- : path;
-
+ private void setupMappedFile(String inputFilePath) throws FileNotFoundException, IOException {
File x = new File(FLINK_TMP_DATA_DIR);
x.mkdirs();
@@ -100,7 +96,7 @@ public class Receiver implements Serializable {
count++;
}
if (fileBuffer.get(0) == 0) {
- throw new RuntimeException("External process not respoonding.");
+ throw new RuntimeException("External process not responding.");
}
fileBuffer.position(1);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Sender.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Sender.java b/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Sender.java
index d45b019..3e0c317 100644
--- a/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Sender.java
+++ b/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Sender.java
@@ -65,11 +65,7 @@ public class Sender implements Serializable {
setupMappedFile(path);
}
- private void setupMappedFile(String path) throws FileNotFoundException, IOException {
- String outputFilePath = function == null
- ? FLINK_TMP_DATA_DIR + "/" + "input"
- : path;
-
+ private void setupMappedFile(String outputFilePath) throws FileNotFoundException, IOException {
File x = new File(FLINK_TMP_DATA_DIR);
x.mkdirs();
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/PythonPlanBinder.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/PythonPlanBinder.java b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/PythonPlanBinder.java
index c278f5c..7c30418 100644
--- a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/PythonPlanBinder.java
+++ b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/PythonPlanBinder.java
@@ -52,7 +52,6 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
public static final String FLINK_PYTHON_DC_ID = "flink";
public static final String FLINK_PYTHON_PLAN_NAME = "/plan.py";
- public static final String FLINK_PYTHON_EXECUTOR_NAME = "/executor.py";
public static final String FLINK_PYTHON2_BINARY_KEY = "python.binary.python2";
public static final String FLINK_PYTHON3_BINARY_KEY = "python.binary.python3";
@@ -64,6 +63,8 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
protected static final String FLINK_DIR = System.getenv("FLINK_ROOT_DIR");
protected static String FULL_PATH;
+ public static StringBuilder arguments = new StringBuilder();
+
private Process process;
public static boolean usePython3 = false;
@@ -172,12 +173,11 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
}
private void startPython(String[] args) throws IOException {
- StringBuilder argsBuilder = new StringBuilder();
for (String arg : args) {
- argsBuilder.append(" ").append(arg);
+ arguments.append(" ").append(arg);
}
receiver = new Receiver(null);
- receiver.open(null);
+ receiver.open(FLINK_TMP_DATA_DIR + "/output");
String pythonBinaryPath = usePython3 ? FLINK_PYTHON3_BINARY_PATH : FLINK_PYTHON2_BINARY_PATH;
@@ -186,7 +186,7 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
} catch (IOException ex) {
throw new RuntimeException(pythonBinaryPath + " does not point to a valid python binary.");
}
- process = Runtime.getRuntime().exec(pythonBinaryPath + " -B " + FLINK_PYTHON_FILE_PATH + FLINK_PYTHON_PLAN_NAME + argsBuilder.toString());
+ process = Runtime.getRuntime().exec(pythonBinaryPath + " -B " + FLINK_PYTHON_FILE_PATH + FLINK_PYTHON_PLAN_NAME + arguments.toString());
new StreamPrinter(process.getInputStream()).start();
new StreamPrinter(process.getErrorStream()).start();
@@ -201,8 +201,15 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
if (value != 0) {
throw new RuntimeException("Plan file caused an error. Check log-files for details.");
}
+ if (value == 0) {
+ throw new RuntimeException("Plan file exited prematurely without an error.");
+ }
} catch (IllegalThreadStateException ise) {//Process still running
}
+
+ process.getOutputStream().write("plan\n".getBytes());
+ process.getOutputStream().write((FLINK_TMP_DATA_DIR + "/output\n").getBytes());
+ process.getOutputStream().flush();
}
private void close() {
@@ -231,10 +238,7 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
//=====Plan Binding=================================================================================================
protected class PythonOperationInfo extends OperationInfo {
- protected byte[] operator;
- protected String meta;
protected boolean combine;
- protected byte[] combineOperator;
protected String name;
@Override
@@ -244,11 +248,8 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
sb.append("ParentID: ").append(parentID).append("\n");
sb.append("OtherID: ").append(otherID).append("\n");
sb.append("Name: ").append(name).append("\n");
- sb.append("Operator: ").append(operator == null ? null : "<operator>").append("\n");
- sb.append("Meta: ").append(meta).append("\n");
sb.append("Types: ").append(types).append("\n");
sb.append("Combine: ").append(combine).append("\n");
- sb.append("CombineOP: ").append(combineOperator == null ? null : "<combineop>").append("\n");
sb.append("Keys1: ").append(Arrays.toString(keys1)).append("\n");
sb.append("Keys2: ").append(Arrays.toString(keys2)).append("\n");
sb.append("Projections: ").append(Arrays.toString(projections)).append("\n");
@@ -264,8 +265,6 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
otherID = (Integer) receiver.getRecord(true);
keys1 = tupleToIntArray((Tuple) receiver.getRecord(true));
keys2 = tupleToIntArray((Tuple) receiver.getRecord(true));
- operator = (byte[]) receiver.getRecord();
- meta = (String) receiver.getRecord();
tmpType = receiver.getRecord();
types = tmpType == null ? null : getForObject(tmpType);
name = (String) receiver.getRecord();
@@ -274,8 +273,6 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
case CROSS_H:
case CROSS_T:
otherID = (Integer) receiver.getRecord(true);
- operator = (byte[]) receiver.getRecord();
- meta = (String) receiver.getRecord();
tmpType = receiver.getRecord();
types = tmpType == null ? null : getForObject(tmpType);
int cProjectCount = (Integer) receiver.getRecord(true);
@@ -289,9 +286,6 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
break;
case REDUCE:
case GROUPREDUCE:
- operator = (byte[]) receiver.getRecord();
- combineOperator = (byte[]) receiver.getRecord();
- meta = (String) receiver.getRecord();
tmpType = receiver.getRecord();
types = tmpType == null ? null : getForObject(tmpType);
combine = (Boolean) receiver.getRecord();
@@ -303,8 +297,6 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
keys1 = tupleToIntArray((Tuple) receiver.getRecord(true));
keys2 = tupleToIntArray((Tuple) receiver.getRecord(true));
otherID = (Integer) receiver.getRecord(true);
- operator = (byte[]) receiver.getRecord();
- meta = (String) receiver.getRecord();
tmpType = receiver.getRecord();
types = tmpType == null ? null : getForObject(tmpType);
int jProjectCount = (Integer) receiver.getRecord(true);
@@ -320,8 +312,6 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
case FLATMAP:
case MAP:
case FILTER:
- operator = (byte[]) receiver.getRecord();
- meta = (String) receiver.getRecord();
tmpType = receiver.getRecord();
types = tmpType == null ? null : getForObject(tmpType);
name = (String) receiver.getRecord();
@@ -344,7 +334,7 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
op2,
new Keys.ExpressionKeys(firstKeys, op1.getType()),
new Keys.ExpressionKeys(secondKeys, op2.getType()),
- new PythonCoGroup(info.setID, info.operator, info.types, info.meta),
+ new PythonCoGroup(info.setID, info.types),
info.types, info.name);
}
@@ -353,13 +343,13 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
switch (mode) {
case NONE:
return op1.cross(op2).name("PythonCrossPreStep")
- .mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ .mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
case HUGE:
return op1.crossWithHuge(op2).name("PythonCrossPreStep")
- .mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ .mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
case TINY:
return op1.crossWithTiny(op2).name("PythonCrossPreStep")
- .mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ .mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
default:
throw new IllegalArgumentException("Invalid Cross mode specified: " + mode);
}
@@ -367,25 +357,25 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
@Override
protected DataSet applyFilterOperation(DataSet op1, PythonOperationInfo info) {
- return op1.mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ return op1.mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
}
@Override
protected DataSet applyFlatMapOperation(DataSet op1, PythonOperationInfo info) {
- return op1.mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ return op1.mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
}
@Override
protected DataSet applyGroupReduceOperation(DataSet op1, PythonOperationInfo info) {
if (info.combine) {
- return op1.reduceGroup(new PythonCombineIdentity(info.setID, info.combineOperator, info.meta))
+ return op1.reduceGroup(new PythonCombineIdentity(info.setID * -1))
.setCombinable(true).name("PythonCombine")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
} else {
return op1.reduceGroup(new PythonCombineIdentity())
.setCombinable(false).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
}
}
@@ -393,14 +383,14 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
@Override
protected DataSet applyGroupReduceOperation(UnsortedGrouping op1, PythonOperationInfo info) {
if (info.combine) {
- return op1.reduceGroup(new PythonCombineIdentity(info.setID, info.combineOperator, info.meta))
+ return op1.reduceGroup(new PythonCombineIdentity(info.setID * -1))
.setCombinable(true).name("PythonCombine")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
} else {
return op1.reduceGroup(new PythonCombineIdentity())
.setCombinable(false).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
}
}
@@ -408,14 +398,14 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
@Override
protected DataSet applyGroupReduceOperation(SortedGrouping op1, PythonOperationInfo info) {
if (info.combine) {
- return op1.reduceGroup(new PythonCombineIdentity(info.setID, info.combineOperator, info.meta))
+ return op1.reduceGroup(new PythonCombineIdentity(info.setID * -1))
.setCombinable(true).name("PythonCombine")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
} else {
return op1.reduceGroup(new PythonCombineIdentity())
.setCombinable(false).name("PythonGroupReducePreStep")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
}
}
@@ -425,13 +415,13 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
switch (mode) {
case NONE:
return op1.join(op2).where(firstKeys).equalTo(secondKeys).name("PythonJoinPreStep")
- .mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ .mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
case HUGE:
return op1.joinWithHuge(op2).where(firstKeys).equalTo(secondKeys).name("PythonJoinPreStep")
- .mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ .mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
case TINY:
return op1.joinWithTiny(op2).where(firstKeys).equalTo(secondKeys).name("PythonJoinPreStep")
- .mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ .mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
default:
throw new IllegalArgumentException("Invalid join mode specified.");
}
@@ -439,33 +429,33 @@ public class PythonPlanBinder extends PlanBinder<PythonOperationInfo> {
@Override
protected DataSet applyMapOperation(DataSet op1, PythonOperationInfo info) {
- return op1.mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ return op1.mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
}
@Override
protected DataSet applyMapPartitionOperation(DataSet op1, PythonOperationInfo info) {
- return op1.mapPartition(new PythonMapPartition(info.setID, info.operator, info.types, info.meta)).name(info.name);
+ return op1.mapPartition(new PythonMapPartition(info.setID, info.types)).name(info.name);
}
@Override
protected DataSet applyReduceOperation(DataSet op1, PythonOperationInfo info) {
return op1.reduceGroup(new PythonCombineIdentity())
.setCombinable(false).name("PythonReducePreStep")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
}
@Override
protected DataSet applyReduceOperation(UnsortedGrouping op1, PythonOperationInfo info) {
if (info.combine) {
- return op1.reduceGroup(new PythonCombineIdentity(info.setID, info.combineOperator, info.meta))
+ return op1.reduceGroup(new PythonCombineIdentity(info.setID * -1))
.setCombinable(true).name("PythonCombine")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
} else {
return op1.reduceGroup(new PythonCombineIdentity())
.setCombinable(false).name("PythonReducePreStep")
- .mapPartition(new PythonMapPartition(info.setID * -1, info.operator, info.types, info.meta))
+ .mapPartition(new PythonMapPartition(info.setID, info.types))
.name(info.name);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCoGroup.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCoGroup.java b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCoGroup.java
index 01f18eb..26d554d 100644
--- a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCoGroup.java
+++ b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCoGroup.java
@@ -31,9 +31,9 @@ public class PythonCoGroup<IN1, IN2, OUT> extends RichCoGroupFunction<IN1, IN2,
private final PythonStreamer streamer;
private transient final TypeInformation<OUT> typeInformation;
- public PythonCoGroup(int id, byte[] operator, TypeInformation<OUT> typeInformation, String metaInformation) {
+ public PythonCoGroup(int id, TypeInformation<OUT> typeInformation) {
this.typeInformation = typeInformation;
- streamer = new PythonStreamer(this, id, operator, metaInformation);
+ streamer = new PythonStreamer(this, id);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCombineIdentity.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCombineIdentity.java b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCombineIdentity.java
index 3395f07..a8ff96c 100644
--- a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCombineIdentity.java
+++ b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonCombineIdentity.java
@@ -31,8 +31,8 @@ public class PythonCombineIdentity<IN> extends RichGroupReduceFunction<IN, IN> {
streamer = null;
}
- public PythonCombineIdentity(int id, byte[] operator, String metaInformation) {
- streamer = new PythonStreamer(this, id, operator, metaInformation);
+ public PythonCombineIdentity(int id) {
+ streamer = new PythonStreamer(this, id);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonMapPartition.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonMapPartition.java b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonMapPartition.java
index f582e3d..1f13e5c 100644
--- a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonMapPartition.java
+++ b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/functions/PythonMapPartition.java
@@ -31,9 +31,9 @@ public class PythonMapPartition<IN, OUT> extends RichMapPartitionFunction<IN, OU
private final PythonStreamer streamer;
private transient final TypeInformation<OUT> typeInformation;
- public PythonMapPartition(int id, byte[] operator, TypeInformation<OUT> typeInformation, String metaInformation) {
+ public PythonMapPartition(int id, TypeInformation<OUT> typeInformation) {
this.typeInformation = typeInformation;
- streamer = new PythonStreamer(this, id, operator, metaInformation);
+ streamer = new PythonStreamer(this, id);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/streaming/PythonStreamer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/streaming/PythonStreamer.java b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/streaming/PythonStreamer.java
index 7dc2240..6d21c4c 100644
--- a/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/streaming/PythonStreamer.java
+++ b/flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/streaming/PythonStreamer.java
@@ -13,10 +13,10 @@
package org.apache.flink.languagebinding.api.java.python.streaming;
import java.io.IOException;
+import java.io.OutputStream;
import java.lang.reflect.Field;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import static org.apache.flink.languagebinding.api.java.common.PlanBinder.DEBUG;
-import static org.apache.flink.languagebinding.api.java.python.PythonPlanBinder.FLINK_PYTHON_EXECUTOR_NAME;
import static org.apache.flink.languagebinding.api.java.python.PythonPlanBinder.FLINK_PYTHON_DC_ID;
import static org.apache.flink.languagebinding.api.java.python.PythonPlanBinder.FLINK_PYTHON_PLAN_NAME;
import static org.apache.flink.languagebinding.api.java.common.PlanBinder.FLINK_TMP_DATA_DIR;
@@ -30,24 +30,22 @@ import static org.apache.flink.languagebinding.api.java.python.PythonPlanBinder.
* This streamer is used by functions to send/receive data to/from an external python process.
*/
public class PythonStreamer extends Streamer {
- private final byte[] operator;
private Process process;
- private final String metaInformation;
private final int id;
private final boolean usePython3;
private final boolean debug;
private Thread shutdownThread;
+ private final String planArguments;
private String inputFilePath;
private String outputFilePath;
- public PythonStreamer(AbstractRichFunction function, int id, byte[] operator, String metaInformation) {
+ public PythonStreamer(AbstractRichFunction function, int id) {
super(function);
- this.operator = operator;
- this.metaInformation = metaInformation;
this.id = id;
this.usePython3 = PythonPlanBinder.usePython3;
this.debug = DEBUG;
+ planArguments = PythonPlanBinder.arguments.toString();
}
/**
@@ -67,21 +65,8 @@ public class PythonStreamer extends Streamer {
sender.open(inputFilePath);
receiver.open(outputFilePath);
- ProcessBuilder pb = new ProcessBuilder();
-
String path = function.getRuntimeContext().getDistributedCache().getFile(FLINK_PYTHON_DC_ID).getAbsolutePath();
- String executorPath = path + FLINK_PYTHON_EXECUTOR_NAME;
- String[] frag = metaInformation.split("\\|");
- StringBuilder importString = new StringBuilder();
- if (frag[0].contains("__main__")) {
- importString.append("from ");
- importString.append(FLINK_PYTHON_PLAN_NAME.substring(1, FLINK_PYTHON_PLAN_NAME.length() - 3));
- importString.append(" import ");
- importString.append(frag[1]);
- } else {
- importString.append("import ");
- importString.append(FLINK_PYTHON_PLAN_NAME.substring(1, FLINK_PYTHON_PLAN_NAME.length() - 3));
- }
+ String planPath = path + FLINK_PYTHON_PLAN_NAME;
String pythonBinaryPath = usePython3 ? FLINK_PYTHON3_BINARY_PATH : FLINK_PYTHON2_BINARY_PATH;
@@ -90,14 +75,13 @@ public class PythonStreamer extends Streamer {
} catch (IOException ex) {
throw new RuntimeException(pythonBinaryPath + " does not point to a valid python binary.");
}
- pb.command(pythonBinaryPath, "-O", "-B", executorPath, "" + server.getLocalPort());
if (debug) {
socket.setSoTimeout(0);
LOG.info("Waiting for Python Process : " + function.getRuntimeContext().getTaskName()
- + " Run python /tmp/flink" + FLINK_PYTHON_EXECUTOR_NAME + " " + server.getLocalPort());
+ + " Run python " + planPath + planArguments);
} else {
- process = pb.start();
+ process = Runtime.getRuntime().exec(pythonBinaryPath + " -O -B " + planPath + planArguments);
new StreamPrinter(process.getInputStream()).start();
new StreamPrinter(process.getErrorStream(), true, msg).start();
}
@@ -114,31 +98,13 @@ public class PythonStreamer extends Streamer {
Runtime.getRuntime().addShutdownHook(shutdownThread);
- socket = server.accept();
- in = socket.getInputStream();
- out = socket.getOutputStream();
-
- byte[] opSize = new byte[4];
- putInt(opSize, 0, operator.length);
- out.write(opSize, 0, 4);
- out.write(operator, 0, operator.length);
-
- byte[] meta = importString.toString().getBytes("utf-8");
- putInt(opSize, 0, meta.length);
- out.write(opSize, 0, 4);
- out.write(meta, 0, meta.length);
-
- byte[] input = inputFilePath.getBytes("utf-8");
- putInt(opSize, 0, input.length);
- out.write(opSize, 0, 4);
- out.write(input, 0, input.length);
-
- byte[] output = outputFilePath.getBytes("utf-8");
- putInt(opSize, 0, output.length);
- out.write(opSize, 0, 4);
- out.write(output, 0, output.length);
-
- out.flush();
+ OutputStream processOutput = process.getOutputStream();
+ processOutput.write("operator\n".getBytes());
+ processOutput.write(("" + server.getLocalPort() + "\n").getBytes());
+ processOutput.write((id + "\n").getBytes());
+ processOutput.write((inputFilePath + "\n").getBytes());
+ processOutput.write((outputFilePath + "\n").getBytes());
+ processOutput.flush();
try { // wait a bit to catch syntax errors
Thread.sleep(2000);
@@ -151,6 +117,10 @@ public class PythonStreamer extends Streamer {
} catch (IllegalThreadStateException ise) { //process still active -> start receiving data
}
}
+
+ socket = server.accept();
+ in = socket.getInputStream();
+ out = socket.getOutputStream();
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__diff.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__diff.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__diff.py
deleted file mode 100644
index 79301a6..0000000
--- a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__diff.py
+++ /dev/null
@@ -1,229 +0,0 @@
-#!/usr/bin/env python
-#
-# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
-# Copyright (c) 2008-2014 California Institute of Technology.
-# License: 3-clause BSD. The full license text is available at:
-# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
-
-"""
-Module to show if an object has changed since it was memorised
-"""
-
-import os
-import sys
-import types
-try:
- import numpy
- HAS_NUMPY = True
-except:
- HAS_NUMPY = False
-try:
- import builtins
-except ImportError:
- import __builtin__ as builtins
-
-# memo of objects indexed by id to a tuple (attributes, sequence items)
-# attributes is a dict indexed by attribute name to attribute id
-# sequence items is either a list of ids, of a dictionary of keys to ids
-memo = {}
-id_to_obj = {}
-# types that cannot have changing attributes
-builtins_types = set((str, list, dict, set, frozenset, int))
-dont_memo = set(id(i) for i in (memo, sys.modules, sys.path_importer_cache,
- os.environ, id_to_obj))
-
-
-def get_attrs(obj):
- """
- Gets all the attributes of an object though its __dict__ or return None
- """
- if type(obj) in builtins_types \
- or type(obj) is type and obj in builtins_types:
- return
- try:
- return obj.__dict__
- except:
- return
-
-
-def get_seq(obj, cache={str: False, frozenset: False, list: True, set: True,
- dict: True, tuple: True, type: False,
- types.ModuleType: False, types.FunctionType: False,
- types.BuiltinFunctionType: False}):
- """
- Gets all the items in a sequence or return None
- """
- o_type = type(obj)
- hsattr = hasattr
- if o_type in cache:
- if cache[o_type]:
- if hsattr(obj, "copy"):
- return obj.copy()
- return obj
- elif HAS_NUMPY and o_type in (numpy.ndarray, numpy.ma.core.MaskedConstant):
- if obj.shape and obj.size:
- return obj
- else:
- return []
- elif hsattr(obj, "__contains__") and hsattr(obj, "__iter__") \
- and hsattr(obj, "__len__") and hsattr(o_type, "__contains__") \
- and hsattr(o_type, "__iter__") and hsattr(o_type, "__len__"):
- cache[o_type] = True
- if hsattr(obj, "copy"):
- return obj.copy()
- return obj
- else:
- cache[o_type] = False
- return None
-
-
-def memorise(obj, force=False):
- """
- Adds an object to the memo, and recursively adds all the objects
- attributes, and if it is a container, its items. Use force=True to update
- an object already in the memo. Updating is not recursively done.
- """
- obj_id = id(obj)
- if obj_id in memo and not force or obj_id in dont_memo:
- return
- id_ = id
- g = get_attrs(obj)
- if g is None:
- attrs_id = None
- else:
- attrs_id = dict((key,id_(value)) for key, value in g.items())
-
- s = get_seq(obj)
- if s is None:
- seq_id = None
- elif hasattr(s, "items"):
- seq_id = dict((id_(key),id_(value)) for key, value in s.items())
- else:
- seq_id = [id_(i) for i in s]
-
- memo[obj_id] = attrs_id, seq_id
- id_to_obj[obj_id] = obj
- mem = memorise
- if g is not None:
- [mem(value) for key, value in g.items()]
-
- if s is not None:
- if hasattr(s, "items"):
- [(mem(key), mem(item))
- for key, item in s.items()]
- else:
- [mem(item) for item in s]
-
-
-def release_gone():
- itop, mp, src = id_to_obj.pop, memo.pop, sys.getrefcount
- [(itop(id_), mp(id_)) for id_, obj in list(id_to_obj.items())
- if src(obj) < 4]
-
-
-def whats_changed(obj, seen=None, simple=False, first=True):
- """
- Check an object against the memo. Returns a list in the form
- (attribute changes, container changed). Attribute changes is a dict of
- attribute name to attribute value. container changed is a boolean.
- If simple is true, just returns a boolean. None for either item means
- that it has not been checked yet
- """
- # Special cases
- if first:
- # ignore the _ variable, which only appears in interactive sessions
- if "_" in builtins.__dict__:
- del builtins._
- if seen is None:
- seen = {}
-
- obj_id = id(obj)
-
- if obj_id in seen:
- if simple:
- return any(seen[obj_id])
- return seen[obj_id]
-
- # Safety checks
- if obj_id in dont_memo:
- seen[obj_id] = [{}, False]
- if simple:
- return False
- return seen[obj_id]
- elif obj_id not in memo:
- if simple:
- return True
- else:
- raise RuntimeError("Object not memorised " + str(obj))
-
- seen[obj_id] = ({}, False)
-
- chngd = whats_changed
- id_ = id
-
- # compare attributes
- attrs = get_attrs(obj)
- if attrs is None:
- changed = {}
- else:
- obj_attrs = memo[obj_id][0]
- obj_get = obj_attrs.get
- changed = dict((key,None) for key in obj_attrs if key not in attrs)
- for key, o in attrs.items():
- if id_(o) != obj_get(key, None) or chngd(o, seen, True, False):
- changed[key] = o
-
- # compare sequence
- items = get_seq(obj)
- seq_diff = False
- if items is not None:
- obj_seq = memo[obj_id][1]
- if len(items) != len(obj_seq):
- seq_diff = True
- elif hasattr(obj, "items"): # dict type obj
- obj_get = obj_seq.get
- for key, item in items.items():
- if id_(item) != obj_get(id_(key)) \
- or chngd(key, seen, True, False) \
- or chngd(item, seen, True, False):
- seq_diff = True
- break
- else:
- for i, j in zip(items, obj_seq): # list type obj
- if id_(i) != j or chngd(i, seen, True, False):
- seq_diff = True
- break
- seen[obj_id] = changed, seq_diff
- if simple:
- return changed or seq_diff
- return changed, seq_diff
-
-
-def has_changed(*args, **kwds):
- kwds['simple'] = True # ignore simple if passed in
- return whats_changed(*args, **kwds)
-
-__import__ = __import__
-
-
-def _imp(*args, **kwds):
- """
- Replaces the default __import__, to allow a module to be memorised
- before the user can change it
- """
- before = set(sys.modules.keys())
- mod = __import__(*args, **kwds)
- after = set(sys.modules.keys()).difference(before)
- for m in after:
- memorise(sys.modules[m])
- return mod
-
-builtins.__import__ = _imp
-if hasattr(builtins, "_"):
- del builtins._
-
-# memorise all already imported modules. This implies that this must be
-# imported first for any changes to be recorded
-for mod in sys.modules.values():
- memorise(mod)
-release_gone()
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__init__.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__init__.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__init__.py
deleted file mode 100644
index b03eda9..0000000
--- a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/__init__.py
+++ /dev/null
@@ -1,73 +0,0 @@
-#!/usr/bin/env python
-#
-# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
-# Copyright (c) 2008-2014 California Institute of Technology.
-# License: 3-clause BSD. The full license text is available at:
-# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
-
-from __future__ import absolute_import
-
-from .dill import dump, dumps, load, loads, dump_session, load_session, \
- Pickler, Unpickler, register, copy, pickle, pickles, \
- HIGHEST_PROTOCOL, DEFAULT_PROTOCOL, PicklingError, UnpicklingError, \
- HANDLE_FMODE, CONTENTS_FMODE, FILE_FMODE
-from . import source, temp, detect
-
-# make sure "trace" is turned off
-detect.trace(False)
-
-try:
- from imp import reload
-except ImportError:
- pass
-
-# put the objects in order, if possible
-try:
- from collections import OrderedDict as odict
-except ImportError:
- try:
- from ordereddict import OrderedDict as odict
- except ImportError:
- odict = dict
-objects = odict()
-# local import of dill._objects
-#from . import _objects
-#objects.update(_objects.succeeds)
-#del _objects
-
-# local import of dill.objtypes
-from . import objtypes as types
-
-def load_types(pickleable=True, unpickleable=True):
- """load pickleable and/or unpickleable types to dill.types"""
- # local import of dill.objects
- from . import _objects
- if pickleable:
- objects.update(_objects.succeeds)
- else:
- [objects.pop(obj,None) for obj in _objects.succeeds]
- if unpickleable:
- objects.update(_objects.failures)
- else:
- [objects.pop(obj,None) for obj in _objects.failures]
- objects.update(_objects.registered)
- del _objects
- # reset contents of types to 'empty'
- [types.__dict__.pop(obj) for obj in list(types.__dict__.keys()) \
- if obj.find('Type') != -1]
- # add corresponding types from objects to types
- reload(types)
-
-def extend(use_dill=True):
- '''add (or remove) dill types to/from pickle'''
- from .dill import _revert_extension, _extend
- if use_dill: _extend()
- else: _revert_extension()
- return
-
-extend()
-
-del absolute_import
-del odict
-
-# end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/_objects.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/_objects.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/_objects.py
deleted file mode 100644
index b89bc0e..0000000
--- a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/_objects.py
+++ /dev/null
@@ -1,530 +0,0 @@
-#!/usr/bin/env python
-#
-# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
-# Copyright (c) 2008-2014 California Institute of Technology.
-# License: 3-clause BSD. The full license text is available at:
-# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
-"""
-all Python Standard Library objects (currently: CH 1-15 @ 2.7)
-and some other common objects (i.e. numpy.ndarray)
-"""
-
-__all__ = ['registered','failures','succeeds']
-
-# helper imports
-import warnings; warnings.filterwarnings("ignore", category=DeprecationWarning)
-import sys
-PY3 = (hex(sys.hexversion) >= '0x30000f0')
-if PY3:
- import queue as Queue
- import dbm as anydbm
-else:
- import Queue
- import anydbm
- import sets # deprecated/removed
- import mutex # removed
-try:
- from cStringIO import StringIO # has StringI and StringO types
-except ImportError: # only has StringIO type
- if PY3:
- from io import BytesIO as StringIO
- else:
- from StringIO import StringIO
-import re
-import array
-import collections
-import codecs
-import struct
-import datetime
-import calendar
-import weakref
-import pprint
-import decimal
-import functools
-import itertools
-import operator
-import tempfile
-import shelve
-import zlib
-import gzip
-import zipfile
-import tarfile
-import xdrlib
-import csv
-import hashlib
-import hmac
-import os
-import logging
-import optparse
-import curses
-#import __hello__
-import threading
-import socket
-import contextlib
-try:
- import bz2
- import sqlite3
- if PY3: import dbm.ndbm as dbm
- else: import dbm
- HAS_ALL = True
-except ImportError: # Ubuntu
- HAS_ALL = False
-try:
- import ctypes
- HAS_CTYPES = True
-except ImportError: # MacPorts
- HAS_CTYPES = False
-from curses import textpad, panel
-
-# helper objects
-class _class:
- def _method(self):
- pass
-# @classmethod
-# def _clsmethod(cls): #XXX: test me
-# pass
-# @staticmethod
-# def _static(self): #XXX: test me
-# pass
-class _class2:
- def __call__(self):
- pass
-_instance2 = _class2()
-class _newclass(object):
- def _method(self):
- pass
-# @classmethod
-# def _clsmethod(cls): #XXX: test me
-# pass
-# @staticmethod
-# def _static(self): #XXX: test me
-# pass
-def _function(x): yield x
-def _function2():
- try: raise
- except:
- from sys import exc_info
- e, er, tb = exc_info()
- return er, tb
-if HAS_CTYPES:
- class _Struct(ctypes.Structure):
- pass
- _Struct._fields_ = [("_field", ctypes.c_int),("next", ctypes.POINTER(_Struct))]
-_filedescrip, _tempfile = tempfile.mkstemp('r') # deleted in cleanup
-_tmpf = tempfile.TemporaryFile('w')
-
-# put the objects in order, if possible
-try:
- from collections import OrderedDict as odict
-except ImportError:
- try:
- from ordereddict import OrderedDict as odict
- except ImportError:
- odict = dict
-# objects used by dill for type declaration
-registered = d = odict()
-# objects dill fails to pickle
-failures = x = odict()
-# all other type objects
-succeeds = a = odict()
-
-# types module (part of CH 8)
-a['BooleanType'] = bool(1)
-a['BuiltinFunctionType'] = len
-a['BuiltinMethodType'] = a['BuiltinFunctionType']
-a['BytesType'] = _bytes = codecs.latin_1_encode('\x00')[0] # bytes(1)
-a['ClassType'] = _class
-a['ComplexType'] = complex(1)
-a['DictType'] = _dict = {}
-a['DictionaryType'] = a['DictType']
-a['FloatType'] = float(1)
-a['FunctionType'] = _function
-a['InstanceType'] = _instance = _class()
-a['IntType'] = _int = int(1)
-a['ListType'] = _list = []
-a['NoneType'] = None
-a['ObjectType'] = object()
-a['StringType'] = _str = str(1)
-a['TupleType'] = _tuple = ()
-a['TypeType'] = type
-if PY3:
- a['LongType'] = _int
- a['UnicodeType'] = _str
-else:
- a['LongType'] = long(1)
- a['UnicodeType'] = unicode(1)
-# built-in constants (CH 4)
-a['CopyrightType'] = copyright
-# built-in types (CH 5)
-a['ClassObjectType'] = _newclass # <type 'type'>
-a['ClassInstanceType'] = _newclass() # <type 'class'>
-a['SetType'] = _set = set()
-a['FrozenSetType'] = frozenset()
-# built-in exceptions (CH 6)
-a['ExceptionType'] = _exception = _function2()[0]
-# string services (CH 7)
-a['SREPatternType'] = _srepattern = re.compile('')
-# data types (CH 8)
-a['ArrayType'] = array.array("f")
-a['DequeType'] = collections.deque([0])
-a['DefaultDictType'] = collections.defaultdict(_function, _dict)
-a['TZInfoType'] = datetime.tzinfo()
-a['DateTimeType'] = datetime.datetime.today()
-a['CalendarType'] = calendar.Calendar()
-if not PY3:
- a['SetsType'] = sets.Set()
- a['ImmutableSetType'] = sets.ImmutableSet()
- a['MutexType'] = mutex.mutex()
-# numeric and mathematical types (CH 9)
-a['DecimalType'] = decimal.Decimal(1)
-a['CountType'] = itertools.count(0)
-# data compression and archiving (CH 12)
-a['TarInfoType'] = tarfile.TarInfo()
-# generic operating system services (CH 15)
-a['LoggerType'] = logging.getLogger()
-a['FormatterType'] = logging.Formatter() # pickle ok
-a['FilterType'] = logging.Filter() # pickle ok
-a['LogRecordType'] = logging.makeLogRecord(_dict) # pickle ok
-a['OptionParserType'] = _oparser = optparse.OptionParser() # pickle ok
-a['OptionGroupType'] = optparse.OptionGroup(_oparser,"foo") # pickle ok
-a['OptionType'] = optparse.Option('--foo') # pickle ok
-if HAS_CTYPES:
- a['CCharType'] = _cchar = ctypes.c_char()
- a['CWCharType'] = ctypes.c_wchar() # fail == 2.6
- a['CByteType'] = ctypes.c_byte()
- a['CUByteType'] = ctypes.c_ubyte()
- a['CShortType'] = ctypes.c_short()
- a['CUShortType'] = ctypes.c_ushort()
- a['CIntType'] = ctypes.c_int()
- a['CUIntType'] = ctypes.c_uint()
- a['CLongType'] = ctypes.c_long()
- a['CULongType'] = ctypes.c_ulong()
- a['CLongLongType'] = ctypes.c_longlong()
- a['CULongLongType'] = ctypes.c_ulonglong()
- a['CFloatType'] = ctypes.c_float()
- a['CDoubleType'] = ctypes.c_double()
- a['CSizeTType'] = ctypes.c_size_t()
- a['CLibraryLoaderType'] = ctypes.cdll
- a['StructureType'] = _Struct
- a['BigEndianStructureType'] = ctypes.BigEndianStructure()
-#NOTE: also LittleEndianStructureType and UnionType... abstract classes
-#NOTE: remember for ctypesobj.contents creates a new python object
-#NOTE: ctypes.c_int._objects is memberdescriptor for object's __dict__
-#NOTE: base class of all ctypes data types is non-public _CData
-
-try: # python 2.6
- import fractions
- import number
- import io
- from io import StringIO as TextIO
- # built-in functions (CH 2)
- a['ByteArrayType'] = bytearray([1])
- # numeric and mathematical types (CH 9)
- a['FractionType'] = fractions.Fraction()
- a['NumberType'] = numbers.Number()
- # generic operating system services (CH 15)
- a['IOBaseType'] = io.IOBase()
- a['RawIOBaseType'] = io.RawIOBase()
- a['TextIOBaseType'] = io.TextIOBase()
- a['BufferedIOBaseType'] = io.BufferedIOBase()
- a['UnicodeIOType'] = TextIO() # the new StringIO
- a['LoggingAdapterType'] = logging.LoggingAdapter(_logger,_dict) # pickle ok
- if HAS_CTYPES:
- a['CBoolType'] = ctypes.c_bool(1)
- a['CLongDoubleType'] = ctypes.c_longdouble()
-except ImportError:
- pass
-try: # python 2.7
- import argparse
- # data types (CH 8)
- a['OrderedDictType'] = collections.OrderedDict(_dict)
- a['CounterType'] = collections.Counter(_dict)
- if HAS_CTYPES:
- a['CSSizeTType'] = ctypes.c_ssize_t()
- # generic operating system services (CH 15)
- a['NullHandlerType'] = logging.NullHandler() # pickle ok # new 2.7
- a['ArgParseFileType'] = argparse.FileType() # pickle ok
-#except AttributeError:
-except ImportError:
- pass
-
-# -- pickle fails on all below here -----------------------------------------
-# types module (part of CH 8)
-a['CodeType'] = compile('','','exec')
-a['DictProxyType'] = type.__dict__
-a['DictProxyType2'] = _newclass.__dict__
-a['EllipsisType'] = Ellipsis
-a['ClosedFileType'] = open(os.devnull, 'wb', buffering=0).close()
-a['GetSetDescriptorType'] = array.array.typecode
-a['LambdaType'] = _lambda = lambda x: lambda y: x #XXX: works when not imported!
-a['MemberDescriptorType'] = type.__dict__['__weakrefoffset__']
-a['MemberDescriptorType2'] = datetime.timedelta.days
-a['MethodType'] = _method = _class()._method #XXX: works when not imported!
-a['ModuleType'] = datetime
-a['NotImplementedType'] = NotImplemented
-a['SliceType'] = slice(1)
-a['UnboundMethodType'] = _class._method #XXX: works when not imported!
-a['TextWrapperType'] = open(os.devnull, 'r') # same as mode='w','w+','r+'
-a['BufferedRandomType'] = open(os.devnull, 'r+b') # same as mode='w+b'
-a['BufferedReaderType'] = open(os.devnull, 'rb') # (default: buffering=-1)
-a['BufferedWriterType'] = open(os.devnull, 'wb')
-try: # oddities: deprecated
- from _pyio import open as _open
- a['PyTextWrapperType'] = _open(os.devnull, 'r', buffering=-1)
- a['PyBufferedRandomType'] = _open(os.devnull, 'r+b', buffering=-1)
- a['PyBufferedReaderType'] = _open(os.devnull, 'rb', buffering=-1)
- a['PyBufferedWriterType'] = _open(os.devnull, 'wb', buffering=-1)
-except ImportError:
- pass
-# other (concrete) object types
-if PY3:
- d['CellType'] = (_lambda)(0).__closure__[0]
- a['XRangeType'] = _xrange = range(1)
-else:
- d['CellType'] = (_lambda)(0).func_closure[0]
- a['XRangeType'] = _xrange = xrange(1)
-d['MethodDescriptorType'] = type.__dict__['mro']
-d['WrapperDescriptorType'] = type.__repr__
-a['WrapperDescriptorType2'] = type.__dict__['__module__']
-# built-in functions (CH 2)
-if PY3: _methodwrap = (1).__lt__
-else: _methodwrap = (1).__cmp__
-d['MethodWrapperType'] = _methodwrap
-a['StaticMethodType'] = staticmethod(_method)
-a['ClassMethodType'] = classmethod(_method)
-a['PropertyType'] = property()
-d['SuperType'] = super(Exception, _exception)
-# string services (CH 7)
-if PY3: _in = _bytes
-else: _in = _str
-a['InputType'] = _cstrI = StringIO(_in)
-a['OutputType'] = _cstrO = StringIO()
-# data types (CH 8)
-a['WeakKeyDictionaryType'] = weakref.WeakKeyDictionary()
-a['WeakValueDictionaryType'] = weakref.WeakValueDictionary()
-a['ReferenceType'] = weakref.ref(_instance)
-a['DeadReferenceType'] = weakref.ref(_class())
-a['ProxyType'] = weakref.proxy(_instance)
-a['DeadProxyType'] = weakref.proxy(_class())
-a['CallableProxyType'] = weakref.proxy(_instance2)
-a['DeadCallableProxyType'] = weakref.proxy(_class2())
-a['QueueType'] = Queue.Queue()
-# numeric and mathematical types (CH 9)
-d['PartialType'] = functools.partial(int,base=2)
-if PY3:
- a['IzipType'] = zip('0','1')
-else:
- a['IzipType'] = itertools.izip('0','1')
-a['ChainType'] = itertools.chain('0','1')
-d['ItemGetterType'] = operator.itemgetter(0)
-d['AttrGetterType'] = operator.attrgetter('__repr__')
-# file and directory access (CH 10)
-if PY3: _fileW = _cstrO
-else: _fileW = _tmpf
-# data persistence (CH 11)
-if HAS_ALL:
- a['ConnectionType'] = _conn = sqlite3.connect(':memory:')
- a['CursorType'] = _conn.cursor()
-a['ShelveType'] = shelve.Shelf({})
-# data compression and archiving (CH 12)
-if HAS_ALL:
- a['BZ2FileType'] = bz2.BZ2File(os.devnull) #FIXME: fail >= 3.3
- a['BZ2CompressorType'] = bz2.BZ2Compressor()
- a['BZ2DecompressorType'] = bz2.BZ2Decompressor()
-#a['ZipFileType'] = _zip = zipfile.ZipFile(os.devnull,'w') #FIXME: fail >= 3.2
-#_zip.write(_tempfile,'x') [causes annoying warning/error printed on import]
-#a['ZipInfoType'] = _zip.getinfo('x')
-a['TarFileType'] = tarfile.open(fileobj=_fileW,mode='w')
-# file formats (CH 13)
-a['DialectType'] = csv.get_dialect('excel')
-a['PackerType'] = xdrlib.Packer()
-# optional operating system services (CH 16)
-a['LockType'] = threading.Lock()
-a['RLockType'] = threading.RLock()
-# generic operating system services (CH 15) # also closed/open and r/w/etc...
-a['NamedLoggerType'] = _logger = logging.getLogger(__name__) #FIXME: fail >= 3.2 and <= 2.6
-#a['FrozenModuleType'] = __hello__ #FIXME: prints "Hello world..."
-# interprocess communication (CH 17)
-if PY3:
- a['SocketType'] = _socket = socket.socket() #FIXME: fail >= 3.3
- a['SocketPairType'] = socket.socketpair()[0] #FIXME: fail >= 3.3
-else:
- a['SocketType'] = _socket = socket.socket()
- a['SocketPairType'] = _socket._sock
-# python runtime services (CH 27)
-if PY3:
- a['GeneratorContextManagerType'] = contextlib.contextmanager(max)([1])
-else:
- a['GeneratorContextManagerType'] = contextlib.GeneratorContextManager(max)
-
-try: # ipython
- __IPYTHON__ is True # is ipython
-except NameError:
- # built-in constants (CH 4)
- a['QuitterType'] = quit
- d['ExitType'] = a['QuitterType']
-try: # numpy
- from numpy import ufunc as _numpy_ufunc
- from numpy import array as _numpy_array
- from numpy import int32 as _numpy_int32
- a['NumpyUfuncType'] = _numpy_ufunc
- a['NumpyArrayType'] = _numpy_array
- a['NumpyInt32Type'] = _numpy_int32
-except ImportError:
- pass
-try: # python 2.6
- # numeric and mathematical types (CH 9)
- a['ProductType'] = itertools.product('0','1')
- # generic operating system services (CH 15)
- a['FileHandlerType'] = logging.FileHandler(os.devnull) #FIXME: fail >= 3.2 and <= 2.6
- a['RotatingFileHandlerType'] = logging.handlers.RotatingFileHandler(os.devnull)
- a['SocketHandlerType'] = logging.handlers.SocketHandler('localhost',514)
- a['MemoryHandlerType'] = logging.handlers.MemoryHandler(1)
-except AttributeError:
- pass
-try: # python 2.7
- # data types (CH 8)
- a['WeakSetType'] = weakref.WeakSet() # 2.7
-# # generic operating system services (CH 15) [errors when dill is imported]
-# a['ArgumentParserType'] = _parser = argparse.ArgumentParser('PROG')
-# a['NamespaceType'] = _parser.parse_args() # pickle ok
-# a['SubParsersActionType'] = _parser.add_subparsers()
-# a['MutuallyExclusiveGroupType'] = _parser.add_mutually_exclusive_group()
-# a['ArgumentGroupType'] = _parser.add_argument_group()
-except AttributeError:
- pass
-
-# -- dill fails in some versions below here ---------------------------------
-# types module (part of CH 8)
-a['FileType'] = open(os.devnull, 'rb', buffering=0) # same 'wb','wb+','rb+'
-# FIXME: FileType fails >= 3.1
-# built-in functions (CH 2)
-a['ListIteratorType'] = iter(_list) # empty vs non-empty FIXME: fail < 3.2
-a['TupleIteratorType']= iter(_tuple) # empty vs non-empty FIXME: fail < 3.2
-a['XRangeIteratorType'] = iter(_xrange) # empty vs non-empty FIXME: fail < 3.2
-# data types (CH 8)
-a['PrettyPrinterType'] = pprint.PrettyPrinter() #FIXME: fail >= 3.2 and == 2.5
-# numeric and mathematical types (CH 9)
-a['CycleType'] = itertools.cycle('0') #FIXME: fail < 3.2
-# file and directory access (CH 10)
-a['TemporaryFileType'] = _tmpf #FIXME: fail >= 3.2 and == 2.5
-# data compression and archiving (CH 12)
-a['GzipFileType'] = gzip.GzipFile(fileobj=_fileW) #FIXME: fail > 3.2 and <= 2.6
-# generic operating system services (CH 15)
-a['StreamHandlerType'] = logging.StreamHandler() #FIXME: fail >= 3.2 and == 2.5
-try: # python 2.6
- # numeric and mathematical types (CH 9)
- a['PermutationsType'] = itertools.permutations('0') #FIXME: fail < 3.2
- a['CombinationsType'] = itertools.combinations('0',1) #FIXME: fail < 3.2
-except AttributeError:
- pass
-try: # python 2.7
- # numeric and mathematical types (CH 9)
- a['RepeatType'] = itertools.repeat(0) #FIXME: fail < 3.2
- a['CompressType'] = itertools.compress('0',[1]) #FIXME: fail < 3.2
- #XXX: ...and etc
-except AttributeError:
- pass
-
-# -- dill fails on all below here -------------------------------------------
-# types module (part of CH 8)
-x['GeneratorType'] = _generator = _function(1) #XXX: priority
-x['FrameType'] = _generator.gi_frame #XXX: inspect.currentframe()
-x['TracebackType'] = _function2()[1] #(see: inspect.getouterframes,getframeinfo)
-# other (concrete) object types
-# (also: Capsule / CObject ?)
-# built-in functions (CH 2)
-x['SetIteratorType'] = iter(_set) #XXX: empty vs non-empty
-# built-in types (CH 5)
-if PY3:
- x['DictionaryItemIteratorType'] = iter(type.__dict__.items())
- x['DictionaryKeyIteratorType'] = iter(type.__dict__.keys())
- x['DictionaryValueIteratorType'] = iter(type.__dict__.values())
-else:
- x['DictionaryItemIteratorType'] = type.__dict__.iteritems()
- x['DictionaryKeyIteratorType'] = type.__dict__.iterkeys()
- x['DictionaryValueIteratorType'] = type.__dict__.itervalues()
-# string services (CH 7)
-x['StructType'] = struct.Struct('c')
-x['CallableIteratorType'] = _srepattern.finditer('')
-x['SREMatchType'] = _srepattern.match('')
-x['SREScannerType'] = _srepattern.scanner('')
-x['StreamReader'] = codecs.StreamReader(_cstrI) #XXX: ... and etc
-# python object persistence (CH 11)
-# x['DbShelveType'] = shelve.open('foo','n')#,protocol=2) #XXX: delete foo
-if HAS_ALL:
- x['DbmType'] = dbm.open(_tempfile,'n')
-# x['DbCursorType'] = _dbcursor = anydbm.open('foo','n') #XXX: delete foo
-# x['DbType'] = _dbcursor.db
-# data compression and archiving (CH 12)
-x['ZlibCompressType'] = zlib.compressobj()
-x['ZlibDecompressType'] = zlib.decompressobj()
-# file formats (CH 13)
-x['CSVReaderType'] = csv.reader(_cstrI)
-x['CSVWriterType'] = csv.writer(_cstrO)
-x['CSVDictReaderType'] = csv.DictReader(_cstrI)
-x['CSVDictWriterType'] = csv.DictWriter(_cstrO,{})
-# cryptographic services (CH 14)
-x['HashType'] = hashlib.md5()
-x['HMACType'] = hmac.new(_in)
-# generic operating system services (CH 15)
-#x['CursesWindowType'] = _curwin = curses.initscr() #FIXME: messes up tty
-#x['CursesTextPadType'] = textpad.Textbox(_curwin)
-#x['CursesPanelType'] = panel.new_panel(_curwin)
-if HAS_CTYPES:
- x['CCharPType'] = ctypes.c_char_p()
- x['CWCharPType'] = ctypes.c_wchar_p()
- x['CVoidPType'] = ctypes.c_void_p()
- x['CDLLType'] = _cdll = ctypes.CDLL(None)
- x['PyDLLType'] = _pydll = ctypes.pythonapi
- x['FuncPtrType'] = _cdll._FuncPtr()
- x['CCharArrayType'] = ctypes.create_string_buffer(1)
- x['CWCharArrayType'] = ctypes.create_unicode_buffer(1)
- x['CParamType'] = ctypes.byref(_cchar)
- x['LPCCharType'] = ctypes.pointer(_cchar)
- x['LPCCharObjType'] = _lpchar = ctypes.POINTER(ctypes.c_char)
- x['NullPtrType'] = _lpchar()
- x['NullPyObjectType'] = ctypes.py_object()
- x['PyObjectType'] = ctypes.py_object(1)
- x['FieldType'] = _field = _Struct._field
- x['CFUNCTYPEType'] = _cfunc = ctypes.CFUNCTYPE(ctypes.c_char)
- x['CFunctionType'] = _cfunc(str)
-try: # python 2.6
- # numeric and mathematical types (CH 9)
- x['MethodCallerType'] = operator.methodcaller('mro') # 2.6
-except AttributeError:
- pass
-try: # python 2.7
- # built-in types (CH 5)
- x['MemoryType'] = memoryview(_in) # 2.7
- x['MemoryType2'] = memoryview(bytearray(_in)) # 2.7
- if PY3:
- x['DictItemsType'] = _dict.items() # 2.7
- x['DictKeysType'] = _dict.keys() # 2.7
- x['DictValuesType'] = _dict.values() # 2.7
- else:
- x['DictItemsType'] = _dict.viewitems() # 2.7
- x['DictKeysType'] = _dict.viewkeys() # 2.7
- x['DictValuesType'] = _dict.viewvalues() # 2.7
- # generic operating system services (CH 15)
- x['RawTextHelpFormatterType'] = argparse.RawTextHelpFormatter('PROG')
- x['RawDescriptionHelpFormatterType'] = argparse.RawDescriptionHelpFormatter('PROG')
- x['ArgDefaultsHelpFormatterType'] = argparse.ArgumentDefaultsHelpFormatter('PROG')
-except NameError:
- pass
-try: # python 2.7 (and not 3.1)
- x['CmpKeyType'] = _cmpkey = functools.cmp_to_key(_methodwrap) # 2.7, >=3.2
- x['CmpKeyObjType'] = _cmpkey('0') #2.7, >=3.2
-except AttributeError:
- pass
-if PY3: # oddities: removed, etc
- x['BufferType'] = x['MemoryType']
-else:
- x['BufferType'] = buffer('')
-
-# -- cleanup ----------------------------------------------------------------
-a.update(d) # registered also succeed
-os.remove(_tempfile)
-
-
-# EOF
http://git-wip-us.apache.org/repos/asf/flink/blob/68b15593/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/detect.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/detect.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/detect.py
deleted file mode 100644
index 749a573..0000000
--- a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/detect.py
+++ /dev/null
@@ -1,222 +0,0 @@
-#!/usr/bin/env python
-#
-# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
-# Copyright (c) 2008-2014 California Institute of Technology.
-# License: 3-clause BSD. The full license text is available at:
-# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
-"""
-Methods for detecting objects leading to pickling failures.
-"""
-
-from __future__ import absolute_import
-from inspect import ismethod, isfunction, istraceback, isframe, iscode
-from .pointers import parent, reference, at, parents, children
-
-from .dill import _trace as trace
-from .dill import PY3
-
-def outermost(func): # is analogous to getsource(func,enclosing=True)
- """get outermost enclosing object (i.e. the outer function in a closure)
-
- NOTE: this is the object-equivalent of getsource(func, enclosing=True)
- """
- if PY3:
- if ismethod(func):
- _globals = func.__func__.__globals__ or {}
- elif isfunction(func):
- _globals = func.__globals__ or {}
- else:
- return #XXX: or raise? no matches
- _globals = _globals.items()
- else:
- if ismethod(func):
- _globals = func.im_func.func_globals or {}
- elif isfunction(func):
- _globals = func.func_globals or {}
- else:
- return #XXX: or raise? no matches
- _globals = _globals.iteritems()
- # get the enclosing source
- from .source import getsourcelines
- try: lines,lnum = getsourcelines(func, enclosing=True)
- except: #TypeError, IOError
- lines,lnum = [],None
- code = ''.join(lines)
- # get all possible names,objects that are named in the enclosing source
- _locals = ((name,obj) for (name,obj) in _globals if name in code)
- # now only save the objects that generate the enclosing block
- for name,obj in _locals: #XXX: don't really need 'name'
- try:
- if getsourcelines(obj) == (lines,lnum): return obj
- except: #TypeError, IOError
- pass
- return #XXX: or raise? no matches
-
-def nestedcode(func): #XXX: or return dict of {co_name: co} ?
- """get the code objects for any nested functions (e.g. in a closure)"""
- func = code(func)
- if not iscode(func): return [] #XXX: or raise? no matches
- nested = []
- for co in func.co_consts:
- if co is None: continue
- co = code(co)
- if co: nested.append(co)
- return nested
-
-def code(func):
- '''get the code object for the given function or method
-
- NOTE: use dill.source.getsource(CODEOBJ) to get the source code
- '''
- if PY3:
- im_func = '__func__'
- func_code = '__code__'
- else:
- im_func = 'im_func'
- func_code = 'func_code'
- if ismethod(func): func = getattr(func, im_func)
- if isfunction(func): func = getattr(func, func_code)
- if istraceback(func): func = func.tb_frame
- if isframe(func): func = func.f_code
- if iscode(func): return func
- return
-
-def nested(func): #XXX: or return dict of {__name__: obj} ?
- """get any functions inside of func (e.g. inner functions in a closure)
-
- NOTE: results may differ if the function has been executed or not.
- If len(nestedcode(func)) > len(nested(func)), try calling func().
- If possible, python builds code objects, but delays building functions
- until func() is called.
- """
- if PY3:
- att1 = '__code__'
- att0 = '__func__'
- else:
- att1 = 'func_code' # functions
- att0 = 'im_func' # methods
-
- import gc
- funcs = []
- # get the code objects, and try to track down by referrence
- for co in nestedcode(func):
- # look for function objects that refer to the code object
- for obj in gc.get_referrers(co):
- # get methods
- _ = getattr(obj, att0, None) # ismethod
- if getattr(_, att1, None) is co: funcs.append(obj)
- # get functions
- elif getattr(obj, att1, None) is co: funcs.append(obj)
- # get frame objects
- elif getattr(obj, 'f_code', None) is co: funcs.append(obj)
- # get code objects
- elif hasattr(obj, 'co_code') and obj is co: funcs.append(obj)
-# frameobjs => func.func_code.co_varnames not in func.func_code.co_cellvars
-# funcobjs => func.func_code.co_cellvars not in func.func_code.co_varnames
-# frameobjs are not found, however funcobjs are...
-# (see: test_mixins.quad ... and test_mixins.wtf)
-# after execution, code objects get compiled, and them may be found by gc
- return funcs
-
-
-def freevars(func):
- """get objects defined in enclosing code that are referred to by func
-
- returns a dict of {name:object}"""
- if PY3:
- im_func = '__func__'
- func_code = '__code__'
- func_closure = '__closure__'
- else:
- im_func = 'im_func'
- func_code = 'func_code'
- func_closure = 'func_closure'
- if ismethod(func): func = getattr(func, im_func)
- if isfunction(func):
- closures = getattr(func, func_closure) or ()
- func = getattr(func, func_code).co_freevars # get freevars
- else:
- return {}
- return dict((name,c.cell_contents) for (name,c) in zip(func,closures))
-
-def globalvars(func):
- """get objects defined in global scope that are referred to by func
-
- return a dict of {name:object}"""
- if PY3:
- im_func = '__func__'
- func_code = '__code__'
- func_globals = '__globals__'
- else:
- im_func = 'im_func'
- func_code = 'func_code'
- func_globals = 'func_globals'
- if ismethod(func): func = getattr(func, im_func)
- if isfunction(func):
- globs = getattr(func, func_globals) or {}
- func = getattr(func, func_code).co_names # get names
- else:
- return {}
- #NOTE: if name not in func_globals, then we skip it...
- return dict((name,globs[name]) for name in func if name in globs)
-
-def varnames(func):
- """get names of variables defined by func
-
- returns a tuple (local vars, local vars referrenced by nested functions)"""
- func = code(func)
- if not iscode(func):
- return () #XXX: better ((),())? or None?
- return func.co_varnames, func.co_cellvars
-
-
-def baditems(obj, exact=False, safe=False): #XXX: obj=globals() ?
- """get items in object that fail to pickle"""
- if not hasattr(obj,'__iter__'): # is not iterable
- return [j for j in (badobjects(obj,0,exact,safe),) if j is not None]
- obj = obj.values() if getattr(obj,'values',None) else obj
- _obj = [] # can't use a set, as items may be unhashable
- [_obj.append(badobjects(i,0,exact,safe)) for i in obj if i not in _obj]
- return [j for j in _obj if j is not None]
-
-
-def badobjects(obj, depth=0, exact=False, safe=False):
- """get objects that fail to pickle"""
- from dill import pickles
- if not depth:
- if pickles(obj,exact,safe): return None
- return obj
- return dict(((attr, badobjects(getattr(obj,attr),depth-1,exact,safe)) \
- for attr in dir(obj) if not pickles(getattr(obj,attr),exact,safe)))
-
-def badtypes(obj, depth=0, exact=False, safe=False):
- """get types for objects that fail to pickle"""
- from dill import pickles
- if not depth:
- if pickles(obj,exact,safe): return None
- return type(obj)
- return dict(((attr, badtypes(getattr(obj,attr),depth-1,exact,safe)) \
- for attr in dir(obj) if not pickles(getattr(obj,attr),exact,safe)))
-
-def errors(obj, depth=0, exact=False, safe=False):
- """get errors for objects that fail to pickle"""
- from dill import pickles, copy
- if not depth:
- try:
- pik = copy(obj)
- if exact:
- assert pik == obj, \
- "Unpickling produces %s instead of %s" % (pik,obj)
- assert type(pik) == type(obj), \
- "Unpickling produces %s instead of %s" % (type(pik),type(obj))
- return None
- except Exception:
- import sys
- return sys.exc_info()[1]
- return dict(((attr, errors(getattr(obj,attr),depth-1,exact,safe)) \
- for attr in dir(obj) if not pickles(getattr(obj,attr),exact,safe)))
-
-del absolute_import
-
-
-# EOF