This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e63885d9db3 [FLINK-40430][python] Add configuration support to
DataFrame API
e63885d9db3 is described below
commit e63885d9db398684d4a935a395ed3c59246e1c94
Author: Federico Dolce <[email protected]>
AuthorDate: Tue Sep 1 16:01:32 2026 +0200
[FLINK-40430][python] Add configuration support to DataFrame API
This closes #29088.
---
.../docs/reference/pyflink.dataframe/config.rst | 51 ++++++
.../docs/reference/pyflink.dataframe/index.rst | 1 +
flink-python/pyflink/dataframe/__init__.py | 2 +
flink-python/pyflink/dataframe/context.py | 66 +++++++-
flink-python/pyflink/dataframe/dataframe_config.py | 148 ++++++++++++++++
.../pyflink/dataframe/tests/test_config.py | 187 +++++++++++++++++++++
6 files changed, 449 insertions(+), 6 deletions(-)
diff --git a/flink-python/docs/reference/pyflink.dataframe/config.rst
b/flink-python/docs/reference/pyflink.dataframe/config.rst
new file mode 100644
index 00000000000..36730b1b597
--- /dev/null
+++ b/flink-python/docs/reference/pyflink.dataframe/config.rst
@@ -0,0 +1,51 @@
+..
################################################################################
+ 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.
+
################################################################################
+
+=============
+Configuration
+=============
+
+A unified entry point for Flink configuration. The module-level singleton
``pf.config``
+accepts any Flink configuration key and buffers the value until
+``pf.get_or_create_table_environment()`` creates the underlying
TableEnvironment. Because
+the values are supplied at creation time, options that can only be chosen
then, such as
+``execution.runtime-mode``, take effect.
+
+Configuration must be set before the environment exists; ``pf.config.set()``
raises once an
+environment is active. An environment injected via
``pf.set_table_environment()`` is treated
+as fully configured and does not receive buffered values.
+
+Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> _ = pf.config.set("parallelism.default", "4") \
+ ... .set("execution.runtime-mode", "batch")
+ >>> pf.config.get("parallelism.default")
+ '4'
+
+config
+------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ config
+ config.set
+ config.get
diff --git a/flink-python/docs/reference/pyflink.dataframe/index.rst
b/flink-python/docs/reference/pyflink.dataframe/index.rst
index 197cfccdaca..547cb0f2202 100644
--- a/flink-python/docs/reference/pyflink.dataframe/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/index.rst
@@ -32,3 +32,4 @@ This page gives an overview of all public PyFlink DataFrame
APIs.
sql
datatype
environment
+ config
diff --git a/flink-python/pyflink/dataframe/__init__.py
b/flink-python/pyflink/dataframe/__init__.py
index 50496a3fa50..c8ed1cf08ae 100644
--- a/flink-python/pyflink/dataframe/__init__.py
+++ b/flink-python/pyflink/dataframe/__init__.py
@@ -52,6 +52,7 @@ from pyflink.dataframe.context import (
set_table_environment,
)
from pyflink.dataframe.dataframe import DataFrame, GroupedDataFrame, col, lit
+from pyflink.dataframe.dataframe_config import config
from pyflink.dataframe.datatype import DataType
from pyflink.dataframe.io import read_generic
from pyflink.dataframe.sql import sql
@@ -72,6 +73,7 @@ __all__ = [
"range",
"read_generic",
"sql",
+ "config",
"set_table_environment",
"get_table_environment",
"get_or_create_table_environment",
diff --git a/flink-python/pyflink/dataframe/context.py
b/flink-python/pyflink/dataframe/context.py
index 031c2995108..3abdcdaf403 100644
--- a/flink-python/pyflink/dataframe/context.py
+++ b/flink-python/pyflink/dataframe/context.py
@@ -18,8 +18,11 @@
from typing import Optional
-from pyflink.table import StreamTableEnvironment, TableEnvironment
+from pyflink.common import Configuration
+from pyflink.java_gateway import get_gateway
+from pyflink.table import EnvironmentSettings, StreamTableEnvironment,
TableEnvironment
from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import get_j_env_configuration
__all__ = [
"set_table_environment",
@@ -35,8 +38,15 @@ def set_table_environment(t_env: Optional[TableEnvironment])
-> None:
"""
Set the environment used by DataFrame operations.
+ The injected environment is treated as fully configured: values buffered in
+ :data:`config` are not applied to it. Injecting an environment while
:data:`config`
+ holds buffered values is rejected, because those values would otherwise be
silently
+ ignored. Passing ``None`` clears the environment and leaves the buffered
values intact.
+
:param t_env: Environment to use, or ``None`` to clear it.
:raises TypeError: If ``t_env`` is neither a :class:`TableEnvironment` nor
``None``.
+ :raises RuntimeError: If ``t_env`` is not ``None`` and :data:`config`
holds buffered
+ values.
Example::
@@ -50,6 +60,17 @@ 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:
+ from pyflink.dataframe.dataframe_config import config
+
+ if config._buffered:
+ raise RuntimeError(
+ "pf.config holds buffered values that only apply to an
environment created by "
+ "get_or_create_table_environment(); they would be ignored by
the injected "
+ "environment. Configure that environment through
t_env.get_config() instead of "
+ "pf.config, or call get_or_create_table_environment() to have
the buffered "
+ "values applied."
+ )
_global_table_environment = t_env
@@ -77,7 +98,9 @@ def get_or_create_table_environment() -> TableEnvironment:
"""
Return the configured environment, creating one when necessary.
- The created environment is retained for subsequent DataFrame operations
and calls to
+ The environment is created from the values buffered in :data:`config`, so
options that
+ can only be chosen at creation time, such as ``execution.runtime-mode``,
take effect.
+ It is retained for subsequent DataFrame operations and calls to
:func:`get_table_environment`.
:return: The configured or newly created environment.
@@ -95,9 +118,40 @@ def get_or_create_table_environment() -> TableEnvironment:
global _global_table_environment
if _global_table_environment is None:
- from pyflink.datastream import StreamExecutionEnvironment
-
- stream_environment =
StreamExecutionEnvironment.get_execution_environment()
- _global_table_environment =
StreamTableEnvironment.create(stream_environment)
+ from pyflink.dataframe.dataframe_config import config
+ from pyflink.datastream import RuntimeExecutionMode,
StreamExecutionEnvironment
+
+ configuration = config._to_configuration()
+ stream_environment =
StreamExecutionEnvironment.get_execution_environment(configuration)
+
+ # The execution environment may merge deployment configuration (for
example, from the
+ # CLI or config.yaml) while it is being created. EnvironmentSettings
must use this
+ # effective configuration because some Table API options are consumed
during creation.
+ j_environment_configuration = get_j_env_configuration(
+ stream_environment._j_stream_execution_environment
+ )
+ environment_configuration = Configuration(
+ j_configuration=j_environment_configuration
+ )
+ settings_builder =
EnvironmentSettings.new_instance().with_configuration(
+ environment_configuration
+ )
+
+ # Match StreamTableEnvironment.create(executionEnvironment): Table API
supports only an
+ # explicit batch or streaming mode, so AUTOMATIC is treated as
streaming.
+ j_execution_options = (
+ get_gateway().jvm.org.apache.flink.configuration.ExecutionOptions
+ )
+ runtime_mode = RuntimeExecutionMode._from_j_execution_mode(
+ j_environment_configuration.get(j_execution_options.RUNTIME_MODE)
+ )
+ if runtime_mode == RuntimeExecutionMode.BATCH:
+ settings_builder.in_batch_mode()
+ else:
+ settings_builder.in_streaming_mode()
+ settings = settings_builder.build()
+ _global_table_environment = StreamTableEnvironment.create(
+ stream_environment, environment_settings=settings
+ )
return _global_table_environment
diff --git a/flink-python/pyflink/dataframe/dataframe_config.py
b/flink-python/pyflink/dataframe/dataframe_config.py
new file mode 100644
index 00000000000..f431c904b63
--- /dev/null
+++ b/flink-python/pyflink/dataframe/dataframe_config.py
@@ -0,0 +1,148 @@
+################################################################################
+# 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.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+ "config",
+]
+
+
+class _DataFrameConfig:
+ """
+ A unified entry point for Flink configuration in the DataFrame API.
+
+ Accepts any Flink configuration key and buffers the value until
+ :func:`get_or_create_table_environment` creates the underlying
+ :class:`~pyflink.table.TableEnvironment`. Because the values are supplied
at creation
+ time, options that can only be chosen then, such as
``execution.runtime-mode`` or
+ ``table.builtin-catalog-name``, take effect.
+
+ Configuration must therefore be set before the environment exists. Once an
environment
+ is active, whether created or injected via :func:`set_table_environment`,
use its own
+ :meth:`~pyflink.table.TableEnvironment.get_config` instead. An environment
passed to
+ :func:`set_table_environment` is treated as fully configured and does not
receive
+ buffered values. Buffered values survive clearing the environment with
+ ``set_table_environment(None)`` and feed the next environment created.
+
+ 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] = {}
+
+ @PublicEvolving()
+ def set(self, key: str, value: str) -> "_DataFrameConfig":
+ """
+ Sets a string-based value for the given string-based key.
+
+ The value is buffered and supplied to the environment created by
+ :func:`get_or_create_table_environment`. It cannot be called while an
environment
+ is active, because options consumed at creation time could no longer
take effect;
+ configure the active environment through its
+ :meth:`~pyflink.table.TableEnvironment.get_config` instead.
+
+ :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.
+ :raises RuntimeError: If an environment is already active.
+
+ 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
+
+ if get_table_environment() is not None:
+ raise RuntimeError(
+ "DataFrame configuration must be set before the table
environment exists. "
+ "Configure the active environment through t_env.get_config(),
or clear it "
+ "with set_table_environment(None) before calling config.set()."
+ )
+ self._buffered[key] = value
+ return self
+
+ @PublicEvolving()
+ def get(self, key: str, default: Optional[str] = None) -> Optional[str]:
+ """
+ Returns the value associated with the given key as a string.
+
+ When an environment is active, the value is read from its
configuration, so values
+ set outside this object are visible as well; otherwise the value is
read from the
+ buffered values.
+
+ :param key: The configuration key.
+ :param default: The value returned when there is no value associated
with ``key``.
+ :return: The (default) value associated with ``key``.
+ :raises TypeError: If ``key`` is not a string, or ``default`` is
neither a string
+ nor ``None``.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> _ = pf.config.set("parallelism.default", "4")
+ >>> pf.config.get("parallelism.default")
+ '4'
+ >>> pf.config.get("pipeline.name", "unnamed")
+ 'unnamed'
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(key, str):
+ raise TypeError("key must be a string")
+ if default is not None and not isinstance(default, str):
+ raise TypeError("default must be a string or None")
+
+ from pyflink.dataframe.context import get_table_environment
+
+ t_env = get_table_environment()
+ if t_env is not None:
+ return t_env.get_config().get(key, default)
+ return self._buffered.get(key, default)
+
+ def _to_configuration(self) -> Configuration:
+ configuration = Configuration()
+ for key, value in self._buffered.items():
+ configuration.set_string(key, value)
+ return configuration
+
+
+config = _DataFrameConfig()
+"""The singleton :class:`_DataFrameConfig` used by the DataFrame API."""
diff --git a/flink-python/pyflink/dataframe/tests/test_config.py
b/flink-python/pyflink/dataframe/tests/test_config.py
new file mode 100644
index 00000000000..d242bd9b2ee
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_config.py
@@ -0,0 +1,187 @@
+################################################################################
+# 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 unittest
+from typing import get_type_hints, Optional
+
+import pyflink.dataframe as pf
+from pyflink.dataframe.dataframe_config import _DataFrameConfig
+from pyflink.testing.test_case_utils import PyFlinkUTTestCase
+
+
+class DataFrameConfigValidationTests(unittest.TestCase):
+ def setUp(self):
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+ self.addCleanup(pf.config._buffered.clear)
+ pf.set_table_environment(None)
+ pf.config._buffered.clear()
+
+ def test_config_is_a_dataframe_config_singleton(self):
+ self.assertIsInstance(pf.config, _DataFrameConfig)
+ self.assertNotIn("DataFrameConfig", pf.__all__)
+ self.assertNotIn("_DataFrameConfig", pf.__all__)
+
+ def test_public_type_hints_are_resolvable(self):
+ self.assertEqual(
+ get_type_hints(_DataFrameConfig.set),
+ {
+ "key": str,
+ "value": str,
+ "return": _DataFrameConfig,
+ },
+ )
+ self.assertEqual(
+ get_type_hints(_DataFrameConfig.get),
+ {
+ "key": str,
+ "default": Optional[str],
+ "return": Optional[str],
+ },
+ )
+
+ def test_set_rejects_non_string_key_without_buffering(self):
+ with self.assertRaisesRegex(TypeError, "key must be a string"):
+ pf.config.set(1, "value")
+
+ self.assertEqual(pf.config._buffered, {})
+
+ def test_set_rejects_non_string_value_without_buffering(self):
+ with self.assertRaisesRegex(TypeError, "value must be a string"):
+ pf.config.set("pipeline.name", 1)
+
+ self.assertEqual(pf.config._buffered, {})
+
+ def test_get_rejects_non_string_key(self):
+ with self.assertRaisesRegex(TypeError, "key must be a string"):
+ pf.config.get(1)
+
+ def test_get_rejects_non_string_default(self):
+ with self.assertRaisesRegex(TypeError, "default must be a string or
None"):
+ pf.config.get("pipeline.name", 1)
+
+ def test_set_returns_the_config_for_chaining(self):
+ result = pf.config.set("pipeline.name",
"a").set("parallelism.default", "4")
+
+ self.assertIs(result, pf.config)
+
+ def test_buffered_value_is_returned_before_an_environment_exists(self):
+ pf.config.set("pipeline.name", "buffered")
+
+ self.assertEqual(pf.config.get("pipeline.name"), "buffered")
+
+ def test_default_is_returned_when_not_buffered(self):
+ self.assertIsNone(pf.config.get("pipeline.name"))
+ self.assertEqual(pf.config.get("pipeline.name", "fallback"),
"fallback")
+
+
+class DataFrameConfigTests(PyFlinkUTTestCase):
+ def setUp(self):
+ super().setUp()
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+ self.addCleanup(pf.config._buffered.clear)
+ pf.set_table_environment(None)
+ pf.config._buffered.clear()
+
+ def
test_buffered_values_are_applied_to_the_lazily_created_environment(self):
+ pf.config.set("pipeline.name", "lazy-name")
+
+ created_environment = pf.get_or_create_table_environment()
+
+ self.assertEqual(
+ created_environment.get_config().get("pipeline.name", None),
"lazy-name"
+ )
+
+ def test_table_creation_time_option_takes_effect(self):
+ # The built-in catalog is chosen when the TableEnvironment is
instantiated, so the
+ # buffered value has to reach EnvironmentSettings rather than
TableConfig afterwards.
+ pf.config.set("table.builtin-catalog-name", "my_catalog")
+
+ created_environment = pf.get_or_create_table_environment()
+
+ self.assertEqual(created_environment.get_current_catalog(),
"my_catalog")
+
+ def test_set_is_rejected_while_an_environment_is_active(self):
+ pf.set_table_environment(self.t_env)
+
+ with self.assertRaisesRegex(RuntimeError, "before the table
environment exists"):
+ pf.config.set("pipeline.name", "too-late")
+
+ self.assertEqual(pf.config._buffered, {})
+ self.assertIsNone(self.t_env.get_config().get("pipeline.name", None))
+
+ def test_set_is_allowed_again_after_the_environment_is_cleared(self):
+ pf.set_table_environment(self.t_env)
+ pf.set_table_environment(None)
+
+ pf.config.set("pipeline.name", "after-clear")
+
+ self.assertEqual(pf.config.get("pipeline.name"), "after-clear")
+
+ def
test_injecting_an_environment_is_rejected_when_values_are_buffered(self):
+ pf.config.set("pipeline.name", "buffered-name")
+
+ with self.assertRaisesRegex(RuntimeError, "buffered values"):
+ pf.set_table_environment(self.t_env)
+
+ self.assertIsNone(pf.get_table_environment())
+ self.assertIsNone(self.t_env.get_config().get("pipeline.name", None))
+
+ def test_injected_environment_is_not_modified(self):
+ self.t_env.get_config().set("pipeline.name", "explicit")
+
+ pf.set_table_environment(self.t_env)
+
+ self.assertIs(pf.get_table_environment(), self.t_env)
+ self.assertEqual(self.t_env.get_config().get("pipeline.name", None),
"explicit")
+
+ def test_clearing_the_environment_keeps_buffered_values(self):
+ pf.config.set("pipeline.name", "kept")
+ pf.get_or_create_table_environment()
+
+ pf.set_table_environment(None)
+
+ self.assertEqual(pf.config.get("pipeline.name"), "kept")
+ created_environment = pf.get_or_create_table_environment()
+ self.assertEqual(created_environment.get_config().get("pipeline.name",
None), "kept")
+
+ def test_get_reads_from_the_active_environment(self):
+ pf.set_table_environment(self.t_env)
+ self.t_env.get_config().set("pipeline.name", "from-environment")
+
+ self.assertEqual(pf.config.get("pipeline.name"), "from-environment")
+
+ def
test_get_returns_default_when_missing_from_the_active_environment(self):
+ pf.set_table_environment(self.t_env)
+
+ self.assertEqual(pf.config.get("pipeline.name", "fallback"),
"fallback")
+
+ def test_runtime_mode_buffered_before_creation_takes_effect(self):
+ # The planner is chosen when the environment is instantiated, so a
buffered
+ # runtime mode must be visible at creation time rather than applied
afterwards.
+ pf.config.set("execution.runtime-mode", "batch")
+
+ created_environment = pf.get_or_create_table_environment()
+ table = created_environment.from_elements([(1, "a"), (2, "b")], ["id",
"name"])
+ with table.execute().collect() as rows:
+ self.assertEqual(len(list(rows)), 2)
+
+
+if __name__ == "__main__":
+ unittest.main()
