You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fdolce (via GitHub)" <gi...@apache.org> on 2026/09/03 12:35:42 UTC

[PR] Flink dataframe configuration [flink]

fdolce opened a new pull request, #29088:
URL: https://github.com/apache/flink/pull/29088

   ## What is the purpose of the change
   
   This pull request adds configuration support to the PyFlink DataFrame API (FLIP-591, FLINK-40430). It introduces a unified configuration entry point, `pf.config`, a singleton `DataFrameConfig` object that accepts any Flink configuration key. Values can be set at any time, even before a `TableEnvironment` exists, because they are buffered and applied automatically once the underlying environment is lazily created or injected via `set_table_environment`. This means users neither manage environment-creation order nor choose between environment objects (`StreamExecutionEnvironment` vs. `TableConfig`) for the common case.
   
   ```python
   import pyflink.dataframe as pf
   
   pf.config.set("parallelism.default", "4")
   pf.config.set("execution.runtime-mode", "batch")
   ```
   
   Semantics, where FLIP-591 leaves them open:
   
     - The lazily created environment is built *from* the buffered values (they are passed as the `Configuration` of the `StreamExecutionEnvironment`), so options that can only be chosen at creation time (`execution.runtime-mode`) work. Applying them to the `TableConfig` after creation would make the planner reject the first `execute()` with "Mismatch between configured runtime mode and actual runtime mode".
     - An injected environment receives the buffered values only for keys it does not already set explicitly in its own `TableConfig`, so an environment the user configured themselves is not silently overridden.
     - Buffered values persist across environments: they are re-applied to every environment created or injected later, not consumed on first use.
     - Applying the buffer happens before the environment is stored as the active one. If it fails, the previously active environment is kept and a retry re-applies the buffer, rather than retaining a half-configured environment.
     - A value the active environment rejects (e.g. an invalid `pipeline.jars` URL) is not buffered, so one bad `set` cannot make every later `set_table_environment` fail.
   
   ## Brief change log
   
     - Added `pyflink/dataframe/_config.py` with the `DataFrameConfig` class (chainable `set(key, value)` and `get(key, default)`) and the module-level `config` singleton. The module is private (`_config`) because a public module named `config` would be shadowed by the `pf.config` attribute (`import pyflink.dataframe.config as m` would yield the singleton instead of the module).
     - `set` writes the value through to the active environment's `TableConfig` when one exists and buffers it only on success; `get` reads from the active environment (falling back to the root configuration via `TableConfig.get`) or from the buffer when no environment exists
     - `get_or_create_table_environment` creates the `StreamExecutionEnvironment` from a `Configuration` built from the buffered values, applies the buffer to the resulting `TableConfig`, and only then stores the environment
     - `set_table_environment` applies the buffered values that the injected environment does not set explicitly, and only then stores the environment
     - Exported `DataFrameConfig` and `config` from `pyflink.dataframe`
     - Added a Configuration page to the PyFlink DataFrame API reference docs, listing both `config` and `DataFrameConfig`
   
   ## Verifying this change
   
   This change added tests and can be verified as follows:
   
     - Added `pyflink/dataframe/tests/test_config.py` covering: type validation of `set`/`get` arguments, chaining, buffering before an environment exists, returning defaults for unset keys, applying buffered values on environment injection and on lazy environment creation, writing through to the active environment, and reading values set directly on the active environment
     - Added tests for the semantics above: `execution.runtime-mode` buffered before creation is honored by executing a query on the created environment; a failure while applying the buffer leaves `get_table_environment()` unchanged for both `set_table_environment` and `get_or_create_table_environment`, and a subsequent `get_or_create_table_environment` re-applies the buffer; a value rejected by the active environment is not buffered and not replayed onto a later environment; an injected environment keeps a value it set explicitly; no `pyflink.dataframe` submodule is shadowed by a package attribute
     - Existing tests in `pyflink/dataframe/tests/` all pass
   
   ## 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)`: yes (new `PublicEvolving` Python API `DataFrameConfig` / `pf.config`)
     - 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, ZooKeeper: no
     - The S3 file system connector: no
   
   ## Documentation
   
     - Does this pull request introduce a new feature? yes
     - If yes, how is the feature documented? docs (new Configuration page in the PyFlink DataFrame API reference, `flink-python/docs/reference/pyflink.dataframe/config.rst`) and Python docstrings
   
   ---
   
   ##### Was generative AI tooling used to co-author this PR?
   
   - [X] Yes (please specify the tool below)
   
   Generated-by: Claude Code


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40430][python] Add configuration support to DataFrame API [flink]

Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29088:
URL: https://github.com/apache/flink/pull/29088#discussion_r3954623215


