| ################################################################################ |
| # 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. |
| ################################################################################ |
| import warnings |
| from pyflink.java_gateway import get_gateway |
| |
| from pyflink.common import Configuration |
| |
| __all__ = ['EnvironmentSettings'] |
| |
| |
| class EnvironmentSettings(object): |
| """ |
| Defines all parameters that initialize a table environment. Those parameters are used only |
| during instantiation of a :class:`~pyflink.table.TableEnvironment` and cannot be changed |
| afterwards. |
| |
| Example: |
| :: |
| |
| >>> EnvironmentSettings.new_instance() \\ |
| ... .in_streaming_mode() \\ |
| ... .with_built_in_catalog_name("my_catalog") \\ |
| ... .with_built_in_database_name("my_database") \\ |
| ... .build() |
| |
| :func:`EnvironmentSettings.in_streaming_mode` or :func:`EnvironmentSettings.in_batch_mode` |
| might be convenient as shortcuts. |
| """ |
| |
| class Builder(object): |
| """ |
| A builder for :class:`EnvironmentSettings`. |
| """ |
| |
| def __init__(self): |
| gateway = get_gateway() |
| self._j_builder = gateway.jvm.EnvironmentSettings.Builder() |
| |
| def use_old_planner(self) -> 'EnvironmentSettings.Builder': |
| """ |
| .. note:: The old planner has been removed in Flink 1.14. Since there is only one |
| planner left (previously called the 'blink' planner), this setting will |
| throw an exception. |
| """ |
| warnings.warn( |
| "Deprecated in 1.13. Please update to the new planner (i.e. Blink planner).", |
| DeprecationWarning) |
| self._j_builder = self._j_builder.useOldPlanner() |
| return self |
| |
| def use_blink_planner(self) -> 'EnvironmentSettings.Builder': |
| """ |
| Sets the Blink planner as the required module. |
| |
| This is the default behavior. |
| |
| .. note:: The old planner has been removed in Flink 1.14. Since there is only one |
| planner left (previously called the 'blink' planner), this setting is |
| obsolete and will be removed in future versions. |
| |
| :return: This object. |
| """ |
| warnings.warn( |
| "Deprecated in 1.14. A planner declaration is not required anymore.", |
| DeprecationWarning) |
| self._j_builder = self._j_builder.useBlinkPlanner() |
| return self |
| |
| def use_any_planner(self) -> 'EnvironmentSettings.Builder': |
| """ |
| Does not set a planner requirement explicitly. |
| |
| A planner will be discovered automatically, if there is only one planner available. |
| |
| By default, :func:`use_blink_planner` is enabled. |
| |
| .. note:: The old planner has been removed in Flink 1.14. Since there is only one |
| planner left (previously called the 'blink' planner), this setting is |
| obsolete and will be removed in future versions. |
| |
| :return: This object. |
| """ |
| warnings.warn( |
| "Deprecated in 1.14. A planner declaration is not required anymore.", |
| DeprecationWarning) |
| self._j_builder = self._j_builder.useAnyPlanner() |
| return self |
| |
| def in_batch_mode(self) -> 'EnvironmentSettings.Builder': |
| """ |
| Sets that the components should work in a batch mode. Streaming mode by default. |
| |
| :return: This object. |
| """ |
| self._j_builder = self._j_builder.inBatchMode() |
| return self |
| |
| def in_streaming_mode(self) -> 'EnvironmentSettings.Builder': |
| """ |
| Sets that the components should work in a streaming mode. Enabled by default. |
| |
| :return: This object. |
| """ |
| self._j_builder = self._j_builder.inStreamingMode() |
| return self |
| |
| def with_built_in_catalog_name(self, built_in_catalog_name: str) \ |
| -> 'EnvironmentSettings.Builder': |
| """ |
| Specifies the name of the initial catalog to be created when instantiating |
| a :class:`~pyflink.table.TableEnvironment`. |
| |
| This catalog is an in-memory catalog that will be used to store all temporary objects |
| (e.g. from :func:`~pyflink.table.TableEnvironment.create_temporary_view` or |
| :func:`~pyflink.table.TableEnvironment.create_temporary_system_function`) that cannot |
| be persisted because they have no serializable representation. |
| |
| It will also be the initial value for the current catalog which can be altered via |
| :func:`~pyflink.table.TableEnvironment.use_catalog`. |
| |
| Default: "default_catalog". |
| |
| :param built_in_catalog_name: The specified built-in catalog name. |
| :return: This object. |
| """ |
| self._j_builder = self._j_builder.withBuiltInCatalogName(built_in_catalog_name) |
| return self |
| |
| def with_built_in_database_name(self, built_in_database_name: str) \ |
| -> 'EnvironmentSettings.Builder': |
| """ |
| Specifies the name of the default database in the initial catalog to be |
| created when instantiating a :class:`~pyflink.table.TableEnvironment`. |
| |
| This database is an in-memory database that will be used to store all temporary |
| objects (e.g. from :func:`~pyflink.table.TableEnvironment.create_temporary_view` or |
| :func:`~pyflink.table.TableEnvironment.create_temporary_system_function`) that cannot |
| be persisted because they have no serializable representation. |
| |
| It will also be the initial value for the current catalog which can be altered via |
| :func:`~pyflink.table.TableEnvironment.use_catalog`. |
| |
| Default: "default_database". |
| |
| :param built_in_database_name: The specified built-in database name. |
| :return: This object. |
| """ |
| self._j_builder = self._j_builder.withBuiltInDatabaseName(built_in_database_name) |
| return self |
| |
| def build(self) -> 'EnvironmentSettings': |
| """ |
| Returns an immutable instance of EnvironmentSettings. |
| |
| :return: an immutable instance of EnvironmentSettings. |
| """ |
| return EnvironmentSettings(self._j_builder.build()) |
| |
| def __init__(self, j_environment_settings): |
| self._j_environment_settings = j_environment_settings |
| |
| def get_built_in_catalog_name(self) -> str: |
| """ |
| Gets the specified name of the initial catalog to be created when instantiating a |
| :class:`~pyflink.table.TableEnvironment`. |
| |
| :return: The specified name of the initial catalog to be created. |
| """ |
| return self._j_environment_settings.getBuiltInCatalogName() |
| |
| def get_built_in_database_name(self) -> str: |
| """ |
| Gets the specified name of the default database in the initial catalog to be created when |
| instantiating a :class:`~pyflink.table.TableEnvironment`. |
| |
| :return: The specified name of the default database in the initial catalog to be created. |
| """ |
| return self._j_environment_settings.getBuiltInDatabaseName() |
| |
| def is_blink_planner(self) -> bool: |
| """ |
| Tells if :class:`~pyflink.table.TableEnvironment` should work in a blink or old |
| planner. |
| |
| .. note:: The old planner has been removed in Flink 1.14. Since there is only one |
| planner left (previously called the 'blink' planner), this method is |
| obsolete and will be removed in future versions. |
| |
| :return: True if the TableEnvironment should work in a blink planner, false otherwise. |
| """ |
| warnings.warn( |
| "Deprecated in 1.14. There is only one planner anymore.", |
| DeprecationWarning) |
| return self._j_environment_settings.isBlinkPlanner() |
| |
| def is_streaming_mode(self) -> bool: |
| """ |
| Tells if the :class:`~pyflink.table.TableEnvironment` should work in a batch or streaming |
| mode. |
| |
| :return: True if the TableEnvironment should work in a streaming mode, false otherwise. |
| """ |
| return self._j_environment_settings.isStreamingMode() |
| |
| def to_configuration(self) -> Configuration: |
| """ |
| Convert to `pyflink.common.Configuration`. |
| |
| It sets the `table.planner` and `execution.runtime-mode` according to the current |
| EnvironmentSetting. |
| |
| :return: Configuration with specified value. |
| """ |
| return Configuration(j_configuration=self._j_environment_settings.toConfiguration()) |
| |
| @staticmethod |
| def new_instance() -> 'EnvironmentSettings.Builder': |
| """ |
| Creates a builder for creating an instance of EnvironmentSettings. |
| |
| By default, it does not specify a required planner and will use the one that is available |
| on the classpath via discovery. |
| |
| :return: A builder of EnvironmentSettings. |
| """ |
| return EnvironmentSettings.Builder() |
| |
| @staticmethod |
| def from_configuration(config: Configuration) -> 'EnvironmentSettings': |
| """ |
| Creates the EnvironmentSetting with specified Configuration. |
| |
| :return: EnvironmentSettings. |
| """ |
| return EnvironmentSettings( |
| get_gateway().jvm.EnvironmentSettings.fromConfiguration(config._j_configuration)) |
| |
| @staticmethod |
| def in_streaming_mode() -> 'EnvironmentSettings': |
| """ |
| Creates a default instance of EnvironmentSettings in streaming execution mode. |
| |
| In this mode, both bounded and unbounded data streams can be processed. |
| |
| This method is a shortcut for creating a :class:`~pyflink.table.TableEnvironment` with |
| little code. Use the builder provided in :func:`EnvironmentSettings.new_instance` for |
| advanced settings. |
| |
| :return: EnvironmentSettings. |
| """ |
| return EnvironmentSettings( |
| get_gateway().jvm.EnvironmentSettings.inStreamingMode()) |
| |
| @staticmethod |
| def in_batch_mode() -> 'EnvironmentSettings': |
| """ |
| Creates a default instance of EnvironmentSettings in batch execution mode. |
| |
| This mode is highly optimized for batch scenarios. Only bounded data streams can be |
| processed in this mode. |
| |
| This method is a shortcut for creating a :class:`~pyflink.table.TableEnvironment` with |
| little code. Use the builder provided in :func:`EnvironmentSettings.new_instance` for |
| advanced settings. |
| |
| :return: EnvironmentSettings. |
| """ |
| return EnvironmentSettings( |
| get_gateway().jvm.EnvironmentSettings.inBatchMode()) |
