[
https://issues.apache.org/jira/browse/FLINK-40471?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18126217#comment-18126217
]
ChenZhuo commented on FLINK-40471:
----------------------------------
Hi Liu Liu, I'd like to work on FLINK-40471. Could a committer assign this
issue to me (JIRA username: chenzhuo)?
I have checked the current DataFrame UDF wrapper and async scalar execution
path. I propose adding the optional concurrency and batch_size arguments,
preserving existing behavior when omitted, validating supplied values, and
carrying the settings through to execution rather than changing the shared
TableEnvironment configuration. Tests would cover invalid values, default
fallback, and two UDFs with different settings in the same pipeline, plus
documentation examples.
Before implementation, could you clarify whether concurrency is intended only
for async scalar UDFs, and whether batch_size applies to pandas/Arrow UDFs?
What should happen if either option is supplied for an unsupported UDF kind? I
will confirm the implementation approach here before starting. Thanks!
> Add per-UDF concurrency and batch size support to DataFrame API
> ---------------------------------------------------------------
>
> Key: FLINK-40471
> URL: https://issues.apache.org/jira/browse/FLINK-40471
> Project: Flink
> Issue Type: Sub-task
> Reporter: Liu Liu
> Priority: Major
>
> Support specifying per-UDF concurrency and batch size in UDF creation:
> {code:java}
> def udf(
> func=None,
> *,
> return_dtype: DataType = None,
> deterministic: bool = True,
> name: str = None,
> concurrency: Optional[int] = None, // per-UDF concurrency
> batch_size: Optional[int] = None, // per-UDF batch size
> ) -> DataFrameUDFWrapper {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