##########
flink-python/pyflink/dataframe/context.py:
##########
@@ -95,9 +105,14 @@ def get_or_create_table_environment() -> TableEnvironment:
     global _global_table_environment
 
     if _global_table_environment is None:
+        from pyflink.dataframe._config import config
         from pyflink.datastream import StreamExecutionEnvironment
 
-        stream_environment = StreamExecutionEnvironment.get_execution_environment()
-        _global_table_environment = StreamTableEnvironment.create(stream_environment)
+        stream_environment = StreamExecutionEnvironment.get_execution_environment(
+            config._to_configuration()
+        )
+        t_env = StreamTableEnvironment.create(stream_environment)

Review Comment:
   Could we pass the buffered configuration to EnvironmentSettings as well and use the two-argument StreamTableEnvironment.create() as following?
   ```
   configuration = config._to_configuration()
   t_env = StreamExecutionEnvironment.get_execution_environment(configuration)
   
   settings = (
       EnvironmentSettings.new_instance()
       .with_configuration(configuration)
       .build()
   )
   
   t_env = StreamTableEnvironment.create(
       t_env,
       environment_settings=settings,
   )
   ```
   
   The current overload only derives the runtime mode from the StreamExecutionEnvironment. Other TableEnvironment creation-time options are applied too late by _apply_to(). For example:
   
   ```
       pf.config.set("table.builtin-catalog-name", "my_catalog")
       t_env = pf.get_or_create_table_environment()
   
       assert pf.config.get("table.builtin-catalog-name") == "my_catalog"
       assert t_env.get_current_catalog() == "default_catalog"  # not applied
   ```
   
   PS: in this case, `config._apply_to(t_env, overwrite=True)` is not required any more?



##########
flink-python/pyflink/dataframe/context.py:
##########
@@ -50,6 +54,10 @@ def set_table_environment(t_env: Optional[TableEnvironment]) -> None:
     global _global_table_environment
     if t_env is not None and not isinstance(t_env, TableEnvironment):
         raise TypeError("t_env must be a TableEnvironment or None")
+    if t_env is not None:

Review Comment:
   Some Flink options are consumed while the StreamExecutionEnvironment or TableEnvironment is being created. Applying buffered values later through `t_env.get_config().set(...)` only updates TableConfig and cannot reconfigure the existing environment.
   
   Do you think it makes sense to separate the two initialization paths:
   
   1. `pf.config` configures only environments created by `get_or_create_table_environment()`.
   2. An environment passed to `set_table_environment()` is already constructed and should be treated as authoritative. So `set_table_environment()` should not call `config._apply_to()`. I guess we could check if config is empty to ensure users use it correctly.



##########
flink-python/pyflink/dataframe/_config.py:
##########
@@ -0,0 +1,151 @@
+################################################################################
+#  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.
+################################################################################
+
+from typing import Dict, Optional
+
+from pyflink.common import Configuration
+from pyflink.table import TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+    "DataFrameConfig",

Review Comment:
   `DataFrameConfig` is exported as a `@PublicEvolving` class, so users can reasonably instantiate it. However, independently created instances are not consumed by `get_or_create_table_environment()`, which always reads the module-level `pf.config` singleton.
   
   For example:
   
       config = pf.DataFrameConfig()
       config.set("execution.runtime-mode", "batch")
       t_env = pf.get_or_create_table_environment()
   
   The call to `set()` succeeds, but the environment is created from `pf.config`, so the value stored in `config` is silently ignored.
   
   Since this API is designed around a single global configuration, so I think we need make the implementation class private and export only the `pf.config` singleton. 



