auroflow opened a new pull request, #29048:
URL: https://github.com/apache/flink/pull/29048
## What is the purpose of the change
This pull request adds per-UDF concurrency and batch-size settings to the
PyFlink
DataFrame API.
Users can configure these settings when declaring a UDF:
```python
@udf(concurrency=4)
def add_one(value: int) -> int:
return value + 1
@udf(
return_dtype=DataType.int64(),
func_type="pandas",
batch_size=256,
)
def pandas_add_one(values):
return values + 1
```
The planner applies an explicitly configured concurrency as the Python
operator
parallelism. UDFs with incompatible explicit concurrency values are placed in
separate operators.
When compatible pandas UDFs are fused into one operator, the smallest
explicitly
configured batch size is used. If no UDF specifies a batch size, the existing
global Arrow batch-size configuration remains effective.
This change builds on the PyFlink DataFrame UDF support currently under
review
and is intended to be rebased onto `master` after that change is merged.
## Brief change log
- Added optional `concurrency` and `batch_size` arguments to DataFrame and
Table
scalar UDF declarations.
- Added validation for positive integer values and restricted `batch_size` to
pandas UDFs.
- Propagated the settings through Python function metadata while retaining
backward-compatible Java constructors and defaults.
- Added planner rules that separate Python calculations with incompatible
explicit concurrency values, including nested UDF calls.
- Applied explicit concurrency as configured operator parallelism for
synchronous
and asynchronous Python operators.
- Applied the minimum explicitly requested batch size to fused pandas UDFs
while
retaining the global configuration when no per-UDF value is set.
- Prevented Table map operators with conflicting explicit execution settings
from
being merged.
- Updated the DataFrame UDF API documentation and examples.
## Verifying this change
This change added and extended tests that can be verified as follows:
- Extended DataFrame UDF tests to cover argument validation, metadata
propagation,
backward compatibility, and execution.
- Added planner tests for recursive concurrency grouping, operator
splitting, and
map-merge compatibility.
- Added translation tests that verify configured operator parallelism,
inherited
parallelism, fused minimum batch size, global batch-size fallback, and
asynchronous UDF concurrency without starting a MiniCluster.
- Ran `pyflink.dataframe.tests.test_udf`: 15 tests passed.
- Ran the focused 66-module Maven reactor build: 41 planner tests and 3
translation
tests passed.
- Ran flake8 and targeted mypy checks for the affected Python API files.
## 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,
the public PyFlink DataFrame and Table UDF creation APIs gain optional
arguments
- 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? API documentation and docstrings
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Codex (GPT-5)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
