You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by zohar-mizrahi <gi...@git.apache.org> on 2017/05/07 14:45:23 UTC
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
GitHub user zohar-mizrahi opened a pull request:
https://github.com/apache/flink/pull/3838
[FLINK-5886] Python API for streaming applications
Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html).
In addition to going through the list, please provide a meaningful description of your changes.
- [ ] General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text")
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message (including the JIRA id)
- [ ] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
- [ ] Tests & Build
- Functionality added by the pull request is covered by tests
- `mvn clean verify` has been executed successfully locally or a Travis build has passed
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zohar-mizrahi/flink master
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3838.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3838
----
commit c1333c3424897caa615683d3494b41e7ab88d45d
Author: Zohar Mizrahi <zo...@parallelmachines.com>
Date: 2016-11-15T12:46:36Z
[FLINK-5886] Python API for streaming applications
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121900689
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ public PyObject read (Kryo kryo, Input input, Class<PyObject> type) {
--- End diff --
remove space after `read`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117723303
--- Diff: flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main2.py ---
@@ -16,12 +16,12 @@
# 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.Aggregation import Max, Min, Sum
+from flink.functions.CoGroupFunction import CoGroupFunction
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 flink.functions.MapFunction import MapFunction
+from flink.plan.Environment import get_environment
--- End diff --
should be reverted
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121901283
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/PythonCollector.java ---
@@ -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.
+ */
+package org.apache.flink.streaming.python.util;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.util.Collector;
+import org.python.core.PyObject;
+
+/**
+ * Collects a {@code PyObject} record and forwards it. It makes sure that the record is converted,
+ * if necessary, to a {@code PyObject}.
+ */
+@Public
+public class PythonCollector implements Collector<PyObject> {
+ private Collector<PyObject> collector;
+
+ public void setCollector(Collector<PyObject> collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void collect(PyObject record) {
+ PyObject po = UtilityFunctions.adapt(record);
--- End diff --
AM i missing something or is this call unnecessary. Isn't it guaranteed that `record instanceof PyObject` is `true`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133392462
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
My bad - I missed the java unchecked exceptions part (.e.g runtime exception). It'll be much better to use it here.
As for the `read`, we can either return `null` or again use the runtime exception.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133171218
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
ObjectOutputStreams do not fail with an IOException if the object is null,
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133192783
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/PythonCollector.java ---
@@ -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.
+ */
+package org.apache.flink.streaming.python.util;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.util.Collector;
+import org.python.core.PyObject;
+
+/**
+ * Collects a {@code PyObject} record and forwards it. It makes sure that the record is converted,
+ * if necessary, to a {@code PyObject}.
+ */
+@Public
+public class PythonCollector implements Collector<PyObject> {
+ private Collector<PyObject> collector;
+
+ public void setCollector(Collector<PyObject> collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void collect(PyObject record) {
+ PyObject po = UtilityFunctions.adapt(record);
--- End diff --
Actually you're right. This line can be dropped. And now, it seems that the `PythonCollector` is redundant, though it provides a more safety layer to report users about casting problems in case they provide UDF from the Java source code. If their function would not handle PyObject that it will break in runtime, saying something like and object cannot be cast to 'PyObject'.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
I've been digging into this for the past week. I found a number of things to improve and did so in a local branch. Once I've finalized/tests things (probably tomorrow) I'll link the branch here or open another PR.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724303
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java ---
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.core.fs.FileSystem.WriteMode;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.python.api.functions.PyKey;
+import org.apache.flink.streaming.python.api.functions.PythonFilterFunction;
+import org.apache.flink.streaming.python.api.functions.PythonFlatMapFunction;
+import org.apache.flink.streaming.python.api.functions.PythonKeySelector;
+import org.apache.flink.streaming.python.api.functions.PythonMapFunction;
+import org.apache.flink.streaming.python.api.functions.PythonOutputSelector;
+import org.apache.flink.streaming.python.api.functions.PythonSinkFunction;
+import org.apache.flink.streaming.python.util.serialization.PythonSerializationSchema;
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+
+/**
+ * A {@code PythonDataStream} is a thin wrapper layer over {@link DataStream}, which represents a
+ * stream of elements of the same type. A {@code PythonDataStream} can be transformed into
+ * another {@code PythonDataStream} by applying various transformation functions, such as
+ * <ul>
+ * <li>{@link PythonDataStream#map}
+ * <li>{@link PythonDataStream#split}
+ * </ul>
+ *
+ * <p>A thin wrapper layer means that the functionality itself is performed by the
+ * {@link DataStream}, however instead of working directly with the streaming data sets,
+ * this layer handles Python wrappers (e.g. {@code PythonDataStream}) to comply with the
+ * Python standard coding styles.</p>
+ */
+@Public
--- End diff --
I would make the entire class as `@PublicEvolving`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117723223
--- Diff: flink-libraries/flink-python/pom.xml ---
@@ -59,20 +59,32 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>${project.version}</version>
- <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${project.version}</version>
- <scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
--- End diff --
These changes should be reverted i guess.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on the issue:
https://github.com/apache/flink/pull/3838
The thing is that I use the ```PythonEnvironmentConfig.pythonTmpCachePath``` attribute to deliver information from the ```PythonStreamBinder``` to a class that is called from the python script.
How would you suggest to do it otherwise?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on the issue:
https://github.com/apache/flink/pull/3838
One of the critical attributes is ```PythonEnvironmentConfig::pythonTmpCachePath```, which is used in the following places:
- ```PythonStreamExecutionEnvironment::execute:362```
- ```PythonStreamExecutionEnvironment::execute:400```
- ```PythonStreamBinder::prepareFiles:117```
On the client side, the temporary files are prepared for distribution by the ```PythonStreamBinder``` and then processed by the ``````PythonStreamExecutionEnvironment::execute``` function, which is called from the Python script. When the python script is executed on the TaskManager, this attribute remains ```null``` and thus, the ```execute``` returns immediately.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on the issue:
https://github.com/apache/flink/pull/3838
Regarding the exception -
```java.io.IOException: java.io.IOException: The given HDFS file URI ...```
In general, using the python interface requires a valid configuration of shared file system (.e.g HDFS), which designed to distribute the python files.
Someone can bypass this issue by set the second argument to 'True' when calling to ```env.execute(...)``` in the python script.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121900761
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
is it intended to silently continue here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
I can't seem to get the tests running.
In the IDE i get this exception:
```
null
Traceback (most recent call last):
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_7475612982544297474\plan.py", line 3, in <module>
run_all_testsae1bf92fc871d56dae4598b332a87804.main()
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_7475612982544297474\run_all_testsae1bf92fc871d56dae4598b332a87804.py", line 71, in main
Main().run()
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_7475612982544297474\run_all_testsae1bf92fc871d56dae4598b332a87804.py", line 45, in run
tests.append(__import__(test_module_name, globals(), locals()))
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_7475612982544297474\test_filter.py", line 25, in <module>
from utils.python_test_base import TestBase
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_7475612982544297474\utils\python_test_base.py", line 19, in <module>
from org.apache.flink.api.java.utils import ParameterTool
java.lang.NoClassDefFoundError: org/apache/flink/api/java/utils (wrong name: org/apache/flink/api/java/Utils)
at java.lang.ClassLoader.defineClass1(Native Method)
```
On the command-line the tests do run, but fail with this exception:
```
Submitting job ... 'test_filter'
Get execution environment
2> (50, u'hello')
null
Traceback (most recent call last):
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_-7332098182433468284\plan.py", line 3, in <module>
run_all_testsae1bf92fc871d56dae4598b332a87804.main()
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_-7332098182433468284\run_all_testsae1bf92fc871d56dae4598b332a87804.py", line 71, in main
Main().run()
File "C:\Users\Zento\AppData\Local\Temp\flink_streaming_plan_-7332098182433468284\run_all_testsae1bf92fc871d56dae4598b332a87804.py", line 59, in run
print("\n{}\n{}\n{}\n".format('#'*len(ex_type), ex_type, '#'*len(ex_type)))
TypeError: object of type 'java.lang.Class' has no len()
Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 17.835 sec <<< FAILURE! - in org.apache.flink.streaming.python.api.PythonStreamBinderTest
testJob(org.apache.flink.streaming.python.api.PythonStreamBinderTest) Time elapsed: 17.393 sec <<< FAILURE!
java.lang.AssertionError: Error while calling the test program: null
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117726281
--- Diff: flink-libraries/pom.xml ---
@@ -52,7 +53,7 @@ under the License.
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
- <scope>provided</scope>
--- End diff --
This should be reverted.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121900122
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/functions/PythonOutputSelector.java ---
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.functions;
+
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.python.util.serialization.SerializationUtils;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * The {@code PythonOutputSelector} is a thin wrapper layer over a Python UDF {@code OutputSelector}.
+ * It receives an {@code OutputSelector} as an input and keeps it internally in a serialized form.
+ * It is then delivered, as part of the job graph, up to the TaskManager, then it is opened and becomes
+ * a sort of mediator to the Python UDF {@code OutputSelector}.
+ *
+ * <p>This function is used internally by the Python thin wrapper layer over the streaming data
+ * functionality</p>
+ */
+public class PythonOutputSelector implements OutputSelector<PyObject> {
+ private static final long serialVersionUID = 909266346633598177L;
+
+ private final byte[] serFun;
+ private transient OutputSelector<PyObject> fun;
+
+ public PythonOutputSelector(OutputSelector<PyObject> fun) throws IOException {
+ this.serFun = SerializationUtils.serializeObject(fun);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Iterable<String> select(PyObject value) {
+ if (this.fun == null) {
+ try {
+ this.fun = (OutputSelector<PyObject>) SerializationUtils.deserializeObject(this.serFun);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
we should fail here, and in case of a `ClassNotFoundException`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724247
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java ---
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.core.fs.FileSystem.WriteMode;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.python.api.functions.PyKey;
+import org.apache.flink.streaming.python.api.functions.PythonFilterFunction;
+import org.apache.flink.streaming.python.api.functions.PythonFlatMapFunction;
+import org.apache.flink.streaming.python.api.functions.PythonKeySelector;
+import org.apache.flink.streaming.python.api.functions.PythonMapFunction;
+import org.apache.flink.streaming.python.api.functions.PythonOutputSelector;
+import org.apache.flink.streaming.python.api.functions.PythonSinkFunction;
+import org.apache.flink.streaming.python.util.serialization.PythonSerializationSchema;
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+
+/**
+ * A {@code PythonDataStream} is a thin wrapper layer over {@link DataStream}, which represents a
+ * stream of elements of the same type. A {@code PythonDataStream} can be transformed into
+ * another {@code PythonDataStream} by applying various transformation functions, such as
+ * <ul>
+ * <li>{@link PythonDataStream#map}
+ * <li>{@link PythonDataStream#split}
+ * </ul>
+ *
+ * <p>A thin wrapper layer means that the functionality itself is performed by the
+ * {@link DataStream}, however instead of working directly with the streaming data sets,
+ * this layer handles Python wrappers (e.g. {@code PythonDataStream}) to comply with the
+ * Python standard coding styles.</p>
+ */
+@Public
+public class PythonDataStream<D extends DataStream<PyObject>> {
+ protected final D stream;
+
+ public PythonDataStream(D stream) {
+ this.stream = stream;
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#union(DataStream[])}.
+ *
+ * @param streams
+ * The Python DataStreams to union output with.
+ * @return The {@link PythonDataStream}.
+ */
+ @SafeVarargs
+ @SuppressWarnings("unchecked")
+ public final PythonDataStream union(PythonDataStream... streams) {
+ ArrayList<DataStream<PyObject>> dsList = new ArrayList<>();
+ for (PythonDataStream ps : streams) {
+ dsList.add(ps.stream);
+ }
+ DataStream<PyObject>[] dsArray = new DataStream[dsList.size()];
+ return new PythonDataStream(stream.union(dsList.toArray(dsArray)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#split(OutputSelector)}.
+ *
+ * @param output_selector
+ * The user defined {@link OutputSelector} for directing the tuples.
+ * @return The {@link PythonSplitStream}
+ */
+ public PythonSplitStream split(OutputSelector<PyObject> output_selector) throws IOException {
+ return new PythonSplitStream(this.stream.split(new PythonOutputSelector(output_selector)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#filter(FilterFunction)}.
+ *
+ * @param filter
+ * The FilterFunction that is called for each element of the DataStream.
+ * @return The filtered {@link PythonDataStream}.
+ */
+ public PythonSingleOutputStreamOperator filter(FilterFunction<PyObject> filter) throws IOException {
+ return new PythonSingleOutputStreamOperator(stream.filter(new PythonFilterFunction(filter)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#map(MapFunction)}.
+ *
+ * @param mapper
+ * The MapFunction that is called for each element of the
+ * DataStream.
+ * @return The transformed {@link PythonDataStream}.
+ */
+ public PythonDataStream<SingleOutputStreamOperator<PyObject>> map(
+ MapFunction<PyObject, PyObject> mapper) throws IOException {
+ return new PythonDataStream<>(stream.map(new PythonMapFunction(mapper)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#flatMap(FlatMapFunction)}.
+ *
+ * @param flat_mapper
+ * The FlatMapFunction that is called for each element of the
+ * DataStream
+ *
+ * @return The transformed {@link PythonDataStream}.
+ */
+ public PythonDataStream<SingleOutputStreamOperator<PyObject>> flat_map(
+ FlatMapFunction<PyObject, PyObject> flat_mapper) throws IOException {
+ return new PythonDataStream<>(stream.flatMap(new PythonFlatMapFunction(flat_mapper)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#keyBy(KeySelector)}.
+ *
+ * @param selector
+ * The KeySelector to be used for extracting the key for partitioning
+ * @return The {@link PythonDataStream} with partitioned state (i.e. {@link PythonKeyedStream})
+ */
+ public PythonKeyedStream key_by(KeySelector<PyObject, PyKey> selector) throws IOException {
+ return new PythonKeyedStream(stream.keyBy(new PythonKeySelector(selector)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#print()}.
+ */
+ @PublicEvolving
+ public void print() { stream.print(); }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#writeAsText(java.lang.String)}.
+ *
+ * @param path
+ * The path pointing to the location the text file is written to.
+ */
+ @PublicEvolving
+ public void write_as_text(String path) { stream.writeAsText(path); }
--- End diff --
`stream.print();` should be on a separate line.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121891397
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@PublicEvolving
+public class PythonStreamExecutionEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+ private final StreamExecutionEnvironment env;
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
--- End diff --
this should refer to `set_parallelism`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121901070
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PythonDeserializationSchema.java ---
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+
+import java.io.IOException;
+
+/**
+ * A Python deserialization schema, which implements {@link DeserializationSchema}. It converts a
+ * serialized form of {@code PyObject} into its Java Object representation.
+ */
+public class PythonDeserializationSchema implements DeserializationSchema<Object> {
+ private static final long serialVersionUID = -9180596504893036458L;
+ private final TypeInformation<Object> resultType = TypeInformation.of(new TypeHint<Object>(){});
+
+ private final byte[] serSchema;
+ private transient DeserializationSchema<Object> schema;
+
+ public PythonDeserializationSchema(DeserializationSchema<Object> schema) throws IOException {
+ this.serSchema = SerializationUtils.serializeObject(schema);
+ }
+
+ @SuppressWarnings("unchecked")
+ public Object deserialize(byte[] message) throws IOException {
+ if (this.schema == null) {
+ try {
+ this.schema = (DeserializationSchema<Object>) SerializationUtils.deserializeObject(this.serSchema);
+ } catch (ClassNotFoundException e) {
+ e.printStackTrace();
--- End diff --
we should fail right here as otherwise he get an NPE later on.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117788892
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java ---
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.python.util.PythonObjectInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+
+/**
+ * A helper class to overcome the inability to set the serialVersionUID in a python user-defined
+ * function (UDF).
+ * <p>The fact the this field is not set, results in a dynamic calculation of this serialVersionUID,
+ * using SHA, to make sure it is a unique number. This unique number is a 64-bit hash of the
+ * class name, interface class names, methods, and fields. If a Python class inherits from a
+ * Java class, as in the case of Python UDFs, then a proxy wrapper class is created. Its name is
+ * constructed using the following pattern:
+ * <b>{@code org.python.proxies.<module-name>$<UDF-class-name>$<number>}</b>. The {@code <number>}
+ * part is increased by one in runtime, for every job submission. It results in different serial
+ * version UID for each run for the same Python class. Therefore, it is required to silently
+ * suppress the serial version UID mismatch check.</p>
+ */
+public class PythonObjectInputStream2 extends PythonObjectInputStream {
--- End diff --
How would you suggest to call it? It extends the `PythonObjectInputStream`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on the issue:
https://github.com/apache/flink/pull/3838
Referring to the issue with the ```PythonEnvironmentConfig ``` above, Is there any other global indication that I can use to test whether a given function is executed on the TaskManager?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117806756
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
+public class PythonStreamExecutionEnvironment {
+ private final StreamExecutionEnvironment env;
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment();
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(config);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(parallelism, config);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, jar_files);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, config, jar_files);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, parallelism, jar_files);
+ }
+
+ private PythonStreamExecutionEnvironment() {
+ this.env = StreamExecutionEnvironment.getExecutionEnvironment();
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(Configuration config) {
+ this.env = new LocalStreamEnvironment(config);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(int parallelism, Configuration config) {
+ this.env = StreamExecutionEnvironment.createLocalEnvironment(parallelism, config);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(String host, int port, String... jar_files) {
+ this.env = StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(String host, int port, Configuration config, String... jar_files) {
+ this.env = StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(String host, int port, int parallelism, String... jar_files) {
+ this.env = StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files);
+ this.registerJythonSerializers();
+ }
+
+ private void registerJythonSerializers() {
+ this.env.registerTypeWithKryoSerializer(PyString.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInteger.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyLong.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
+ }
+
+ public PythonDataStream create_python_source(SourceFunction<Object> src) throws Exception {
+ return new PythonDataStream<>(env.addSource(new PythonGeneratorFunction(src)).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * Add a java source to the streaming topology. The source expected to be an java based
+ * implementation (.e.g. Kafka connector).
+ *
+ * @param src A native java source (e.g. PythonFlinkKafkaConsumer09)
+ * @return Python data stream
+ */
+ public PythonDataStream add_java_source(SourceFunction<Object> src) {
+ return new PythonDataStream<>(env.addSource(src).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#fromElements(java.lang.Object[])}
+ *
+ * @param elements
+ * The array of PyObject elements to create the data stream from.
+ * @return The data stream representing the given array of elements
+ */
+ public PythonDataStream from_elements(PyObject... elements) {
+ return new PythonDataStream<>(env.fromElements(elements));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#fromCollection(java.util.Collection)}
+ *
+ * <p>The input {@code Collection} is of type {@code Object}, because it is a collection
+ * of Python elements. * There type is determined in runtime, by the Jython framework.</p>
+ *
+ * @param collection
+ * The collection of python elements to create the data stream from.
+ * @return
+ * The data stream representing the given collection
+ */
+ public PythonDataStream from_collection(Collection<Object> collection) {
--- End diff --
It concerns the Jython framework. Apparently, when using argument expansion (as in the case of `from_elements`), the Jython framework instantiates PyObject for each element in the the elements. On the other hand, when using a Collection, the Jython framework, instantiates a `PyList` with Java native classes (.e.g Integer, String, etc).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133169331
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
Yes, indeed. According to https://github.com/EsotericSoftware/kryo#serializers: "By default, serializers do not need to handle the object being null."
But anyway, I added a print message to the log.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133159049
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/functions/PythonOutputSelector.java ---
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.functions;
+
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.python.util.serialization.SerializationUtils;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * The {@code PythonOutputSelector} is a thin wrapper layer over a Python UDF {@code OutputSelector}.
+ * It receives an {@code OutputSelector} as an input and keeps it internally in a serialized form.
+ * It is then delivered, as part of the job graph, up to the TaskManager, then it is opened and becomes
+ * a sort of mediator to the Python UDF {@code OutputSelector}.
+ *
+ * <p>This function is used internally by the Python thin wrapper layer over the streaming data
+ * functionality</p>
+ */
+public class PythonOutputSelector implements OutputSelector<PyObject> {
+ private static final long serialVersionUID = 909266346633598177L;
+
+ private final byte[] serFun;
+ private transient OutputSelector<PyObject> fun;
+
+ public PythonOutputSelector(OutputSelector<PyObject> fun) throws IOException {
+ this.serFun = SerializationUtils.serializeObject(fun);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Iterable<String> select(PyObject value) {
+ if (this.fun == null) {
+ try {
+ this.fun = (OutputSelector<PyObject>) SerializationUtils.deserializeObject(this.serFun);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
I would throw a RuntimeException to fail the job. If the function can't be deserialized something went terribly wrong during deployment.
Returning null would cause an NPE later on in `DirectedOutput`, obfuscating what actually caused the error.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117793843
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
+public class PythonStreamExecutionEnvironment {
+ private final StreamExecutionEnvironment env;
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment();
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
--- End diff --
It is somehow refers to your first comment concerning the Python coding style versus maintainability. So, it came up to a point, where we better decide what would the best approach.
As far as I see it there are 3 possible options (maybe you'll come up with others):
1) Strict Python - wrap every single java class to provide uniform interface that comply with the Python coding style guide - this would require a lot of effort, even if we find a way to automatically generate this layer, because it'll always require some adaptations.
2) Java interface - leave the Java interface as is. A python script would be written in Java coding style. IMHO, its not a good option.
3) Mixed Python / Java - in this option, make sure the major APIs, and thus the script body, comply with the Python coding style. In certain circumstances, it is allowed to deviate from the Python coding style and use the Java objects directly. IMHO, it is a fair reasonable option, which has balance between coding style and maintenance.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
When running the example against a local cluster i got this exception
```
java.io.IOException: java.io.IOException: The given HDFS file URI (hdfs:/tmp/flink_cache_-4117839671387669278) did not describe the HDFS NameNode. The attempt to use a default HDFS configuration, as specified in the 'fs.hdfs.hdfsdefault' or 'fs.hdfs.hdfssite' config parameter failed due to the following problem: Either no default file system was registered, or the provided configuration contains no valid authority component (fs.default.name or fs.defaultFS) describing the (hdfs namenode) host and port.
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121900800
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ public PyObject read (Kryo kryo, Input input, Class<PyObject> type) {
+ int len = input.readInt();
+ byte[] serPo = new byte[len];
+ input.read(serPo);
+ PyObject po = null;
+ try {
+ po = (PyObject) SerializationUtils.deserializeObject(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
is it intended to silently continue here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121889042
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
--- End diff --
by convention static final fields are upper case, as in `DEFAULT_PYTHON_SCRIPT_NAME`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117917812
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,99 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.3-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_2.10</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
--- End diff --
Otherwise we include them in the jar-with-dependencies and have them multiple times on the classpath.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117917576
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,99 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.3-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_2.10</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
--- End diff --
argh, let me try again: They should be marked as `<scope>provided</scope>`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
There is none that I'm aware of. It is also possible for the JM and TM to run in the same JVM, say for tests or in local mode.
I can't think of a nice way to solve this, so I suggest we simply disable the check for the PythonEnvironmentConfig class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by meyer-net <gi...@git.apache.org>.
Github user meyer-net commented on the issue:
https://github.com/apache/flink/pull/3838
is the version of 1.5 supports python api for streaming applications?
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r132841381
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,104 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.4-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_${scala.binary.version}</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.python</groupId>
+ <artifactId>jython-standalone</artifactId>
+ <version>2.7.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
--- End diff --
Done for the Scala version.
As for not being able to run the tests:
1. I fixed the issue with the ```TypeError: object of type 'java.lang.Class' has no len()```.
2. I still can't reproduce the main issue, concerning an import of java class from the Python module:
File: ```flink-libraries/flink-streaming-python/src/test/python/org/apache/flink/streaming/python/api/utils/python_test_base.py```
Line:19: ```from org.apache.flink.api.java.utils import ParameterTool```
The given class (ParameterTool) resides in different project `flink-java` and the jython module cannot find it. Probably, It somehow concerns the CLASSPATH.
Any suggestion for how to reproduce it?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133169625
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
+ try {
+ byte[] serPo = SerializationUtils.serializeObject(po);
+ output.writeInt(serPo.length);
+ output.write(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ public PyObject read (Kryo kryo, Input input, Class<PyObject> type) {
+ int len = input.readInt();
+ byte[] serPo = new byte[len];
+ input.read(serPo);
+ PyObject po = null;
+ try {
+ po = (PyObject) SerializationUtils.deserializeObject(serPo);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
Yes. Same as the answer above. In order to verify it, during debugging, I set the returned value to null and it continued without issues.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724604
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java ---
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.python.util.PythonObjectInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+
+/**
+ * A helper class to overcome the inability to set the serialVersionUID in a python user-defined
+ * function (UDF).
+ * <p>The fact the this field is not set, results in a dynamic calculation of this serialVersionUID,
+ * using SHA, to make sure it is a unique number. This unique number is a 64-bit hash of the
+ * class name, interface class names, methods, and fields. If a Python class inherits from a
+ * Java class, as in the case of Python UDFs, then a proxy wrapper class is created. Its name is
+ * constructed using the following pattern:
+ * <b>{@code org.python.proxies.<module-name>$<UDF-class-name>$<number>}</b>. The {@code <number>}
+ * part is increased by one in runtime, for every job submission. It results in different serial
+ * version UID for each run for the same Python class. Therefore, it is required to silently
+ * suppress the serial version UID mismatch check.</p>
+ */
+public class PythonObjectInputStream2 extends PythonObjectInputStream {
--- End diff --
remove ยด2` from name.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724239
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java ---
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.core.fs.FileSystem.WriteMode;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.python.api.functions.PyKey;
+import org.apache.flink.streaming.python.api.functions.PythonFilterFunction;
+import org.apache.flink.streaming.python.api.functions.PythonFlatMapFunction;
+import org.apache.flink.streaming.python.api.functions.PythonKeySelector;
+import org.apache.flink.streaming.python.api.functions.PythonMapFunction;
+import org.apache.flink.streaming.python.api.functions.PythonOutputSelector;
+import org.apache.flink.streaming.python.api.functions.PythonSinkFunction;
+import org.apache.flink.streaming.python.util.serialization.PythonSerializationSchema;
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+
+/**
+ * A {@code PythonDataStream} is a thin wrapper layer over {@link DataStream}, which represents a
+ * stream of elements of the same type. A {@code PythonDataStream} can be transformed into
+ * another {@code PythonDataStream} by applying various transformation functions, such as
+ * <ul>
+ * <li>{@link PythonDataStream#map}
+ * <li>{@link PythonDataStream#split}
+ * </ul>
+ *
+ * <p>A thin wrapper layer means that the functionality itself is performed by the
+ * {@link DataStream}, however instead of working directly with the streaming data sets,
+ * this layer handles Python wrappers (e.g. {@code PythonDataStream}) to comply with the
+ * Python standard coding styles.</p>
+ */
+@Public
+public class PythonDataStream<D extends DataStream<PyObject>> {
+ protected final D stream;
+
+ public PythonDataStream(D stream) {
+ this.stream = stream;
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#union(DataStream[])}.
+ *
+ * @param streams
+ * The Python DataStreams to union output with.
+ * @return The {@link PythonDataStream}.
+ */
+ @SafeVarargs
+ @SuppressWarnings("unchecked")
+ public final PythonDataStream union(PythonDataStream... streams) {
+ ArrayList<DataStream<PyObject>> dsList = new ArrayList<>();
+ for (PythonDataStream ps : streams) {
+ dsList.add(ps.stream);
+ }
+ DataStream<PyObject>[] dsArray = new DataStream[dsList.size()];
+ return new PythonDataStream(stream.union(dsList.toArray(dsArray)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#split(OutputSelector)}.
+ *
+ * @param output_selector
+ * The user defined {@link OutputSelector} for directing the tuples.
+ * @return The {@link PythonSplitStream}
+ */
+ public PythonSplitStream split(OutputSelector<PyObject> output_selector) throws IOException {
+ return new PythonSplitStream(this.stream.split(new PythonOutputSelector(output_selector)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#filter(FilterFunction)}.
+ *
+ * @param filter
+ * The FilterFunction that is called for each element of the DataStream.
+ * @return The filtered {@link PythonDataStream}.
+ */
+ public PythonSingleOutputStreamOperator filter(FilterFunction<PyObject> filter) throws IOException {
+ return new PythonSingleOutputStreamOperator(stream.filter(new PythonFilterFunction(filter)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#map(MapFunction)}.
+ *
+ * @param mapper
+ * The MapFunction that is called for each element of the
+ * DataStream.
+ * @return The transformed {@link PythonDataStream}.
+ */
+ public PythonDataStream<SingleOutputStreamOperator<PyObject>> map(
+ MapFunction<PyObject, PyObject> mapper) throws IOException {
+ return new PythonDataStream<>(stream.map(new PythonMapFunction(mapper)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#flatMap(FlatMapFunction)}.
+ *
+ * @param flat_mapper
+ * The FlatMapFunction that is called for each element of the
+ * DataStream
+ *
+ * @return The transformed {@link PythonDataStream}.
+ */
+ public PythonDataStream<SingleOutputStreamOperator<PyObject>> flat_map(
+ FlatMapFunction<PyObject, PyObject> flat_mapper) throws IOException {
+ return new PythonDataStream<>(stream.flatMap(new PythonFlatMapFunction(flat_mapper)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#keyBy(KeySelector)}.
+ *
+ * @param selector
+ * The KeySelector to be used for extracting the key for partitioning
+ * @return The {@link PythonDataStream} with partitioned state (i.e. {@link PythonKeyedStream})
+ */
+ public PythonKeyedStream key_by(KeySelector<PyObject, PyKey> selector) throws IOException {
+ return new PythonKeyedStream(stream.keyBy(new PythonKeySelector(selector)));
+ }
+
+ /**
+ * A thin wrapper layer over {@link DataStream#print()}.
+ */
+ @PublicEvolving
+ public void print() { stream.print(); }
--- End diff --
`stream.print();` should be on a separate line.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
Here's a preliminary changelog:
General:
- rebase branch to current master
- incremented version to 1.5-SNAPSHOT
- fixed kafka-connector dependency declaration
- set to provided
- scala version set to scala.binary.version
- flink version set to project.version
- applied checkstyle
- disabled method/parameter name rules for API classes
- assigned flink-python-streaming to 'libraries' travis profile
API:
- PDS#map()/flat_map() now return PythonSingleOutputStreamOperator
- renamed PDS#print() to PDS#output()
- print is a keyword in python and thus not usable in native python APIs
- added PythonSingleOutputStreamOperator#name()
- removed env#execute methods that accepted local execution argument as they are redundant due to environment factory methods
Moved/Renamed:
- made SerializerMap top-level class and renamed it to AdapterMap
- Moved UtilityFunctions#adapt to AdapterMap class
- renamed UtilityFunctions to InterpreterUtils
- moved PythonobjectInputStream2 to SerializationUtils
- renamed PythonObjectInputStream2 to SerialVersionOverridingPythonObjectInputStream
Functions:
- Introduced AbstractPythonUDF class for sharing RichRunction#open()/close() implementations
- PythonOutputSelector now throws FlinkRuntimeException when failing during initialization
- added generic return type to Serializationutils#deserializeObject
- added new serializers for PyBoolean/-Float/-Integer/-Long/-String
- PyObjectSerializer not properly fails when an exceptioin occurs
- improved error printing
- PythonCollector now typed to Object and properly converts non-PyObjects
- jython functions that use a collector now have Object has output type
- otherwise you would get ClassCastException if jython returns something that isn't a PyObject
PythonStreamBinder
- adjusted to follow PythonPlanBinder structure
- client-like main() exception handling
- replaced Random usage with UUID.randomUIID()
- now loads GlobalConfiguration
- local/distributed tmp dir now configurable
- introduced PythonOptions
- no longer generate plan.py but instead import it directly via the PythonInterpreter
Environment:
- Reworked static environment factory methods from PythonStreamExecutionEnvironment into a PythonEnvironmentFactory
- program main() method now accepts a PythonEnvironmentFactory
- directories are now passed properly to the environment instead of using static fields
- removed PythonEnvironmentConfig
Tests:
- removed 'if __name__ == '__main__':' blocks from tests since the condition is never fulfilled
- removed python TestBase class
- removed print statements from tests
- standardized test job names
- cleaned up PythonStreamBinderTest / made it more consistent with PythonPlanBinderTest
- run_all_tests improvements
- stop after first failure
- print stacktrace on failure
- no longer relies on dirname() to get cwd but uses the module file location instead
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121900972
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PythonSerializationSchema.java ---
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import java.io.IOException;
+
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.python.core.PyObject;
+
+/**
+ * A Python serialization schema, which implements {@link SerializationSchema}. It converts
+ * a {@code PyObject} into its serialized form.
+ */
+public class PythonSerializationSchema implements SerializationSchema<PyObject> {
+ private static final long serialVersionUID = -9170596504893036458L;
+
+ private final byte[] serSchema;
+ private transient SerializationSchema<PyObject> schema;
+
+ public PythonSerializationSchema(SerializationSchema<PyObject> schema) throws IOException {
+ this.serSchema = SerializationUtils.serializeObject(schema);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public byte[] serialize(PyObject element) {
+ if (this.schema == null) {
+ try {
+ this.schema = (SerializationSchema<PyObject>)SerializationUtils.deserializeObject(this.serSchema);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
we should fail right here, as otherwise we get an NPE later on anyway.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117864592
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,99 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.3-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_2.10</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
--- End diff --
I'm not sure what you mean. Please explain.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133156452
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/functions/PythonOutputSelector.java ---
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.functions;
+
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.python.util.serialization.SerializationUtils;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * The {@code PythonOutputSelector} is a thin wrapper layer over a Python UDF {@code OutputSelector}.
+ * It receives an {@code OutputSelector} as an input and keeps it internally in a serialized form.
+ * It is then delivered, as part of the job graph, up to the TaskManager, then it is opened and becomes
+ * a sort of mediator to the Python UDF {@code OutputSelector}.
+ *
+ * <p>This function is used internally by the Python thin wrapper layer over the streaming data
+ * functionality</p>
+ */
+public class PythonOutputSelector implements OutputSelector<PyObject> {
+ private static final long serialVersionUID = 909266346633598177L;
+
+ private final byte[] serFun;
+ private transient OutputSelector<PyObject> fun;
+
+ public PythonOutputSelector(OutputSelector<PyObject> fun) throws IOException {
+ this.serFun = SerializationUtils.serializeObject(fun);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Iterable<String> select(PyObject value) {
+ if (this.fun == null) {
+ try {
+ this.fun = (OutputSelector<PyObject>) SerializationUtils.deserializeObject(this.serFun);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
The thing is that the signature of the `select` function in `OutputSelector` interface is without any exception throw declaration. This is why I catch the given exceptions here and add a check in the following lines to test whether the variable `this.fun` is null. If it is `null`, then the function returns null. How can I do it otherwise?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on the issue:
https://github.com/apache/flink/pull/3838
I'm trying to track down the root cause for the checks failures without a success. Obviously, the given project (flink-libraries/flink-streaming-python) in master branch passes the `verify` with success in my environment.
Please advise,
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121888394
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
--- End diff --
this import can be removed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121890065
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
+ final private static String pathToStreamingTests = "src/test/python/org/apache/flink/streaming/python/api";
+
+ public PythonStreamBinderTest() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (args.length == 0) {
+ args = prepareDefaultArgs();
+ } else {
+ args[0] = findStreamTestFile(args[0]).getAbsolutePath();
+ }
+ PythonStreamBinder.main(args);
+ }
+
+ @Override
+ public void testProgram() throws Exception {
+ this.main(new String[]{});
+ }
+
+ private static String[] prepareDefaultArgs() throws Exception {
+ File testFullPath = findStreamTestFile(defaultPythonScriptName);
--- End diff --
Since we include this in the list of scripts to run, does that mean that we run all tests twice?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121888058
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,104 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.4-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_${scala.binary.version}</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.python</groupId>
+ <artifactId>jython-standalone</artifactId>
+ <version>2.7.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
+ <version>1.2.0</version>
--- End diff --
replace version with `${project.version}`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
The check failed because the spotbugs plugin found something; this plugin isn't run by default when calling `mvn verify`. You can run the spotbugs locally by adding `-Dspotbugs` to the maven invocation.
The found problem is the PythonEnvironmentConfig class, which contains public static non-final fields. I propose making these non-static and explicitly pass around a config object where needed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121888454
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
--- End diff --
These modifiers should be re-ordered to `private static final`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121888620
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
+ final private static String pathToStreamingTests = "src/test/python/org/apache/flink/streaming/python/api";
+
+ public PythonStreamBinderTest() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (args.length == 0) {
+ args = prepareDefaultArgs();
+ } else {
+ args[0] = findStreamTestFile(args[0]).getAbsolutePath();
+ }
+ PythonStreamBinder.main(args);
+ }
+
+ @Override
+ public void testProgram() throws Exception {
+ this.main(new String[]{});
--- End diff --
you should call `main` directly without referencing `this`, otherwise one would assume that is a non-static method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
It may take a while until i can review this; the 1.3 feature freeze just kicked in and it's time to test all the new features in depth.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133154903
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
+ final private static String pathToStreamingTests = "src/test/python/org/apache/flink/streaming/python/api";
+
+ public PythonStreamBinderTest() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (args.length == 0) {
+ args = prepareDefaultArgs();
+ } else {
+ args[0] = findStreamTestFile(args[0]).getAbsolutePath();
+ }
+ PythonStreamBinder.main(args);
+ }
+
+ @Override
+ public void testProgram() throws Exception {
+ this.main(new String[]{});
+ }
+
+ private static String[] prepareDefaultArgs() throws Exception {
+ File testFullPath = findStreamTestFile(defaultPythonScriptName);
--- End diff --
Instead I've just provided a simplified version of it. Will send for review.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121889286
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
+ final private static String pathToStreamingTests = "src/test/python/org/apache/flink/streaming/python/api";
+
+ public PythonStreamBinderTest() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (args.length == 0) {
+ args = prepareDefaultArgs();
+ } else {
+ args[0] = findStreamTestFile(args[0]).getAbsolutePath();
+ }
+ PythonStreamBinder.main(args);
+ }
+
+ @Override
+ public void testProgram() throws Exception {
+ this.main(new String[]{});
+ }
+
+ private static String[] prepareDefaultArgs() throws Exception {
+ File testFullPath = findStreamTestFile(defaultPythonScriptName);
+ List<String> filesInTestPath = getFilesInFolder(testFullPath.getParent());
+
+ String[] args = new String[filesInTestPath.size() + 1];
+ args[0] = testFullPath.getAbsolutePath();
+
+ for (final ListIterator<String> it = filesInTestPath.listIterator(); it.hasNext();) {
+ final String p = it.next();
+ args[it.previousIndex() + 1] = p;
+ }
+ return args;
+ }
+
+ private static File findStreamTestFile(String name) throws Exception {
+ if (new File(name).exists()) {
+ return new File(name);
+ }
+ FileSystem fs = FileSystem.getLocalFileSystem();
+ String workingDir = fs.getWorkingDirectory().getPath();
+ if (!workingDir.endsWith(flinkPythonRltvPath)) {
+ workingDir += File.separator + flinkPythonRltvPath;
+ }
+ FileStatus[] status = fs.listStatus(
+ new Path( workingDir + File.separator + pathToStreamingTests));
+ for (FileStatus f : status) {
+ String file_name = f.getPath().getName();
+ if (file_name.equals(name)) {
+ return new File(f.getPath().getPath());
+ }
+ }
+ throw new FileNotFoundException();
+ }
+
+ private static List<String> getFilesInFolder(String path) {
--- End diff --
I would rename this to `getTestFilesInFolder` and change the signature to accept a `Fileยด and return a `List<File>`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117725288
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
+public class PythonStreamExecutionEnvironment {
+ private final StreamExecutionEnvironment env;
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment();
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(config);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(parallelism, config);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, jar_files);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, config, jar_files);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, parallelism, jar_files);
+ }
+
+ private PythonStreamExecutionEnvironment() {
+ this.env = StreamExecutionEnvironment.getExecutionEnvironment();
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(Configuration config) {
+ this.env = new LocalStreamEnvironment(config);
--- End diff --
I would throw out all constructors and add one that takes a `StreamExecutionEnvironmen` parameter. We already have the case distinction in the various `create_*` methods.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
I'm gonna take a deeper look now, and play around with it a bit.
Be aware that if you rebase the branch again you will hit a myriad of checkstyle violations, we will have to suppress the checks for method names for this module.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121887994
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,104 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.4-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_${scala.binary.version}</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.python</groupId>
+ <artifactId>jython-standalone</artifactId>
+ <version>2.7.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
--- End diff --
replace `2.10` with `${scala.binary.version`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121889539
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
+ final private static String pathToStreamingTests = "src/test/python/org/apache/flink/streaming/python/api";
+
+ public PythonStreamBinderTest() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (args.length == 0) {
+ args = prepareDefaultArgs();
+ } else {
+ args[0] = findStreamTestFile(args[0]).getAbsolutePath();
+ }
+ PythonStreamBinder.main(args);
+ }
+
+ @Override
+ public void testProgram() throws Exception {
+ this.main(new String[]{});
+ }
+
+ private static String[] prepareDefaultArgs() throws Exception {
+ File testFullPath = findStreamTestFile(defaultPythonScriptName);
+ List<String> filesInTestPath = getFilesInFolder(testFullPath.getParent());
+
+ String[] args = new String[filesInTestPath.size() + 1];
+ args[0] = testFullPath.getAbsolutePath();
+
+ for (final ListIterator<String> it = filesInTestPath.listIterator(); it.hasNext();) {
+ final String p = it.next();
+ args[it.previousIndex() + 1] = p;
+ }
+ return args;
+ }
+
+ private static File findStreamTestFile(String name) throws Exception {
+ if (new File(name).exists()) {
+ return new File(name);
+ }
+ FileSystem fs = FileSystem.getLocalFileSystem();
+ String workingDir = fs.getWorkingDirectory().getPath();
+ if (!workingDir.endsWith(flinkPythonRltvPath)) {
+ workingDir += File.separator + flinkPythonRltvPath;
+ }
+ FileStatus[] status = fs.listStatus(
+ new Path( workingDir + File.separator + pathToStreamingTests));
+ for (FileStatus f : status) {
+ String file_name = f.getPath().getName();
--- End diff --
should be named `fileName`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117722927
--- Diff: docs/dev/stream/python.md ---
@@ -0,0 +1,649 @@
+---
+title: "Python Programming Guide (Streaming)"
+is_beta: true
+nav-title: Python API
+nav-parent_id: streaming
+nav-pos: 63
+---
+<!--
+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.
+-->
+
+Analysis streaming programs in Flink are regular programs that implement transformations on
+streaming data sets (e.g., filtering, mapping, joining, grouping). The streaming data sets are initially
+created from certain sources (e.g., by reading from Apache Kafka, or reading files, or from collections).
+Results are returned via sinks, which may for example write the data to (distributed) files, or to
+standard output (for example the command line terminal). Flink streaming programs run in a variety
+of contexts, standalone, or embedded in other programs. The execution can happen in a local JVM, or
+on clusters of many machines.
+
+In order to create your own Flink streaming program, we encourage you to start with the
+[program skeleton](#program-skeleton) and gradually add your own
+[transformations](#transformations). The remaining sections act as references for additional
+operations and advanced features.
+
+* This will be replaced by the TOC
+{:toc}
+
+Jython Framework
+---------------
+Flink Python streaming API uses Jython framework (see <http://www.jython.org/archive/21/docs/whatis.html>)
+to drive the execution of a given script. The Python streaming layer, is actually a thin wrapper layer for the
+existing Java streaming APIs.
+
+#### Constraints
+There are two main constraints for using Jython:
+
+* The latest Python supported version is 2.7
+* It is not straightforward to use Python C extensions
+
+Streaming Program Example
+-------------------------
+The following streaming program is a complete, working example of WordCount. You can copy & paste the code
+to run it locally (see notes later in this section). It counts the number of each word (case insensitive)
+in a stream of sentences, on a window size of 50 milliseconds and prints the results into the standard output.
+
+{% highlight python %}
+from org.apache.flink.streaming.api.functions.source import SourceFunction
+from org.apache.flink.api.common.functions import FlatMapFunction, ReduceFunction
+from org.apache.flink.api.java.functions import KeySelector
+from org.apache.flink.python.api.jython import PythonStreamExecutionEnvironment
+from org.apache.flink.streaming.api.windowing.time.Time import milliseconds
+
+
+class Generator(SourceFunction):
+ def __init__(self, num_iters):
+ self._running = True
+ self._num_iters = num_iters
+
+ def run(self, ctx):
+ counter = 0
+ while self._running and counter < self._num_iters:
+ ctx.collect('Hello World')
+ counter += 1
+
+ def cancel(self):
+ self._running = False
+
+
+class Tokenizer(FlatMapFunction):
+ def flatMap(self, value, collector):
+ for word in value.lower().split():
+ collector.collect((1, word))
+
+
+class Selector(KeySelector):
+ def getKey(self, input):
+ return input[1]
+
+
+class Sum(ReduceFunction):
+ def reduce(self, input1, input2):
+ count1, word1 = input1
+ count2, word2 = input2
+ return (count1 + count2, word1)
+
+def main():
+ env = PythonStreamExecutionEnvironment.get_execution_environment()
+ env.create_python_source(Generator(num_iters=1000)) \
+ .flat_map(Tokenizer()) \
+ .key_by(Selector()) \
+ .time_window(milliseconds(50)) \
+ .reduce(Sum()) \
+ .print()
+ env.execute()
+
+
+if __name__ == '__main__':
+ main()
+{% endhighlight %}
+
+**Notes:**
+
+- If execution is done on a local cluster, you may replace the last line in the `main()` function
+ with **`env.execute(True)`**
+- Execution on a multi-node cluster requires a shared medium storage, which needs to be configured (.e.g HDFS)
+ upfront.
+- The output from of the given script is directed to the standard output. Consequently, the output
+ is written to the corresponding worker `.out` filed. If the script is executed inside the IntelliJ IDE,
+ then the output will be displayed in the console tab.
+
+{% top %}
+
+Program Skeleton
+----------------
+As we already saw in the example, Flink streaming programs look like regular Python programs.
+Each program consists of the same basic parts:
+
+1. A `main()` function definition, without arguments - the program entry point,
+2. Obtain an `Environment`,
+3. Load/create the initial data,
+4. Specify transformations on this data,
+5. Specify where to put the results of your computations, and
+5. Execute your program.
+
+We will now give an overview of each of those steps but please refer to the respective sections for
+more details.
+
+The `main()` function is a must and it is used by Flink execution layer to run the
+given Python streaming program.
+
+The `Environment` is the basis for all Flink programs. You can
+obtain one using these static methods on class `PythonStreamExecutionEnvironment`:
+
+{% highlight python %}
+get_execution_environment()
+{% endhighlight %}
+
+For specifying data sources the streaming execution environment has several methods.
+To just read a text file as a sequence of lines, you can use:
+
+{% highlight python %}
+env = get_execution_environment()
+text = env.read_text_file("file:///path/to/file")
+{% endhighlight %}
+
+This will give you a DataStream on which you can then apply transformations. For
+more information on data sources and input formats, please refer to
+[Data Sources](#data-sources).
+
+Once you have a DataStream you can apply transformations to create a new
+DataStream which you can then write to a file, transform again, or
+combine with other DataStreams. You apply transformations by calling
+methods on DataStream with your own custom transformation function. For example,
+a map transformation looks like this:
+
+{% highlight python %}
+class Doubler(MapFunction):
+ def map(self, value):
+ return value * 2
+
+data.map(Doubler())
+{% endhighlight %}
+
+This will create a new DataStream by doubling every value in the original DataStream.
+For more information and a list of all the transformations,
+please refer to [Transformations](#transformations).
+
+Once you have a DataStream that needs to be written to disk you can call one
+of these methods on DataStream:
+
+{% highlight python %}
+data.write_as_text("<file-path>")
+data.write_as_text("<file-path>", mode=WriteMode.OVERWRITE)
+data.print()
+{% endhighlight %}
+
+The last method is only useful for developing/debugging on a local machine,
+it will output the contents of the DataSet to standard output. (Note that in
+a cluster, the result goes to the standard out stream of the cluster nodes and ends
+up in the *.out* files of the workers).
+The first two do as the name suggests.
+Please refer to [Data Sinks](#data-sinks) for more information on writing to files.
+
+Once you specified the complete program you need to call `execute` on
+the `Environment`. This will either execute on your local machine or submit your program
+for execution on a cluster, depending on how Flink was started. You can force
+a local execution by using `execute(True)`.
+
+{% top %}
+
+Project setup
+---------------
+
+Apart from setting up Flink, no additional work is required. Using Jython to execute the Python
+script, means that no external packages are needed and the program is executed as if it was a jar file.
+
+The Python API was tested on Linux/OSX systems.
+
+{% top %}
+
+Lazy Evaluation
+---------------
+
+All Flink programs are executed lazily: When the program's main method is executed, the data loading
+and transformations do not happen directly. Rather, each operation is created and added to the
+program's plan. The operations are actually executed when one of the `execute()` methods is invoked
+on the Environment object. Whether the program is executed locally or on a cluster depends
+on the environment of the program.
+
+The lazy evaluation lets you construct sophisticated programs that Flink executes as one
+holistically planned unit.
+
+{% top %}
+
+Transformations
+---------------
+
+Data transformations transform one or more DataStreams into a new DataStream. Programs can combine
+multiple transformations into sophisticated assemblies.
+
+This section gives a brief overview of the available transformations. The [transformations
+documentation](dataset_transformations.html) has a full description of all transformations with
+examples.
+
+<br />
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Transformation</th>
+ <th class="text-center">Description</th>
+ </tr>
+ </thead>
+
+ <tbody>
+ <tr>
+ <td><strong>Map</strong><br>PythonDataStream → PythonDataStream</td>
+ <td>
+ <p>Takes one element and produces one element.</p>
+{% highlight python %}
+class Doubler(MapFunction):
+ def map(self, value):
+ return value * 2
+
+data_stream.map(Doubler())
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>FlatMap</strong><br>PythonDataStream → PythonDataStream</td>
+ <td>
+ <p>Takes one element and produces zero, one, or more elements. </p>
+{% highlight python %}
+class Tokenizer(FlatMapFunction):
+ def flatMap(self, word, collector):
+ collector.collect((1, word))
+
+data_stream.flat_map(Tokenizer())
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Filter</strong><br>PythonDataStream → PythonDataStream</td>
+ <td>
+ <p>Evaluates a boolean function for each element and retains those for which the function
+ returns true.</p>
+{% highlight python %}
+class GreaterThen1000(FilterFunction):
+ def filter(self, value):
+ return value > 1000
+
+data_stream.filter(GreaterThen1000())
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>KeyBy</strong><br>PythonDataStream → PythonKeyedStream</td>
+ <td>
+ <p>Logically partitions a stream into disjoint partitions, each partition containing elements of the same key.
+ Internally, this is implemented with hash partitioning. See <a href="/dev/api_concepts#specifying-keys">keys</a> on how to specify keys.
+ This transformation returns a PythonKeyedDataStream.</p>
+ {% highlight python %}
+class Selector(KeySelector):
+ def getKey(self, input):
+ return input[1] # Key by the second element in a tuple
+
+data_stream.key_by(Selector()) // Key by field "someKey"
+ {% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Reduce</strong><br>PythonKeyedStream → PythonDataStream</td>
+ <td>
+ <p>A "rolling" reduce on a keyed data stream. Combines the current element with the last reduced value and
+ emits the new value.</p>
+{% highlight python %}
+class Sum(ReduceFunction):
+ def reduce(self, input1, input2):
+ count1, val1 = input1
+ count2, val2 = input2
+ return (count1 + count2, val1)
+
+data.reduce(Sum())
+{% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Window</strong><br>PythonKeyedStream → PythonWindowedStream</td>
+ <td>
+ <p>Windows can be defined on already partitioned KeyedStreams. Windows group the data in each
+ key according to some characteristic (e.g., the data that arrived within the last 5 seconds).
+ See <a href="windows.html">windows</a> for a complete description of windows.
+ {% highlight python %}
+keyed_stream.count_window(10, 5) # Last 10 elements, sliding (jumping) by 5 elements
+
+keyed_stream.time_window(milliseconds(30)) # Last 30 milliseconds of data
+
+keted_stream.time_window(milliseconds(100), milliseconds(20)) # Last 100 milliseconds of data, sliding (jumping) by 20 milliseconds
+ {% endhighlight %}
+ </p>
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Window Apply</strong><br>PythonWindowedStream → PythonDataStream</td>
+ <td>
+ <p>Applies a general function to the window as a whole. Below is a function that manually sums
+ the elements of a window.</p>
+ {% highlight python %}
+class WindowSum(WindowFunction):
+ def apply(self, key, window, values, collector):
+ sum = 0
+ for value in values:
+ sum += value[0]
+ collector.collect((key, sum))
+
+windowed_stream.apply(WindowSum())
+ {% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Window Reduce</strong><br>PythonWindowedStream → PythonDataStream</td>
+ <td>
+ <p>Applies a functional reduce function to the window and returns the reduced value.</p>
+ {% highlight python %}
+class Sum(ReduceFunction):
+ def reduce(self, input1, input2):
+ count1, val1 = input1
+ count2, val2 = input2
+ return (count1 + count2, val1)
+
+windowed_stream.reduce(Sum())
+ {% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Union</strong><br>PythonDataStream* → PythonDataStream</td>
+ <td>
+ <p>Union of two or more data streams creating a new stream containing all the elements from all
+ the streams. Note: If you union a data stream with itself you will get each element twice
+ in the resulting stream.</p>
+ {% highlight python %}
+data_stream.union(other_stream1, other_stream2, ...);
+ {% endhighlight %}
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Split</strong><br>PythonDataStream → PythonSplitStream</td>
+ <td>
+ <p>Split the stream into two or more streams according to some criterion.
+ {% highlight python %}
+class StreamSelector(OutputSelector):
+ def select(self, value):
+ return ["even"] if value % 2 == 0 else ["odd"]
+
+splited_stream = data_stream.split(StreamSelector())
+ {% endhighlight %}
+ </p>
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Select</strong><br>SplitStream → DataStream</td>
+ <td>
+ <p> Select one or more streams from a split stream.
+ {% highlight python %}
+even_data_stream = splited_stream.select("even")
+odd_data_stream = splited_stream.select("odd")
+all_data_stream = splited_stream.select("even", "odd")
+ {% endhighlight %}
+ </p>
+ </td>
+ </tr>
+
+ <tr>
+ <td><strong>Iterate</strong><br>PythonDataStream → PythonIterativeStream → PythonDataStream</td>
+ <td>
+ <p> Creates a "feedback" loop in the flow, by redirecting the output of one operator
+ to some previous operator. This is especially useful for defining algorithms that
+ continuously update a model. The following code starts with a stream and applies
+ the iteration body continuously. Elements that are greater than 0 are sent back
+ to the feedback channel, and the rest of the elements are forwarded downstream.
+ See <a href="#iterations">iterations</a> for a complete description.
+ {% highlight java %}
+class MinusOne(MapFunction):
+ def map(self, value):
+ return value - 1
+
+class PositiveNumber(FilterFunction):
+ def filter(self, value):
+ return value > 0
+
+class LessEquelToZero(FilterFunction):
+ def filter(self, value):
+ return value <= 0
+
+iteration = initial_stream.iterate(5000)
+iteration_body = iteration.map(MinusOne())
+feedback = iteration_body.filter(PositiveNumber())
+iteration.close_with(feedback)
+output = iteration_body.filter(LessEquelToZero())
+ {% endhighlight %}
+ </p>
+ </td>
+ </tr>
+
+ </tbody>
+</table>
+
+{% top %}
+
+Passing Functions to Flink
+--------------------------
+
+Certain operations require user-defined functions as arguments. All the functions should be
+defined as Python classes that derived from the relevant Flink function. User-defined functions
+are serialized and sent over to the TaskManagers for execution.
+
+{% highlight python %}
+class Filter(FilterFunction):
+ def filter(self, value):
+ return value > 5
+
+data_stream.filter(Filter())
+{% endhighlight %}
+
+Rich functions (.e.g `RichFileterFunction`) enable to define (override) the optional operations: `open` & `close`.
--- End diff --
type: Fileter -> Filter
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121891478
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@PublicEvolving
+public class PythonStreamExecutionEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+ private final StreamExecutionEnvironment env;
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
--- End diff --
the method parameter is named `config`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117726359
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java ---
@@ -233,6 +234,20 @@ boolean holdsStillReference(String name, JobID jobId) {
// Utilities
// ------------------------------------------------------------------------
+ /**
+ * Remove a given path recursively if exists. The path can be a filepath or directory.
+ *
+ * @param path The root path to remove.
+ * @throws IOException
+ * @throws URISyntaxException
+ */
+ public static void clearPath(String path) throws IOException, URISyntaxException {
--- End diff --
I thought we decided against adding this method?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117722682
--- Diff: docs/dev/stream/python.md ---
@@ -0,0 +1,649 @@
+---
+title: "Python Programming Guide (Streaming)"
+is_beta: true
+nav-title: Python API
+nav-parent_id: streaming
+nav-pos: 63
+---
+<!--
+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.
+-->
+
+Analysis streaming programs in Flink are regular programs that implement transformations on
+streaming data sets (e.g., filtering, mapping, joining, grouping). The streaming data sets are initially
+created from certain sources (e.g., by reading from Apache Kafka, or reading files, or from collections).
+Results are returned via sinks, which may for example write the data to (distributed) files, or to
+standard output (for example the command line terminal). Flink streaming programs run in a variety
+of contexts, standalone, or embedded in other programs. The execution can happen in a local JVM, or
+on clusters of many machines.
+
+In order to create your own Flink streaming program, we encourage you to start with the
+[program skeleton](#program-skeleton) and gradually add your own
+[transformations](#transformations). The remaining sections act as references for additional
+operations and advanced features.
+
+* This will be replaced by the TOC
+{:toc}
+
+Jython Framework
+---------------
+Flink Python streaming API uses Jython framework (see <http://www.jython.org/archive/21/docs/whatis.html>)
+to drive the execution of a given script. The Python streaming layer, is actually a thin wrapper layer for the
+existing Java streaming APIs.
+
+#### Constraints
+There are two main constraints for using Jython:
+
+* The latest Python supported version is 2.7
+* It is not straightforward to use Python C extensions
+
+Streaming Program Example
+-------------------------
+The following streaming program is a complete, working example of WordCount. You can copy & paste the code
+to run it locally (see notes later in this section). It counts the number of each word (case insensitive)
+in a stream of sentences, on a window size of 50 milliseconds and prints the results into the standard output.
+
+{% highlight python %}
+from org.apache.flink.streaming.api.functions.source import SourceFunction
+from org.apache.flink.api.common.functions import FlatMapFunction, ReduceFunction
+from org.apache.flink.api.java.functions import KeySelector
+from org.apache.flink.python.api.jython import PythonStreamExecutionEnvironment
+from org.apache.flink.streaming.api.windowing.time.Time import milliseconds
+
+
+class Generator(SourceFunction):
+ def __init__(self, num_iters):
+ self._running = True
+ self._num_iters = num_iters
+
+ def run(self, ctx):
+ counter = 0
+ while self._running and counter < self._num_iters:
+ ctx.collect('Hello World')
+ counter += 1
+
+ def cancel(self):
+ self._running = False
+
+
+class Tokenizer(FlatMapFunction):
+ def flatMap(self, value, collector):
+ for word in value.lower().split():
+ collector.collect((1, word))
+
+
+class Selector(KeySelector):
+ def getKey(self, input):
+ return input[1]
+
+
+class Sum(ReduceFunction):
+ def reduce(self, input1, input2):
+ count1, word1 = input1
+ count2, word2 = input2
+ return (count1 + count2, word1)
+
+def main():
+ env = PythonStreamExecutionEnvironment.get_execution_environment()
+ env.create_python_source(Generator(num_iters=1000)) \
+ .flat_map(Tokenizer()) \
+ .key_by(Selector()) \
+ .time_window(milliseconds(50)) \
+ .reduce(Sum()) \
+ .print()
+ env.execute()
+
+
+if __name__ == '__main__':
+ main()
+{% endhighlight %}
+
+**Notes:**
+
+- If execution is done on a local cluster, you may replace the last line in the `main()` function
+ with **`env.execute(True)`**
+- Execution on a multi-node cluster requires a shared medium storage, which needs to be configured (.e.g HDFS)
+ upfront.
+- The output from of the given script is directed to the standard output. Consequently, the output
+ is written to the corresponding worker `.out` filed. If the script is executed inside the IntelliJ IDE,
--- End diff --
type: filed -> file
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r132843982
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@PublicEvolving
+public class PythonStreamExecutionEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+ private final StreamExecutionEnvironment env;
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(new LocalStreamEnvironment(config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createLocalEnvironment(parallelism, config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files));
+ }
+
+ private PythonStreamExecutionEnvironment(StreamExecutionEnvironment env) {
+ this.env = env;
+ this.registerJythonSerializers();
+ }
+
+ private void registerJythonSerializers() {
+ this.env.registerTypeWithKryoSerializer(PyString.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInteger.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyLong.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
+ }
+
+ public PythonDataStream create_python_source(SourceFunction<Object> src) throws Exception {
--- End diff --
Probably yes. All Jython user-defined functions/Classes are generated under `org.python.proxies` scope. So, using the class canonical name, we can identify these functions. WDYT?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r132967578
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java ---
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.python.util.PythonObjectInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+
+/**
+ * A helper class to overcome the inability to set the serialVersionUID in a python user-defined
+ * function (UDF).
+ * <p>The fact the this field is not set, results in a dynamic calculation of this serialVersionUID,
+ * using SHA, to make sure it is a unique number. This unique number is a 64-bit hash of the
+ * class name, interface class names, methods, and fields. If a Python class inherits from a
+ * Java class, as in the case of Python UDFs, then a proxy wrapper class is created. Its name is
+ * constructed using the following pattern:
+ * <b>{@code org.python.proxies.<module-name>$<UDF-class-name>$<number>}</b>. The {@code <number>}
+ * part is increased by one in runtime, for every job submission. It results in different serial
+ * version UID for each run for the same Python class. Therefore, it is required to silently
+ * suppress the serial version UID mismatch check.</p>
+ */
+public class PythonObjectInputStream2 extends PythonObjectInputStream {
+
+ public PythonObjectInputStream2(InputStream in) throws IOException {
+ super(in);
+ }
+
+ protected ObjectStreamClass readClassDescriptor() throws IOException, ClassNotFoundException {
+ ObjectStreamClass resultClassDescriptor = super.readClassDescriptor(); // initially streams descriptor
+
+ Class<?> localClass;
+ try {
+ localClass = resolveClass(resultClassDescriptor);
+ } catch (ClassNotFoundException e) {
+ System.out.println("No local class for " + resultClassDescriptor.getName());
--- End diff --
The purpose of this function is to try to return the class according to the given description. If it fails, it probably means that the Jython interpreter was not initialised yet, and as a result it is initialised. This is handled in `org.apache.flink.streaming.python.api.functions.UtilityFunctions::smartFunctionDeserialization`.
I'm currently checking whether the `catch` here is redundant and it's a left-over from the debugging phase. We can probably let the exception be propagated up the call stack.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121898935
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonKeyedStream.java ---
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.streaming.api.datastream.KeyedStream;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.streaming.python.api.functions.PyKey;
+import org.apache.flink.streaming.python.api.functions.PythonReduceFunction;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+
+/**
+ * A thin wrapper layer over {@link KeyedStream}.
+ *
+ * <p>A {@code PythonKeyedStream} represents a {@link PythonDataStream} on which operator state is
+ * partitioned by key using a provided {@link org.apache.flink.api.java.functions.KeySelector;}</p>
+ */
+@Public
+public class PythonKeyedStream extends PythonDataStream<KeyedStream<PyObject, PyKey>> {
+
+ public PythonKeyedStream(KeyedStream<PyObject, PyKey> stream) {
--- End diff --
we can remove the public modifier; this applied to other variations of `PythonDataStream` as well, exception `PythonDataStream` itself.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121899652
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java ---
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.python.util.PythonObjectInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+
+/**
+ * A helper class to overcome the inability to set the serialVersionUID in a python user-defined
+ * function (UDF).
+ * <p>The fact the this field is not set, results in a dynamic calculation of this serialVersionUID,
+ * using SHA, to make sure it is a unique number. This unique number is a 64-bit hash of the
+ * class name, interface class names, methods, and fields. If a Python class inherits from a
+ * Java class, as in the case of Python UDFs, then a proxy wrapper class is created. Its name is
+ * constructed using the following pattern:
+ * <b>{@code org.python.proxies.<module-name>$<UDF-class-name>$<number>}</b>. The {@code <number>}
+ * part is increased by one in runtime, for every job submission. It results in different serial
+ * version UID for each run for the same Python class. Therefore, it is required to silently
+ * suppress the serial version UID mismatch check.</p>
+ */
+public class PythonObjectInputStream2 extends PythonObjectInputStream {
+
+ public PythonObjectInputStream2(InputStream in) throws IOException {
+ super(in);
+ }
+
+ protected ObjectStreamClass readClassDescriptor() throws IOException, ClassNotFoundException {
+ ObjectStreamClass resultClassDescriptor = super.readClassDescriptor(); // initially streams descriptor
+
+ Class<?> localClass;
+ try {
+ localClass = resolveClass(resultClassDescriptor);
+ } catch (ClassNotFoundException e) {
+ System.out.println("No local class for " + resultClassDescriptor.getName());
--- End diff --
Remove `println`. If an exception is thrown, what exactly does that mean to the job execution? I.e., why don't we fail here completely?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133132675
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@PublicEvolving
+public class PythonStreamExecutionEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+ private final StreamExecutionEnvironment env;
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(new LocalStreamEnvironment(config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createLocalEnvironment(parallelism, config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files));
+ }
+
+ private PythonStreamExecutionEnvironment(StreamExecutionEnvironment env) {
+ this.env = env;
+ this.registerJythonSerializers();
+ }
+
+ private void registerJythonSerializers() {
+ this.env.registerTypeWithKryoSerializer(PyString.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInteger.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyLong.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
+ }
+
+ public PythonDataStream create_python_source(SourceFunction<Object> src) throws Exception {
--- End diff --
sounds good to me.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724857
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
--- End diff --
mark as `@PublicEvolving`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724828
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
+public class PythonStreamExecutionEnvironment {
+ private final StreamExecutionEnvironment env;
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
--- End diff --
static fields should be above non-static fields.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121893907
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@PublicEvolving
+public class PythonStreamExecutionEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+ private final StreamExecutionEnvironment env;
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(new LocalStreamEnvironment(config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createLocalEnvironment(parallelism, config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files));
+ }
+
+ private PythonStreamExecutionEnvironment(StreamExecutionEnvironment env) {
+ this.env = env;
+ this.registerJythonSerializers();
+ }
+
+ private void registerJythonSerializers() {
+ this.env.registerTypeWithKryoSerializer(PyString.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInteger.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyLong.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
+ }
+
+ public PythonDataStream create_python_source(SourceFunction<Object> src) throws Exception {
--- End diff --
Is there a way for to determine internally whether a function is java or jython function? This seems a bit brittle to me and leaks implementation details.
Besides that a javadoc is missing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117723487
--- Diff: flink-libraries/flink-streaming-python/pom.xml ---
@@ -0,0 +1,99 @@
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-libraries</artifactId>
+ <version>1.3-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-streaming-python_2.10</artifactId>
+ <name>flink-streaming-python</name>
+ <packaging>jar</packaging>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ <version>${project.version}</version>
--- End diff --
all flink dependencies should be marked as `<provided>ยด.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121900659
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/util/serialization/PyObjectSerializer.java ---
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.util.serialization;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * A serializer implementation for PyObject class type. Is is used by the Kryo serialization
+ * framework. {@see https://github.com/EsotericSoftware/kryo#serializers}
+ */
+public class PyObjectSerializer extends Serializer<PyObject> {
+
+ public void write (Kryo kryo, Output output, PyObject po) {
--- End diff --
remove space after `write`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117723282
--- Diff: flink-libraries/flink-python/src/test/python/org/apache/flink/python/api/test_main.py ---
@@ -15,18 +15,18 @@
# 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.FlatMapFunction import FlatMapFunction
+import struct
from flink.functions.FilterFunction import FilterFunction
+from flink.functions.FlatMapFunction import FlatMapFunction
+from flink.functions.GroupReduceFunction import GroupReduceFunction
+from flink.functions.MapFunction import MapFunction
--- End diff --
should be reverted.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on the issue:
https://github.com/apache/flink/pull/3838
In the last change, I've rebased locally on top of origin/master, so I did `git push -f` to the master branch in my fork.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117725620
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
+public class PythonStreamExecutionEnvironment {
+ private final StreamExecutionEnvironment env;
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment();
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(config);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(parallelism, config);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, jar_files);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, config, jar_files);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(host, port, parallelism, jar_files);
+ }
+
+ private PythonStreamExecutionEnvironment() {
+ this.env = StreamExecutionEnvironment.getExecutionEnvironment();
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(Configuration config) {
+ this.env = new LocalStreamEnvironment(config);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(int parallelism, Configuration config) {
+ this.env = StreamExecutionEnvironment.createLocalEnvironment(parallelism, config);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(String host, int port, String... jar_files) {
+ this.env = StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(String host, int port, Configuration config, String... jar_files) {
+ this.env = StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files);
+ this.registerJythonSerializers();
+ }
+
+ private PythonStreamExecutionEnvironment(String host, int port, int parallelism, String... jar_files) {
+ this.env = StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files);
+ this.registerJythonSerializers();
+ }
+
+ private void registerJythonSerializers() {
+ this.env.registerTypeWithKryoSerializer(PyString.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInteger.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyLong.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
+ }
+
+ public PythonDataStream create_python_source(SourceFunction<Object> src) throws Exception {
+ return new PythonDataStream<>(env.addSource(new PythonGeneratorFunction(src)).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * Add a java source to the streaming topology. The source expected to be an java based
+ * implementation (.e.g. Kafka connector).
+ *
+ * @param src A native java source (e.g. PythonFlinkKafkaConsumer09)
+ * @return Python data stream
+ */
+ public PythonDataStream add_java_source(SourceFunction<Object> src) {
+ return new PythonDataStream<>(env.addSource(src).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#fromElements(java.lang.Object[])}
+ *
+ * @param elements
+ * The array of PyObject elements to create the data stream from.
+ * @return The data stream representing the given array of elements
+ */
+ public PythonDataStream from_elements(PyObject... elements) {
+ return new PythonDataStream<>(env.fromElements(elements));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#fromCollection(java.util.Collection)}
+ *
+ * <p>The input {@code Collection} is of type {@code Object}, because it is a collection
+ * of Python elements. * There type is determined in runtime, by the Jython framework.</p>
+ *
+ * @param collection
+ * The collection of python elements to create the data stream from.
+ * @return
+ * The data stream representing the given collection
+ */
+ public PythonDataStream from_collection(Collection<Object> collection) {
--- End diff --
How come these aren't typed to PyObject like `from_elements`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117807878
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java ---
@@ -233,6 +234,20 @@ boolean holdsStillReference(String name, JobID jobId) {
// Utilities
// ------------------------------------------------------------------------
+ /**
+ * Remove a given path recursively if exists. The path can be a filepath or directory.
+ *
+ * @param path The root path to remove.
+ * @throws IOException
+ * @throws URISyntaxException
+ */
+ public static void clearPath(String path) throws IOException, URISyntaxException {
--- End diff --
My bad! I don't use it anymore.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
The only usage i found is in `UtilityFunctions#smartFunctionDeserialization`, which is only called from various java UDF classes. Unless there is another usage hidden somewhere i would suggest to add a `PythonEnvironmentConfig` argument to the `smartFunctionDeserialization` method, and all UDF classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r132842489
--- Diff: flink-libraries/flink-streaming-python/src/test/java/org/apache/flink/streaming/python/api/PythonStreamBinderTest.java ---
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api;
+
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.python.api.PythonStreamBinder;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+public class PythonStreamBinderTest extends StreamingProgramTestBase {
+ final private static String defaultPythonScriptName = "run_all_tests.py";
+ final private static String flinkPythonRltvPath = "flink-libraries/flink-streaming-python";
+ final private static String pathToStreamingTests = "src/test/python/org/apache/flink/streaming/python/api";
+
+ public PythonStreamBinderTest() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ if (args.length == 0) {
+ args = prepareDefaultArgs();
+ } else {
+ args[0] = findStreamTestFile(args[0]).getAbsolutePath();
+ }
+ PythonStreamBinder.main(args);
+ }
+
+ @Override
+ public void testProgram() throws Exception {
+ this.main(new String[]{});
+ }
+
+ private static String[] prepareDefaultArgs() throws Exception {
+ File testFullPath = findStreamTestFile(defaultPythonScriptName);
--- End diff --
I agree that the function names here are a bit confusing - in essence this function locates a single test file, while the function in the next code line `getFilesInFolder` collects files that start with `test_`, thus the main test file `run_all_tests.py` will be filtered and not be included.
So, in order to be more readable and robust, I changed the `getFilesInFolder` to receive one more argument of `excludes` and call it with the main test file in the `excludes` argument.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zohar-mizrahi <gi...@git.apache.org>.
Github user zohar-mizrahi commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r133170150
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/functions/PythonOutputSelector.java ---
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.functions;
+
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.python.util.serialization.SerializationUtils;
+import org.python.core.PyObject;
+
+import java.io.IOException;
+
+/**
+ * The {@code PythonOutputSelector} is a thin wrapper layer over a Python UDF {@code OutputSelector}.
+ * It receives an {@code OutputSelector} as an input and keeps it internally in a serialized form.
+ * It is then delivered, as part of the job graph, up to the TaskManager, then it is opened and becomes
+ * a sort of mediator to the Python UDF {@code OutputSelector}.
+ *
+ * <p>This function is used internally by the Python thin wrapper layer over the streaming data
+ * functionality</p>
+ */
+public class PythonOutputSelector implements OutputSelector<PyObject> {
+ private static final long serialVersionUID = 909266346633598177L;
+
+ private final byte[] serFun;
+ private transient OutputSelector<PyObject> fun;
+
+ public PythonOutputSelector(OutputSelector<PyObject> fun) throws IOException {
+ this.serFun = SerializationUtils.serializeObject(fun);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Iterable<String> select(PyObject value) {
+ if (this.fun == null) {
+ try {
+ this.fun = (OutputSelector<PyObject>) SerializationUtils.deserializeObject(this.serFun);
+ } catch (IOException e) {
+ e.printStackTrace();
--- End diff --
Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117802052
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java ---
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.python.api.datastream;
+
+import org.python.util.PythonObjectInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+
+/**
+ * A helper class to overcome the inability to set the serialVersionUID in a python user-defined
+ * function (UDF).
+ * <p>The fact the this field is not set, results in a dynamic calculation of this serialVersionUID,
+ * using SHA, to make sure it is a unique number. This unique number is a 64-bit hash of the
+ * class name, interface class names, methods, and fields. If a Python class inherits from a
+ * Java class, as in the case of Python UDFs, then a proxy wrapper class is created. Its name is
+ * constructed using the following pattern:
+ * <b>{@code org.python.proxies.<module-name>$<UDF-class-name>$<number>}</b>. The {@code <number>}
+ * part is increased by one in runtime, for every job submission. It results in different serial
+ * version UID for each run for the same Python class. Therefore, it is required to silently
+ * suppress the serial version UID mismatch check.</p>
+ */
+public class PythonObjectInputStream2 extends PythonObjectInputStream {
--- End diff --
Ah didn't see that; figured it was a leftover from the development.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r121894570
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@PublicEvolving
+public class PythonStreamExecutionEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+ private final StreamExecutionEnvironment env;
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(new LocalStreamEnvironment(config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}
+ *
+ * @param parallelism
+ * The parallelism for the local environment.
+ * @param config
+ * Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createLocalEnvironment(parallelism, config));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(java.lang.String, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, Configuration, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config
+ * The configuration used by the client that connects to the remote cluster.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ *
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ * java.lang.String, int, int, java.lang.String...)}
+ *
+ * @param host
+ * The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port
+ * The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism
+ * The parallelism to use during the execution.
+ * @param jar_files
+ * The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files));
+ }
+
+ private PythonStreamExecutionEnvironment(StreamExecutionEnvironment env) {
+ this.env = env;
+ this.registerJythonSerializers();
+ }
+
+ private void registerJythonSerializers() {
+ this.env.registerTypeWithKryoSerializer(PyString.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInteger.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyLong.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
+ this.env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
+ }
+
+ public PythonDataStream create_python_source(SourceFunction<Object> src) throws Exception {
+ return new PythonDataStream<>(env.addSource(new PythonGeneratorFunction(src)).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * Add a java source to the streaming topology. The source expected to be an java based
+ * implementation (.e.g. Kafka connector).
+ *
+ * @param src A native java source (e.g. PythonFlinkKafkaConsumer09)
+ * @return Python data stream
+ */
+ public PythonDataStream add_java_source(SourceFunction<Object> src) {
+ return new PythonDataStream<>(env.addSource(src).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#fromElements(java.lang.Object[])}
+ *
+ * @param elements
+ * The array of PyObject elements to create the data stream from.
+ * @return The data stream representing the given array of elements
+ */
+ public PythonDataStream from_elements(PyObject... elements) {
+ return new PythonDataStream<>(env.fromElements(elements));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#fromCollection(java.util.Collection)}
+ *
+ * <p>The input {@code Collection} is of type {@code Object}, because it is a collection
+ * of Python elements. * There type is determined in runtime, by the Jython framework.</p>
+ *
+ * @param collection
+ * The collection of python elements to create the data stream from.
+ * @return
+ * The data stream representing the given collection
+ */
+ public PythonDataStream from_collection(Collection<Object> collection) {
+ return new PythonDataStream<>(env.fromCollection(collection).map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * Creates a python data stream from the given iterator.
+ *
+ * <p>Note that this operation will result in a non-parallel data stream source, i.e.,
+ * a data stream source with a parallelism of one.</p>
+ *
+ * @param iter
+ * The iterator of elements to create the data stream from
+ * @return The data stream representing the elements in the iterator
+ * @see StreamExecutionEnvironment#fromCollection(java.util.Iterator, org.apache.flink.api.common.typeinfo.TypeInformation)
+ */
+ public PythonDataStream from_collection(Iterator<Object> iter) throws Exception {
+ return new PythonDataStream<>(env.addSource(new PythonIteratorFunction(iter), TypeExtractor.getForClass(Object.class))
+ .map(new UtilityFunctions.SerializerMap<>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#generateSequence(long, long)}.
+ *
+ * @param from
+ * The number to start at (inclusive)
+ * @param to
+ * The number to stop at (inclusive)
+ * @return A python data stream, containing all number in the [from, to] interval
+ */
+ public PythonDataStream generate_sequence(long from, long to) {
+ return new PythonDataStream<>(env.generateSequence(from, to).map(new UtilityFunctions.SerializerMap<Long>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#readTextFile(java.lang.String)}.
+ *
+ * @param path
+ * The path of the file, as a URI (e.g., "file:///some/local/file" or "hdfs://host:port/file/path").
+ * @return The data stream that represents the data read from the given file as text lines
+ * @throws IOException
+ */
+
+ public PythonDataStream read_text_file(String path) throws IOException {
+ return new PythonDataStream<>(env.readTextFile(path).map(new UtilityFunctions.SerializerMap<String>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#socketTextStream(java.lang.String, int)}.
+ *
+ * @param host
+ * The host name which a server socket binds
+ * @param port
+ * The port number which a server socket binds. A port number of 0 means that the port number is automatically
+ * allocated.
+ * @return A python data stream containing the strings received from the socket
+ */
+ public PythonDataStream socket_text_stream(String host, int port) {
+ return new PythonDataStream<>(env.socketTextStream(host, port).map(new UtilityFunctions.SerializerMap<String>()));
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#enableCheckpointing(long)}.
+ *
+ * @param interval Time interval between state checkpoints in milliseconds.
+ * @return The same {@code PythonStreamExecutionEnvironment} instance of the caller
+ */
+ public PythonStreamExecutionEnvironment enable_checkpointing(long interval) {
+ this.env.enableCheckpointing(interval);
+ return this;
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#enableCheckpointing(long, CheckpointingMode)}.
+ *
+ * @param interval Time interval between state checkpoints in milliseconds.
+ * @param mode
+ * The checkpointing mode, selecting between "exactly once" and "at least once" guaranteed.
+ * @return The same {@code PythonStreamExecutionEnvironment} instance of the caller
+ */
+ public PythonStreamExecutionEnvironment enable_checkpointing(long interval, CheckpointingMode mode) {
+ this.env.enableCheckpointing(interval, mode);
+ return this;
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#setParallelism(int)}.
+ *
+ * @param parallelism The parallelism
+ * @return The same {@code PythonStreamExecutionEnvironment} instance of the caller
+ */
+ public PythonStreamExecutionEnvironment set_parallelism(int parallelism) {
+ this.env.setParallelism(parallelism);
+ return this;
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#execute()}.
+ *
+ * @return The result of the job execution
+ */
+ public JobExecutionResult execute() throws Exception {
+ return execute(false);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#execute()}.
+ *
+ * <p>In addition, it enables the caller to provide a hint about the execution mode - whether it is local
+ * or remote. In the case of local execution, the relevant cached files are distributed using the
+ * local machine temporary folder, otherwise a shared storage medium is used for this purpose.</p>
+ *
+ * @return The result of the job execution
+ */
+ public JobExecutionResult execute(Boolean local) throws Exception {
--- End diff --
use a primitive `boolean` instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3838#discussion_r117724977
--- Diff: flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonStreamExecutionEnvironment.java ---
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.python.api.environment;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.filecache.FileCache;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.python.api.datastream.PythonDataStream;
+import org.apache.flink.streaming.python.api.functions.PythonGeneratorFunction;
+import org.apache.flink.streaming.python.api.functions.PythonIteratorFunction;
+import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.util.serialization.PyObjectSerializer;
+import org.python.core.PyObject;
+import org.python.core.PyString;
+import org.python.core.PyInteger;
+import org.python.core.PyLong;
+import org.python.core.PyUnicode;
+import org.python.core.PyTuple;
+import org.python.core.PyObjectDerived;
+import org.python.core.PyInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Random;
+
+
+/**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment}.
+ *
+ * <p>The PythonStreamExecutionEnvironment is the context in which a streaming program is executed.
+ * </p>
+ *
+ * <p>The environment provides methods to control the job execution (such as setting the parallelism
+ * or the fault tolerance/checkpointing parameters) and to interact with the outside world
+ * (data access).</p>
+ */
+@Public
+public class PythonStreamExecutionEnvironment {
+ private final StreamExecutionEnvironment env;
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamExecutionEnvironment.class);
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public static PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment();
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link #setParallelism(int)}.
+ *
+ * @param configuration
+ * Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public static PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
--- End diff --
Don't we need a separate Configuration wrapper class?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3838: [FLINK-5886] Python API for streaming applications
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/3838
---