##########
flink-python/pyflink/dataframe/_config.py:
##########
@@ -0,0 +1,151 @@
+################################################################################
+#  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.
+################################################################################
+
+from typing import Dict, Optional
+
+from pyflink.common import Configuration
+from pyflink.table import TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+    "DataFrameConfig",
+    "config",
+]
+
+
+@PublicEvolving()
+class DataFrameConfig:
+    """
+    A unified entry point for Flink configuration in the DataFrame API.
+
+    Accepts any Flink configuration key and buffers the value, so configuration can be set
+    at any time -- even before an environment exists. Buffered values are used when
+    :func:`get_or_create_table_environment` creates the underlying
+    :class:`~pyflink.table.TableEnvironment`, so options that can only be chosen at creation
+    time, such as ``execution.runtime-mode``, take effect. An environment injected via
+    :func:`set_table_environment` receives the buffered values for every key it does not
+    already set explicitly. While an environment is active, values are also written through
+    to its configuration immediately.
+
+    Use the module-level singleton :data:`config` instead of instantiating this class.
+
+    Example::
+
+        >>> import pyflink.dataframe as pf
+        >>> _ = pf.config.set("parallelism.default", "4")
+        >>> pf.config.get("parallelism.default")
+        '4'
+
+    .. versionadded:: 2.4.0
+    """
+
+    def __init__(self: "DataFrameConfig"):
+        self._buffered: Dict[str, str] = {}
+
+    def set(self, key: str, value: str) -> "DataFrameConfig":
+        """
+        Sets a string-based value for the given string-based key.
+
+        The value is buffered and applied to the underlying environment once it is created
+        or injected; when an environment is already active, the value is applied to its
+        configuration immediately as well. A value the active environment rejects is not
+        buffered.
+
+        :param key: The configuration key.
+        :param value: The configuration value. It will be parsed by the framework on access.
+        :return: This object, to allow chaining of calls.
+        :raises TypeError: If ``key`` or ``value`` is not a string.
+
+        Example::
+
+            >>> import pyflink.dataframe as pf
+            >>> _ = pf.config.set("parallelism.default", "4") \\
+            ...              .set("execution.runtime-mode", "batch")
+
+        .. versionadded:: 2.4.0
+        """
+        if not isinstance(key, str):
+            raise TypeError("key must be a string")
+        if not isinstance(value, str):
+            raise TypeError("value must be a string")
+
+        from pyflink.dataframe.context import get_table_environment
+
+        t_env = get_table_environment()
+        if t_env is not None:

Review Comment:
   Some Flink options are consumed when the StreamExecutionEnvironment or TableEnvironment is created. For those options, forwarding a later update to
   
       _global_table_environment.get_config().set(key, value)
   
   only changes the TableConfig value; it does not reconfigure the already-created environment. For example, changing `execution.runtime-mode` here cannot change how the existing environment was initialized.
   
   Do you think it make sense to let DataFrameConfig.set() fail fast whenever a global TableEnvironment already exists, and require all DataFrame configuration to be set before the environment is created?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40430][python] Add configuration support to DataFrame API [flink]

Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on PR #29088:
URL: https://github.com/apache/flink/pull/29088#issuecomment-5582100525

   Thanks for the review @dianfu , looking into it


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Flink dataframe configuration [flink]

Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29088:
URL: https://github.com/apache/flink/pull/29088#issuecomment-5525973364

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "9a8d8d703d2803e0dc73d9327c57d1d704185059",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "9a8d8d703d2803e0dc73d9327c57d1d704185059",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 9a8d8d703d2803e0dc73d9327c57d1d704185059 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40430][python] Add configuration support to DataFrame API [flink]

Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu closed pull request #29088: [FLINK-40430][python] Add configuration support to DataFrame API
URL: https://github.com/apache/flink/pull/29088


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40430][python] Add configuration support to DataFrame API [flink]

Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on code in PR #29088:
URL: https://github.com/apache/flink/pull/29088#discussion_r3958521326


##########
flink-python/pyflink/dataframe/context.py:
##########
@@ -95,9 +105,14 @@ def get_or_create_table_environment() -> TableEnvironment:
     global _global_table_environment
 
     if _global_table_environment is None:
+        from pyflink.dataframe._config import config
         from pyflink.datastream import StreamExecutionEnvironment
 
-        stream_environment = StreamExecutionEnvironment.get_execution_environment()
-        _global_table_environment = StreamTableEnvironment.create(stream_environment)
+        stream_environment = StreamExecutionEnvironment.get_execution_environment(
+            config._to_configuration()
+        )
+        t_env = StreamTableEnvironment.create(stream_environment)

