[FLINK-22052][python] Add FLIP-142 public classes to python API #15441

sjwiesman · 2021-03-31T01:16:38Z

What is the purpose of the change

This adds the new public APIs from FLIP-142 to the Python DataStream API. It is based on top of #15429 so only the last 3 commits are relevant.

Brief change log

96aa09c Add new state backend classes and sync updated JavaDoc

7193a21 Add new checkpoint storage classes

57c27b5 Add new methods to StreamExecutionEnvironment

Verifying this change

Added new Python Unit Tests. Note to the reviewer, I had some issues running the python tests locally. I will monitor the CI to ensure everything passes.

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

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

Documentation

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

flinkbot · 2021-03-31T01:19:10Z

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 43c1bd1 (Thu Sep 23 17:28:42 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{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 · 2021-03-31T01:51:11Z

CI report:

43c1bd1 Azure: SUCCESS

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

dianfu

@sjwiesman Thanks for the PR.

flink-python/pyflink/datastream/checkpoint_storage.py

dianfu · 2021-03-31T03:37:39Z

flink-python/pyflink/datastream/checkpoint_storage.py

+                file_state_size_threshold,
+                write_buffer_size)
+
+        self._j_filesystem_checkpoint_storage = j_filesystem_checkpoint_storage


_j_filesystem_checkpoint_storage could be removed

flink-python/pyflink/datastream/checkpoint_storage.py

flink-python/pyflink/datastream/state_backend.py

dianfu · 2021-03-31T03:55:08Z

flink-python/pyflink/datastream/state_backend.py

+        j_predefined_options = self._j_embedded_rocks_db_state_backend.getPredefinedOptions()
+        return PredefinedOptions._from_j_predefined_options(j_predefined_options)
+
+    def set_options(self, options_factory_class_name: str):


I guess it's not enough to only pass the class name of the options_factory. For example, for DefaultConfigurableOptionsFactory, there are many methods in it to configure it.

No, this is fine (it's actually copied directly from RocksDBStateBackend). This is just loading a factory class that is defined in Java. We could think about making the factory definable in Python but that is a separate issue.

sjwiesman · 2021-03-31T12:28:06Z

@dianfu thank you for the ver fast review. I've applied all your suggestions. Please take another look and let me know if you have any other concerns.

dianfu

@sjwiesman Thanks a lot for the update. LGTM overall with just one minor comment. Feel free to merge it after addressing this issue.

dianfu · 2021-03-31T15:24:49Z

flink-python/pyflink/datastream/stream_execution_environment.py

+        :param directory The savepoint directory
+        :return: This object.
+        """
+        return self._j_stream_execution_environment.setDefaultSavepointDirectory(directory)


should return self

sjwiesman · 2021-03-31T21:06:07Z

Thanks @dianfu. I've made the final fix. Since this is based #15429 I will wait on that PR to finish before merging.

sjwiesman · 2021-04-01T13:53:57Z

@dianfu there are some relevant test failures that seem to be related. I am having trouble debugging and was hoping you could take another look.

dianfu · 2021-04-02T05:33:46Z

@sjwiesman Sure. I'll take a look asap!

dianfu

@sjwiesman The test failures should be caused by the following issues.

dianfu · 2021-04-02T05:35:24Z

flink-python/pyflink/datastream/checkpoint_storage.py

+        raise TypeError("%s is not an instance of CheckpointStorage." % j_checkpoint_storage)
+
+    if get_java_class(JJobManagerCheckpointStorage).isAssignableFrom(j_clz):
+        return JobManagerCheckpointStorage(j_jobmanager_checkpoint_storage=j_clz)


Suggested change

return JobManagerCheckpointStorage(j_jobmanager_checkpoint_storage=j_clz)

return JobManagerCheckpointStorage(j_jobmanager_checkpoint_storage= j_checkpoint_storage)

dianfu · 2021-04-02T05:35:35Z

flink-python/pyflink/datastream/checkpoint_storage.py

+    if get_java_class(JJobManagerCheckpointStorage).isAssignableFrom(j_clz):
+        return JobManagerCheckpointStorage(j_jobmanager_checkpoint_storage=j_clz)
+    elif get_java_class(JFileSystemCheckpointStorage).isAssignableFrom(j_clz):
+        return FileSystemCheckpointStorage(j_filesystem_checkpoint_storage=j_clz)


Suggested change

return FileSystemCheckpointStorage(j_filesystem_checkpoint_storage=j_clz)

return FileSystemCheckpointStorage(j_filesystem_checkpoint_storage= j_checkpoint_storage)

dianfu · 2021-04-02T05:37:13Z

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

+
+    def test_create_jobmanager_checkpoint_storage(self):
+
+        self.assertItNotNone(JobManagerCheckpointStorage())


Suggested change

self.assertItNotNone(JobManagerCheckpointStorage())

self.assertIsNotNone(JobManagerCheckpointStorage())

dianfu · 2021-04-02T05:38:56Z

flink-python/pyflink/datastream/checkpoint_storage.py

+                checkpoint_path = JPath(checkpoint_path)
+            if max_state_size is None:
+                max_state_size = JJobManagerCheckpointStorage.DEFAULT_MAX_STATE_SIZE
+            j_jobmanager_checkpoint_storage = JobManagerCheckpointStorage(checkpoint_path,


Suggested change

j_jobmanager_checkpoint_storage = JobManagerCheckpointStorage(checkpoint_path,

j_jobmanager_checkpoint_storage = JJobManagerCheckpointStorage(checkpoint_path,

dianfu · 2021-04-02T05:43:44Z

flink-python/pyflink/datastream/checkpoint_storage.py

+                checkpoint_path = JPath(checkpoint_path)
+
+            if file_state_size_threshold is None:
+                file_state_size_threshold = FileSystemCheckpointStorage.MAX_FILE_STATE_THRESHOLD


It seems that we should set it as -1 if not specified.

dianfu · 2021-04-02T05:47:47Z

flink-python/pyflink/datastream/state_backend.py

+
+        :return: The number of threads used to transfer files while snapshotting/restoring.
+        """
+        return self._j_state_backend.getNumberOfTransferingThreads()


Suggested change

return self._j_state_backend.getNumberOfTransferingThreads()

return self._j_state_backend.getNumberOfTransferThreads()

dianfu · 2021-06-08T05:43:29Z

@sjwiesman What's the status of this PR?

sjwiesman · 2021-06-23T16:20:58Z

@dianfu I'm very sorry, this fell off my radar. I've just rebased the branch, if CI is good we can go ahead and merge.

sjwiesman requested review from HuangXingBo, dianfu and StephanEwen March 31, 2021 01:16

sjwiesman mentioned this pull request Mar 31, 2021

[FLINK-21935] Remove 'state.backend.async' option #15429

Closed

rmetzger added the review=description? label Mar 31, 2021

rmetzger added the component=API/Python label Mar 31, 2021

dianfu reviewed Mar 31, 2021

View reviewed changes

sjwiesman force-pushed the state-backend-py branch from 24f4019 to ca16802 Compare March 31, 2021 12:27

dianfu approved these changes Mar 31, 2021

View reviewed changes

sjwiesman force-pushed the state-backend-py branch from ca16802 to 5f2a1ad Compare March 31, 2021 21:04

sjwiesman force-pushed the state-backend-py branch from 5f2a1ad to 5f75c7d Compare April 1, 2021 13:53

dianfu reviewed Apr 2, 2021

View reviewed changes

sjwiesman force-pushed the state-backend-py branch 2 times, most recently from 8b19876 to a4daf4b Compare April 2, 2021 12:51

[FLINK-22052][python] Add new state backends to PyFlink

34d9926

sjwiesman force-pushed the state-backend-py branch from a4daf4b to 567ae87 Compare June 23, 2021 15:43

sjwiesman force-pushed the state-backend-py branch from 567ae87 to 3333be5 Compare June 23, 2021 18:51

sjwiesman added 2 commits June 23, 2021 16:40

[FLINK-22052][python] Add new checkpoint storage classes to PyFlink

7d13fd0

[FLINK-22052][python] Add set_default_savepoint_directory to PyFlink

43c1bd1

sjwiesman force-pushed the state-backend-py branch from 3333be5 to 43c1bd1 Compare June 23, 2021 21:40

sjwiesman closed this in 2d3779d Jun 24, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-22052][python] Add FLIP-142 public classes to python API #15441

[FLINK-22052][python] Add FLIP-142 public classes to python API #15441

sjwiesman commented Mar 31, 2021

flinkbot commented Mar 31, 2021 •

edited

Loading

flinkbot commented Mar 31, 2021 •

edited

Loading

dianfu left a comment

dianfu Mar 31, 2021

dianfu Mar 31, 2021

sjwiesman Mar 31, 2021

sjwiesman commented Mar 31, 2021

dianfu left a comment

dianfu Mar 31, 2021

sjwiesman commented Mar 31, 2021

sjwiesman commented Apr 1, 2021

dianfu commented Apr 2, 2021 •

edited

Loading

dianfu left a comment

dianfu Apr 2, 2021

dianfu Apr 2, 2021

dianfu Apr 2, 2021

dianfu Apr 2, 2021

dianfu Apr 2, 2021

dianfu Apr 2, 2021

dianfu commented Jun 8, 2021

sjwiesman commented Jun 23, 2021

	return JobManagerCheckpointStorage(j_jobmanager_checkpoint_storage=j_clz)
	return JobManagerCheckpointStorage(j_jobmanager_checkpoint_storage= j_checkpoint_storage)

	return FileSystemCheckpointStorage(j_filesystem_checkpoint_storage=j_clz)
	return FileSystemCheckpointStorage(j_filesystem_checkpoint_storage= j_checkpoint_storage)


		def test_create_jobmanager_checkpoint_storage(self):

		self.assertItNotNone(JobManagerCheckpointStorage())

	self.assertItNotNone(JobManagerCheckpointStorage())
	self.assertIsNotNone(JobManagerCheckpointStorage())

	j_jobmanager_checkpoint_storage = JobManagerCheckpointStorage(checkpoint_path,
	j_jobmanager_checkpoint_storage = JJobManagerCheckpointStorage(checkpoint_path,

	return self._j_state_backend.getNumberOfTransferingThreads()
	return self._j_state_backend.getNumberOfTransferThreads()

[FLINK-22052][python] Add FLIP-142 public classes to python API #15441

[FLINK-22052][python] Add FLIP-142 public classes to python API #15441

Conversation

sjwiesman commented Mar 31, 2021

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 Mar 31, 2021 • edited Loading

Automated Checks

Review Progress

flinkbot commented Mar 31, 2021 • edited Loading

CI report:

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

sjwiesman commented Mar 31, 2021

dianfu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sjwiesman commented Mar 31, 2021

sjwiesman commented Apr 1, 2021

dianfu commented Apr 2, 2021 • edited Loading

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

dianfu commented Jun 8, 2021

sjwiesman commented Jun 23, 2021

flinkbot commented Mar 31, 2021 •

edited

Loading

flinkbot commented Mar 31, 2021 •

edited

Loading

dianfu commented Apr 2, 2021 •

edited

Loading