[FLINK-40431][python] Support per-UDF concurrency and batch size in PyFlink DataFrame API

flink-python/docs/reference/pyflink.dataframe/index.rst

@@ -26,6 +26,7 @@ This page gives an overview of all public PyFlink DataFrame APIs.
     :maxdepth: 1
 
     dataframe
+    udf
     creation
     io
     datatype

flink-python/docs/reference/pyflink.dataframe/udf.rst

@@ -0,0 +1,47 @@
+.. ################################################################################
+     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.
+   ################################################################################
+
+=============================
+User-Defined Scalar Functions
+=============================
+
+Use :func:`pyflink.dataframe.udf` to apply Python code to one or more DataFrame
+columns. A scalar UDF produces one logical output column and can be used in
+:meth:`~pyflink.dataframe.DataFrame.with_column`,
+:meth:`~pyflink.dataframe.DataFrame.with_columns`, and
+:meth:`~pyflink.dataframe.DataFrame.select`.
+
+DataFrame scalar UDFs support synchronous, asynchronous, and pandas-vectorized
+callables. See :func:`pyflink.dataframe.udf` for declaration forms, type
+inference, execution modes, and examples.
+
+Use the ``concurrency`` argument to set the parallelism of the Python operator
+that executes a UDF. UDFs with different explicit concurrency values are placed
+in separate operators. For pandas UDFs, ``batch_size`` sets the maximum Arrow
+batch size. If compatible pandas UDFs share an operator, the smallest explicit
+batch size is used; otherwise the configured default applies.
+
+API Reference
+=============
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+    :toctree: api/
+
+    udf

flink-python/pyflink/dataframe/__init__.py

@@ -54,13 +54,15 @@
 from pyflink.dataframe.dataframe import DataFrame, GroupedDataFrame, col, lit
 from pyflink.dataframe.datatype import DataType
 from pyflink.dataframe.io import read_generic
+from pyflink.dataframe.udf import udf
 
 __all__ = [
     "DataFrame",
     "GroupedDataFrame",
     "DataType",
     "col",
     "lit",
+    "udf",
     "from_arrow",
     "from_dict",
     "from_pandas",

flink-python/pyflink/dataframe/dataframe.py

@@ -210,10 +210,19 @@ def with_column(
 
             >>> import pyflink.dataframe as pf
             >>> df = pf.from_records([{"left": 1, "right": 2}])
-            >>> result = df.with_column(
+
+            >>> with_expression = df.with_column(
             ...     "total", lambda current: current["left"] + current["right"]
             ... )
 
+            >>> @pf.udf
+            ... def add(left: int, right: int) -> int:
+            ...     return left + right
+
+            >>> with_udf = df.with_column(
+            ...     "total", add(pf.col("left"), pf.col("right"))
+            ... )
+
         .. versionadded:: 2.4.0
         """
         if not isinstance(name, str):