Review Comment:
   Oh, nice, this looks way better. Still getting learning the pyflink api, so sometimes I miss things. Fixing this, and yes, we shouldn't need the apply_to at this point



##########
flink-python/pyflink/dataframe/_config.py:
##########
@@ -0,0 +1,151 @@
+################################################################################
+#  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.
+################################################################################
+
+from typing import Dict, Optional
+
+from pyflink.common import Configuration
+from pyflink.table import TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+    "DataFrameConfig",
+    "config",
+]
+
+
+@PublicEvolving()
+class DataFrameConfig:
+    """
+    A unified entry point for Flink configuration in the DataFrame API.
+
+    Accepts any Flink configuration key and buffers the value, so configuration can be set
+    at any time -- even before an environment exists. Buffered values are used when
+    :func:`get_or_create_table_environment` creates the underlying
+    :class:`~pyflink.table.TableEnvironment`, so options that can only be chosen at creation
+    time, such as ``execution.runtime-mode``, take effect. An environment injected via
+    :func:`set_table_environment` receives the buffered values for every key it does not
+    already set explicitly. While an environment is active, values are also written through
+    to its configuration immediately.
+
+    Use the module-level singleton :data:`config` instead of instantiating this class.
+
+    Example::
+
+        >>> import pyflink.dataframe as pf
+        >>> _ = pf.config.set("parallelism.default", "4")
+        >>> pf.config.get("parallelism.default")
+        '4'
+
+    .. versionadded:: 2.4.0
+    """
+
+    def __init__(self: "DataFrameConfig"):
+        self._buffered: Dict[str, str] = {}
+
+    def set(self, key: str, value: str) -> "DataFrameConfig":
+        """
+        Sets a string-based value for the given string-based key.
+
+        The value is buffered and applied to the underlying environment once it is created
+        or injected; when an environment is already active, the value is applied to its
+        configuration immediately as well. A value the active environment rejects is not
+        buffered.
+
+        :param key: The configuration key.
+        :param value: The configuration value. It will be parsed by the framework on access.
+        :return: This object, to allow chaining of calls.
+        :raises TypeError: If ``key`` or ``value`` is not a string.
+
+        Example::
+
+            >>> import pyflink.dataframe as pf
+            >>> _ = pf.config.set("parallelism.default", "4") \\
+            ...              .set("execution.runtime-mode", "batch")
+
+        .. versionadded:: 2.4.0
+        """
+        if not isinstance(key, str):
+            raise TypeError("key must be a string")
+        if not isinstance(value, str):
+            raise TypeError("value must be a string")
+
+        from pyflink.dataframe.context import get_table_environment
+
+        t_env = get_table_environment()
+        if t_env is not None:

Review Comment:
   Yes, this was one of my main doubts about the design. Making the DataFrameConfig incompatible with a global environment and failing fast sounds a lot better to me, and also requiring the config to be set before the environment is created makes things easier, so yep, I'm going to change this.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40430][python] Add configuration support to DataFrame API [flink]

Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on code in PR #29088:
URL: https://github.com/apache/flink/pull/29088#discussion_r3958538085


##########
flink-python/pyflink/dataframe/_config.py:
##########
@@ -0,0 +1,151 @@
+################################################################################
+#  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.
+################################################################################
+
+from typing import Dict, Optional
+
+from pyflink.common import Configuration
+from pyflink.table import TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+    "DataFrameConfig",

Review Comment:
   Mh, right. I'll make only `get` and `set` part of the public api, hide the class itself. Since we are here, I'll also rename the _config.py module to dataframe_config.py since I saw table api has a table_config.py



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40430][python] Add configuration support to DataFrame API [flink]

Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on code in PR #29088:
URL: https://github.com/apache/flink/pull/29088#discussion_r3958526422


##########
flink-python/pyflink/dataframe/context.py:
##########
@@ -50,6 +54,10 @@ def set_table_environment(t_env: Optional[TableEnvironment]) -> None:
     global _global_table_environment
     if t_env is not None and not isinstance(t_env, TableEnvironment):
         raise TypeError("t_env must be a TableEnvironment or None")
+    if t_env is not None:

Review Comment:
   Yep, agree on this, fixing it



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org