[FLINK-19097][python] Support add_jars() for Python StreamExecutionEnvironment. #13292

shuiqiangchen · 2020-09-01T07:16:23Z

What is the purpose of the change

Add add_jars() interface in Python StreamExecutionEnvironment to enable users to specify jar dependencies in their Python DataStream Job.

Brief change log

Added StreamExecutionEnvironment.add_jars() interface.

Verifying this change

This pull request has been tested by test_add_jars() in test_stream_execution_environment.py.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): ( no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? ( not documented)

flinkbot · 2020-09-01T07:18:21Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 81162c9 (Tue Sep 01 07:18:20 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-09-01T07:34:38Z

CI report:

d56c661 Azure: SUCCESS
7217e3a Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

HuangXingBo

@shuiqiangchen Thanks a lot for the PR. I have a question is why not add two methods that add_jars for pipeline.jars and add_class_paths for pipeline.classpaths

HuangXingBo · 2020-09-01T11:26:28Z

flink-python/pyflink/datastream/stream_execution_environment.py

+        jvm = get_gateway().jvm
+        jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
+        classpaths_key = jvm.org.apache.flink.configuration.PipelineOptions.CLASSPATHS.key()
+        if key in [jars_key, classpaths_key]:


Raise a exception if the key is not pipeline.jars or pipeline.classpaths

HuangXingBo · 2020-09-01T11:28:11Z

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -553,6 +553,29 @@ def set_python_executable(self, python_exec: str):
            .getEnvironmentConfig(self._j_stream_execution_environment)
        env_config.setString(jvm.PythonOptions.PYTHON_EXECUTABLE.key(), python_exec)

+    def add_jars(self, key: str, jars_path: str):


Why not add two methods that add_jars for pipeline.jars and add_class_paths for pipeline.classpaths

Yes, it will be more intuitive if we separate it into two interfaces named add_jars() and add_classpaths().

shuiqiangchen · 2020-09-02T02:20:22Z

Hi @HuangXingBo , thanks for your comments, I have updated the PR according to your suggestions, please have look at it.

HuangXingBo

@shuiqiangchen Thanks a lot for the update. It looks better now. I only left a minor comments.

HuangXingBo · 2020-09-02T09:13:23Z

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -553,6 +553,34 @@ def set_python_executable(self, python_exec: str):
            .getEnvironmentConfig(self._j_stream_execution_environment)
        env_config.setString(jvm.PythonOptions.PYTHON_EXECUTABLE.key(), python_exec)

+    def add_jars(self, jars_path: str):
+        """
+        Adds a list of jar files that contain the user-defined function (UDF) classes and all


I think add_jars is not only used in udf situations, maybe you need to change the annotation

HuangXingBo · 2020-09-02T09:14:24Z

flink-python/pyflink/datastream/stream_execution_environment.py

+    def add_classpaths(self, classpaths: str):
+        """
+        Adds a list of URLs that are added to the classpath of each user code classloader of the
+        program. Paths must specify a protocol (e.g. file://) and be accessible on all nodes


Suggested change

program. Paths must specify a protocol (e.g. file://) and be accessible on all nodes

program. Paths must specify a protocol (e.g. file://) and be accessible by all nodes

HuangXingBo · 2020-09-02T09:15:50Z

flink-python/pyflink/datastream/tests/test_stream_execution_environment.py

+        deserialization_schema = JsonRowDeserializationSchema.builder() \
+            .type_info(type_info=type_info).build()
+
+        # Will get a ClassNotFoundException if not add the kafka connector into the pipeline


Suggested change

# Will get a ClassNotFoundException if not add the kafka connector into the pipeline

# It will raise a ClassNotFoundException if the kafka connector is not added into the pipeline

dianfu

@shuiqiangchen Thanks for the PR. I notice that we have introduced two public API in this PR. Could you explain why not use configuration.set_string('pipeline.jars', 'xxx') just as the Python Table API?

shuiqiangchen · 2020-09-04T07:22:59Z

Hi @dianfu Thank you for your comments. Currently, users are not able to get/set Configurations for a StreamExecutionEnvironment since the StreamExecutionEnvironment.getConfiguration() is a protected method.
Here we have two options:

Making the StreamExecutionEnvironment.getConfiguration() public in Python DataStream API.
Add these two new interface that the Java StreamExecutionEnvironment does not have, this is actually a bit controversial.

dianfu

@shuiqiangchen Thanks for the explanation. It makes sense to me.

dianfu · 2020-09-04T11:43:05Z

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -553,6 +553,33 @@ def set_python_executable(self, python_exec: str):
            .getEnvironmentConfig(self._j_stream_execution_environment)
        env_config.setString(jvm.PythonOptions.PYTHON_EXECUTABLE.key(), python_exec)

+    def add_jars(self, jars_path: str):


Do you think it makes sense to change the interface a bit as following?

Suggested change

def add_jars(self, jars_path: str):

def add_jars(self, *jars: str):

Besides, we should also consider cases where users may call add_jars multiple times.

dianfu · 2020-09-04T11:43:12Z

flink-python/pyflink/datastream/stream_execution_environment.py

+            .getEnvironmentConfig(self._j_stream_execution_environment)
+        env_config.setString(jars_key, jars_path)
+
+    def add_classpaths(self, classpaths: str):


…vironment.

dianfu · 2020-09-06T14:01:20Z

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -553,6 +553,45 @@ def set_python_executable(self, python_exec: str):
            .getEnvironmentConfig(self._j_stream_execution_environment)
        env_config.setString(jvm.PythonOptions.PYTHON_EXECUTABLE.key(), python_exec)

+    def add_jars(self, *jars_path: List[str]):


I think the signature should be:

Suggested change

def add_jars(self, *jars_path: List[str]):

def add_jars(self, *jars_path: str):

What do you think?

dianfu · 2020-09-06T14:01:53Z

flink-python/pyflink/datastream/stream_execution_environment.py

+        jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
+        env_config = jvm.org.apache.flink.python.util.PythonConfigUtil \
+            .getEnvironmentConfig(self._j_stream_execution_environment)
+        set_jar_paths = env_config.getString(jars_key, None)


Why not as following?

Suggested change

set_jar_paths = env_config.getString(jars_key, None)

set_jar_paths = env_config.getString(jars_key, '')

dianfu · 2020-09-06T14:02:49Z

flink-python/pyflink/datastream/stream_execution_environment.py

+        if set_jar_paths is None:
+            set_jar_paths = ''
+        jars_path = jvm.PythonDependencyUtils.FILE_DELIMITER.join(*jars_path)
+        python_files = jvm.PythonDependencyUtils.FILE_DELIMITER.join([set_jar_paths,


Why named python_files?

Sorry, we could reuse the 'jars_path'

dianfu · 2020-09-06T14:03:39Z

flink-python/pyflink/datastream/stream_execution_environment.py

+        """
+        Adds a list of jar files that will be uploaded to the cluster and referenced by the job.
+
+        :param jars_path: Path of jars that delimited by ';'.


I think the comment isn't correct and should also be updated.

shuiqiangchen · 2020-09-06T14:29:37Z

@dianfu Thank you for your comments, I have updated the PR according to your suggestions, please have a look.

rmetzger added the review=description? label Sep 1, 2020

rmetzger added the component=API/Python label Sep 1, 2020

HuangXingBo reviewed Sep 1, 2020

View reviewed changes

HuangXingBo reviewed Sep 2, 2020

View reviewed changes

shuiqiangchen force-pushed the FLINK-19097-add-jar branch from 44b8754 to a7b36aa Compare September 3, 2020 03:01

dianfu reviewed Sep 4, 2020

View reviewed changes

shuiqiangchen added 4 commits September 6, 2020 17:15

[FLINK-19097][python] Support add_jars() for Python StreamExecutionEn…

0bdae80

…vironment.

[FLINK-19097][python] separate the add_jars() into two interfaces.

2b40bd8

[FLINK-19097][python] Resolve review.

6e639db

[FLINK-19097][python] Resolve review issues.

5e08985

shuiqiangchen force-pushed the FLINK-19097-add-jar branch from a7b36aa to 5e08985 Compare September 6, 2020 09:56

dianfu requested changes Sep 6, 2020

View reviewed changes

[FLINK-19097][python] Resolve review issues.

d56c661

[FLINK-19097][python] Resolve review issues.

7217e3a

dianfu approved these changes Sep 7, 2020

View reviewed changes

dianfu closed this in 71ffab6 Sep 7, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-19097][python] Support add_jars() for Python StreamExecutionEnvironment. #13292

[FLINK-19097][python] Support add_jars() for Python StreamExecutionEnvironment. #13292

shuiqiangchen commented Sep 1, 2020

flinkbot commented Sep 1, 2020

flinkbot commented Sep 1, 2020 •

edited

HuangXingBo left a comment

HuangXingBo Sep 1, 2020

HuangXingBo Sep 1, 2020

shuiqiangchen Sep 2, 2020

shuiqiangchen commented Sep 2, 2020

HuangXingBo left a comment

HuangXingBo Sep 2, 2020

HuangXingBo Sep 2, 2020

HuangXingBo Sep 2, 2020

dianfu left a comment

shuiqiangchen commented Sep 4, 2020 •

edited

dianfu left a comment

dianfu Sep 4, 2020

dianfu Sep 4, 2020

dianfu Sep 6, 2020

dianfu Sep 6, 2020

dianfu Sep 6, 2020

shuiqiangchen Sep 6, 2020

dianfu Sep 6, 2020

shuiqiangchen commented Sep 6, 2020

	program. Paths must specify a protocol (e.g. file://) and be accessible on all nodes
	program. Paths must specify a protocol (e.g. file://) and be accessible by all nodes

	# Will get a ClassNotFoundException if not add the kafka connector into the pipeline
	# It will raise a ClassNotFoundException if the kafka connector is not added into the pipeline

	def add_jars(self, jars_path: str):
	def add_jars(self, *jars: str):

	def add_jars(self, *jars_path: List[str]):
	def add_jars(self, *jars_path: str):

	set_jar_paths = env_config.getString(jars_key, None)
	set_jar_paths = env_config.getString(jars_key, '')

[FLINK-19097][python] Support add_jars() for Python StreamExecutionEnvironment. #13292

[FLINK-19097][python] Support add_jars() for Python StreamExecutionEnvironment. #13292

Conversation

shuiqiangchen commented Sep 1, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Sep 1, 2020

Automated Checks

Review Progress

flinkbot commented Sep 1, 2020 • edited

CI report:

HuangXingBo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shuiqiangchen commented Sep 2, 2020

HuangXingBo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dianfu left a comment

Choose a reason for hiding this comment

shuiqiangchen commented Sep 4, 2020 • edited

dianfu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shuiqiangchen commented Sep 6, 2020

flinkbot commented Sep 1, 2020 •

edited

shuiqiangchen commented Sep 4, 2020 •

edited