[
https://issues.apache.org/jira/browse/FLINK-1927?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14646088#comment-14646088
]
ASF GitHub Bot commented on FLINK-1927:
---------------------------------------
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/931#discussion_r35763245
--- Diff:
flink-staging/flink-language-binding/flink-python/src/main/java/org/apache/flink/languagebinding/api/java/python/streaming/PythonStreamer.java
---
@@ -114,31 +97,12 @@ public void run() {
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();
+ process.getOutputStream().write("operator\n".getBytes());
+ process.getOutputStream().write(("" + server.getLocalPort() +
"\n").getBytes());
+ process.getOutputStream().write((id + "\n").getBytes());
+ process.getOutputStream().write((inputFilePath +
"\n").getBytes());
+ process.getOutputStream().write((outputFilePath +
"\n").getBytes());
+ process.getOutputStream().flush();
--- End diff --
We could reuse `process.getOutputStream()` here by saving it to a variable.
> [Py] Rework operator distribution
> ---------------------------------
>
> Key: FLINK-1927
> URL: https://issues.apache.org/jira/browse/FLINK-1927
> Project: Flink
> Issue Type: Improvement
> Components: Python API
> Affects Versions: 0.9
> Reporter: Chesnay Schepler
> Assignee: Chesnay Schepler
> Priority: Minor
> Fix For: 0.9
>
>
> Currently, the python operator is created when execution the python plan
> file, serialized using dill and saved as a byte[] in the java function. It is
> then deserialized at runtime on each node.
> The current implementation is fairly hacky, and imposes certain limitations
> that make it hard to work with. Chaining, or generally saving other
> user-code, always requires a separate deserialization step after
> deserializing the operator.
> These issues can be easily circumvented by rebuilding the (python) plan on
> each node, instead of serializing the operator. The plan creation is
> deterministic, and every operator is uniquely identified by an ID that is
> already known to the java function.
> This change will allow us to easily support custom serializers.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
