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 078dc821669 [FLINK-38563][python][docs] Update documentation of Python
AsyncFunction (#27195)
078dc821669 is described below
commit 078dc821669273e366dcfe9204c56dd6b8215114
Author: Dian Fu <[email protected]>
AuthorDate: Wed Nov 5 20:25:13 2025 +0800
[FLINK-38563][python][docs] Update documentation of Python AsyncFunction
(#27195)
---
.../docs/dev/datastream/operators/asyncio.md | 85 +++++++++++++++++--
.../docs/dev/datastream/operators/asyncio.md | 87 ++++++++++++++++++--
.../docs/reference/pyflink.common/index.rst | 3 +-
.../index.rst => pyflink.common/time.rst} | 58 +++++++++----
.../{functions.rst => asyncio.rst} | 94 ++++++++++++----------
.../reference/pyflink.datastream/functions.rst | 1 +
.../docs/reference/pyflink.datastream/index.rst | 1 +
.../pyflink/datastream/async_retry_predicates.py | 29 +++++++
flink-python/pyflink/datastream/functions.py | 5 ++
9 files changed, 294 insertions(+), 69 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/operators/asyncio.md
b/docs/content.zh/docs/dev/datastream/operators/asyncio.md
index e30344fbdf1..e160a0d05e9 100644
--- a/docs/content.zh/docs/dev/datastream/operators/asyncio.md
+++ b/docs/content.zh/docs/dev/datastream/operators/asyncio.md
@@ -61,11 +61,14 @@ Flink 的异步 I/O API 允许用户在流处理中使用异步请求客户端
在具备异步数据库客户端的基础上，实现数据流转换操作与数据库的异步 I/O 交互需要以下三部分：
- 实现分发请求的 `AsyncFunction`
-- 获取数据库交互的结果并发送给 `ResultFuture` 的 *回调* 函数
+- 如果使用的 Java API，获取数据库交互的结果并发送给 `ResultFuture` 的 *回调* 函数；如果使用的 Python API，可以通过
await 获取数据库交互的结果
- 将异步 I/O 操作应用于 `DataStream` 作为 `DataStream` 的一次转换操作, 启用或者不启用重试。
下面是基本的代码模板：
+{{< tabs "6c8c009c-4c12-4338-9eeb-3be83cfa9e36" >}}
+{{< tab "Java" >}}
+
```java
// 这个例子使用 Java 8 的 Future 接口（与 Flink 的 Future 相同）实现了异步请求和回调。
@@ -132,7 +135,74 @@ DataStream<Tuple2<String, String>> resultStream =
AsyncDataStream.unorderedWaitWithRetry(stream, new
AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100, asyncRetryStrategy);
```
-**重要提示**： 第一次调用 `ResultFuture.complete` 后 `ResultFuture` 就完成了。
+{{< /tab >}}
+{{< tab "Python" >}}
+
+```python
+from typing import List
+
+from pyflink.common import Time, Types
+from pyflink.datastream import AsyncFunction, AsyncDataStream,
async_retry_predicates
+from pyflink.datastream.functions import RuntimeContext, AsyncRetryStrategy
+
+
+class AsyncDatabaseRequest(AsyncFunction[str, (str, str)]):
+
+ def __init__(self, host, port, credentials):
+ self._host = host
+ self._port = port
+ self._credentials = credentials
+
+ def open(self, runtime_context: RuntimeContext):
+ # The database specific client that can issue concurrent requests with
callbacks
+ self._client = DatabaseClient(self._host, self._port,
self._credentials)
+
+ def close(self):
+ if self._client:
+ self._client.close()
+
+ async def async_invoke(self, value: str) -> List[(str, str)]:
+ try:
+ # issue the asynchronous request
+ result = await self._client.query(value)
+ return [(value, str(result))]
+ except Exception:
+ return [(value, None)]
+
+
+# 创建初始 DataStream
+stream = ...
+
+# 应用异步 I/O 转换操作，不启用重试
+result_stream = AsyncDataStream.unordered_wait(
+ data_stream=stream,
+ async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+ timeout=Time.seconds(10),
+ capacity=100,
+ output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+
+# 或应用异步 I/O 转换操作并启用重试
+# 通过工具类创建一个异步重试策略, 或用户实现自定义的策略
+async_retry_strategy = AsyncRetryStrategy.fixed_delay(
+ max_attempts=3,
+ backoff_time_millis=100,
+ result_predicate=async_retry_predicates.empty_result_predicate,
+ exception_predicate=async_retry_predicates.has_exception_predicate)
+
+# 应用异步 I/O 转换操作并启用重试
+result_stream_with_retry = AsyncDataStream.unordered_wait_with_retry(
+ data_stream=stream,
+ async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+ timeout=Time.seconds(10),
+ async_retry_strategy=async_retry_strategy,
+ capacity=1000,
+ output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+**重要提示**： 在 Java API 中，第一次调用 `ResultFuture.complete` 后 `ResultFuture` 就完成了。
后续的 `complete` 调用都将被忽略。
下面两个参数控制异步操作：
@@ -149,9 +219,13 @@ DataStream<Tuple2<String, String>> resultStream =
当异步 I/O 请求超时的时候，默认会抛出异常并重启作业。
如果你想处理超时，可以重写 `AsyncFunction#timeout` 方法。
-重写 `AsyncFunction#timeout` 时别忘了调用 `ResultFuture.complete()` 或者
`ResultFuture.completeExceptionally()`
+
+在 Java API 中，重写 `AsyncFunction#timeout` 时别忘了调用 `ResultFuture.complete()` 或者
`ResultFuture.completeExceptionally()`
以便告诉Flink这条记录的处理已经完成。如果超时发生时你不想发出任何记录，你可以调用
`ResultFuture.complete(Collections.emptyList())` 。
+在 Python API 中，可以返回一个 List 或者抛异常以便告诉Flink这条记录的处理已经完成。如果超时发生时你不想发出任何记录，
+你可以调用 `return []` 以返回一个空列表。
+
### 结果的顺序
`AsyncFunction` 发出的并发请求经常以不确定的顺序完成，这取决于请求得到响应的顺序。
@@ -160,9 +234,9 @@ Flink 提供两种模式控制结果记录以何种顺序发出。
- **无序模式**： 异步请求一结束就立刻发出结果记录。
流中记录的顺序在经过异步 I/O 算子之后发生了改变。
当使用 *处理时间* 作为基本时间特征时，这个模式具有最低的延迟和最少的开销。
- 此模式使用 `AsyncDataStream.unorderedWait(...)` 方法。
+ 此模式使用 `AsyncDataStream.unorderedWait(...)` 或者
`AsyncDataStream.unordered_wait(...)` 方法。
- - **有序模式**:
这种模式保持了流的顺序。发出结果记录的顺序与触发异步请求的顺序（记录输入算子的顺序）相同。为了实现这一点，算子将缓冲一个结果记录直到这条记录前面的所有记录都发出（或超时）。由于记录或者结果要在
checkpoint 的状态中保存更长的时间，所以与无序模式相比，有序模式通常会带来一些额外的延迟和 checkpoint 开销。此模式使用
`AsyncDataStream.orderedWait(...)` 方法。
+ - **有序模式**:
这种模式保持了流的顺序。发出结果记录的顺序与触发异步请求的顺序（记录输入算子的顺序）相同。为了实现这一点，算子将缓冲一个结果记录直到这条记录前面的所有记录都发出（或超时）。由于记录或者结果要在
checkpoint 的状态中保存更长的时间，所以与无序模式相比，有序模式通常会带来一些额外的延迟和 checkpoint 开销。此模式使用
`AsyncDataStream.orderedWait(...)` 或者 `AsyncDataStream.ordered_wait(...)` 方法。
### 事件时间
@@ -203,6 +277,7 @@ Flink 提供两种模式控制结果记录以何种顺序发出。
`DirectExecutor` 可以通过
`org.apache.flink.util.concurrent.Executors.directExecutor()` 或
`com.google.common.util.concurrent.MoreExecutors.directExecutor()` 获得。
+**注意:** 这仅适用于 Java API，在 Python API 中，您可以使用 await 等待异步执行的结果。
### 警告
diff --git a/docs/content/docs/dev/datastream/operators/asyncio.md
b/docs/content/docs/dev/datastream/operators/asyncio.md
index b587350ef3a..4c3e3434afb 100644
--- a/docs/content/docs/dev/datastream/operators/asyncio.md
+++ b/docs/content/docs/dev/datastream/operators/asyncio.md
@@ -75,11 +75,14 @@ Assuming one has an asynchronous client for the target
database, three parts are
with asynchronous I/O against the database:
- An implementation of `AsyncFunction` that dispatches the requests
- - A *callback* that takes the result of the operation and hands it to the
`ResultFuture`
+ - A *callback* that takes the result of the operation and hands it to the
`ResultFuture` in Java API or await the result of the operation in Python API
- Applying the async I/O operation on a DataStream as a transformation with
or without retry
The following code example illustrates the basic pattern:
+{{< tabs "6c8c009c-4c12-4338-9eeb-3be83cfa9e36" >}}
+{{< tab "Java" >}}
+
```java
// This example implements the asynchronous request and callback with Futures
that have the
// interface of Java 8's futures (which is the same one followed by Flink's
Future)
@@ -147,7 +150,74 @@ DataStream<Tuple2<String, String>> resultStream =
AsyncDataStream.unorderedWaitWithRetry(stream, new
AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100, asyncRetryStrategy);
```
-**Important note**: The `ResultFuture` is completed with the first call of
`ResultFuture.complete`.
+{{< /tab >}}
+{{< tab "Python" >}}
+
+```python
+from typing import List
+
+from pyflink.common import Time, Types
+from pyflink.datastream import AsyncFunction, AsyncDataStream,
async_retry_predicates
+from pyflink.datastream.functions import RuntimeContext, AsyncRetryStrategy
+
+
+class AsyncDatabaseRequest(AsyncFunction[str, (str, str)]):
+
+ def __init__(self, host, port, credentials):
+ self._host = host
+ self._port = port
+ self._credentials = credentials
+
+ def open(self, runtime_context: RuntimeContext):
+ # The database specific client that can issue concurrent requests with
callbacks
+ self._client = DatabaseClient(self._host, self._port,
self._credentials)
+
+ def close(self):
+ if self._client:
+ self._client.close()
+
+ async def async_invoke(self, value: str) -> List[(str, str)]:
+ try:
+ # issue the asynchronous request
+ result = await self._client.query(value)
+ return [(value, str(result))]
+ except Exception:
+ return [(value, None)]
+
+
+# create the original stream
+stream = ...
+
+# apply the async I/O transformation without retry
+result_stream = AsyncDataStream.unordered_wait(
+ data_stream=stream,
+ async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+ timeout=Time.seconds(10),
+ capacity=100,
+ output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+
+# or apply the async I/O transformation with retry
+# create an async retry strategy via utility class or a user defined strategy
+async_retry_strategy = AsyncRetryStrategy.fixed_delay(
+ max_attempts=3,
+ backoff_time_millis=100,
+ result_predicate=async_retry_predicates.empty_result_predicate,
+ exception_predicate=async_retry_predicates.has_exception_predicate)
+
+# apply the async I/O transformation with retry
+result_stream_with_retry = AsyncDataStream.unordered_wait_with_retry(
+ data_stream=stream,
+ async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+ timeout=Time.seconds(10),
+ async_retry_strategy=async_retry_strategy,
+ capacity=1000,
+ output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+**Important note**: The `ResultFuture` is completed with the first call of
`ResultFuture.complete` in the Java API.
All subsequent `complete` calls will be ignored.
The following three parameters control the asynchronous operations:
@@ -162,17 +232,21 @@ The following three parameters control the asynchronous
operations:
accumulate an ever-growing backlog of pending requests, but that it will
trigger backpressure once the capacity
is exhausted.
- - **AsyncRetryStrategy**: The asyncRetryStrategy defines what conditions
will trigger a delayed retry and the delay strategy,
+ - **AsyncRetryStrategy**: This parameter defines what conditions will
trigger a delayed retry and the delay strategy,
e.g., fixed-delay, exponential-backoff-delay, custom implementation, etc.
### Timeout Handling
When an async I/O request times out, by default an exception is thrown and job
is restarted.
If you want to handle timeouts, you can override the `AsyncFunction#timeout`
method.
-Make sure you call `ResultFuture.complete()` or
`ResultFuture.completeExceptionally()` when overriding
+
+In the Java API, make sure you call `ResultFuture.complete()` or
`ResultFuture.completeExceptionally()` when overriding
in order to indicate to Flink that the processing of this input record has
completed. You can call
`ResultFuture.complete(Collections.emptyList())` if you do not want to emit
any record when timeouts happen.
+In the Python API, you can return a collection of results or raise an
exception when overriding
+in order to indicate to Flink that the processing of this input record has
completed. You can return
+empty list by calling `return []` if you do not want to emit any record when
timeouts happen.
### Order of Results
@@ -182,14 +256,14 @@ To control in which order the resulting records are
emitted, Flink offers two mo
- **Unordered**: Result records are emitted as soon as the asynchronous
request finishes.
The order of the records in the stream is different after the async I/O
operator than before.
This mode has the lowest latency and lowest overhead, when used with
*processing time* as the basic time characteristic.
- Use `AsyncDataStream.unorderedWait(...)` for this mode.
+ Use `AsyncDataStream.unorderedWait(...)` or
`AsyncDataStream.unordered_wait(...)` for this mode.
- **Ordered**: In that case, the stream order is preserved. Result records
are emitted in the same order as the asynchronous
requests are triggered (the order of the operators input records). To
achieve that, the operator buffers a result record
until all its preceding records are emitted (or timed out).
This usually introduces some amount of extra latency and some overhead in
checkpointing, because records or results are maintained
in the checkpointed state for a longer time, compared to the unordered
mode.
- Use `AsyncDataStream.orderedWait(...)` for this mode.
+ Use `AsyncDataStream.orderedWait(...)` or
`AsyncDataStream.ordered_wait(...)` for this mode.
### Event Time
@@ -240,6 +314,7 @@ with the checkpoint bookkeeping happens in a dedicated
thread-pool anyways.
A `DirectExecutor` can be obtained via
`org.apache.flink.util.concurrent.Executors.directExecutor()` or
`com.google.common.util.concurrent.MoreExecutors.directExecutor()`.
+**NOTE:** This only applies for the Java API. In the Python API, you could
just await the asynchronous result.
### Caveats
diff --git a/flink-python/docs/reference/pyflink.common/index.rst
b/flink-python/docs/reference/pyflink.common/index.rst
index f663b0359ac..72eea108bed 100644
--- a/flink-python/docs/reference/pyflink.common/index.rst
+++ b/flink-python/docs/reference/pyflink.common/index.rst
@@ -28,4 +28,5 @@ This page gives an overview of all public PyFlink Common API.
config
typeinfo
job_info
- serializer
\ No newline at end of file
+ serializer
+ time
diff --git a/flink-python/docs/reference/pyflink.datastream/index.rst
b/flink-python/docs/reference/pyflink.common/time.rst
similarity index 62%
copy from flink-python/docs/reference/pyflink.datastream/index.rst
copy to flink-python/docs/reference/pyflink.common/time.rst
index f824a64956f..b188af06d72 100644
--- a/flink-python/docs/reference/pyflink.datastream/index.rst
+++ b/flink-python/docs/reference/pyflink.common/time.rst
@@ -16,24 +16,50 @@
limitations under the License.
################################################################################
-==================
-PyFlink DataStream
-==================
-This page gives an overview of all public PyFlink DataStream API.
+====
+Time
+====
-.. toctree::
- :maxdepth: 2
+Duration
+--------
- stream_execution_environment
- datastream
- functions
- state
- timer
- window
- checkpoint
- sideoutput
- connectors
- formats
+.. currentmodule:: pyflink.common.time
+.. autosummary::
+ :toctree: api/
+ Duration.of_days
+ Duration.of_hours
+ Duration.of_minutes
+ Duration.of_millis
+ Duration.of_seconds
+ Duration.of_nanos
+
+
+Instant
+-------
+
+.. currentmodule:: pyflink.common.time
+
+.. autosummary::
+ :toctree: api/
+
+ Instant.of_epoch_milli
+ Instant.to_epoch_milli
+
+
+Time
+----
+
+.. currentmodule:: pyflink.common.time
+
+.. autosummary::
+ :toctree: api/
+
+ Time.milliseconds
+ Time.seconds
+ Time.minutes
+ Time.hours
+ Time.days
+ Time.to_milliseconds
diff --git a/flink-python/docs/reference/pyflink.datastream/functions.rst
b/flink-python/docs/reference/pyflink.datastream/asyncio.rst
similarity index 50%
copy from flink-python/docs/reference/pyflink.datastream/functions.rst
copy to flink-python/docs/reference/pyflink.datastream/asyncio.rst
index 6337688c696..ed462639293 100644
--- a/flink-python/docs/reference/pyflink.datastream/functions.rst
+++ b/flink-python/docs/reference/pyflink.datastream/asyncio.rst
@@ -17,60 +17,72 @@
################################################################################
-=========
-Functions
-=========
+================
+Asynchronous I/O
+================
-RuntimeContext
---------------
+AsyncDataStream
+---------------
+
+.. currentmodule:: pyflink.datastream.async_data_stream
+
+.. autosummary::
+ :toctree: api/
+
+ AsyncDataStream.unordered_wait
+ AsyncDataStream.unordered_wait_with_retry
+ AsyncDataStream.ordered_wait
+ AsyncDataStream.ordered_wait_with_retry
+
+
+AsyncFunction
+-------------
.. currentmodule:: pyflink.datastream.functions
.. autosummary::
:toctree: api/
- RuntimeContext.get_task_name
- RuntimeContext.get_number_of_parallel_subtasks
- RuntimeContext.get_max_number_of_parallel_subtasks
- RuntimeContext.get_index_of_this_subtask
- RuntimeContext.get_attempt_number
- RuntimeContext.get_task_name_with_subtasks
- RuntimeContext.get_job_parameter
- RuntimeContext.get_metrics_group
- RuntimeContext.get_state
- RuntimeContext.get_list_state
- RuntimeContext.get_map_state
- RuntimeContext.get_reducing_state
- RuntimeContext.get_aggregating_state
+ AsyncFunction.async_invoke
+ AsyncFunction.timeout
+
+AsyncRetryStrategy
+------------------
-Function
---------
+.. currentmodule:: pyflink.datastream.functions
-All user-defined functions.
+.. autosummary::
+ :toctree: api/
+
+ AsyncRetryStrategy.can_retry
+ AsyncRetryStrategy.get_backoff_time_millis
+ AsyncRetryStrategy.get_retry_predicate
+ AsyncRetryStrategy.no_restart
+ AsyncRetryStrategy.fixed_delay
+ AsyncRetryStrategy.exponential_backoff
+
+
+AsyncRetryPredicate
+-------------------
.. currentmodule:: pyflink.datastream.functions
.. autosummary::
:toctree: api/
- MapFunction
- CoMapFunction
- FlatMapFunction
- CoFlatMapFunction
- ReduceFunction
- AggregateFunction
- ProcessFunction
- KeyedProcessFunction
- CoProcessFunction
- KeyedCoProcessFunction
- WindowFunction
- AllWindowFunction
- ProcessWindowFunction
- ProcessAllWindowFunction
- KeySelector
- NullByteKeySelector
- FilterFunction
- Partitioner
- BroadcastProcessFunction
- KeyedBroadcastProcessFunction
+ AsyncRetryPredicate.result_predicate
+ AsyncRetryPredicate.exception_predicate
+
+
+Async Retry Predicates Utilities
+--------------------------------
+
+.. currentmodule:: pyflink.datastream.async_retry_predicates
+
+.. autosummary::
+ :toctree: api/
+
+ empty_result_predicate
+ has_exception_predicate
+ exception_type_predicate
diff --git a/flink-python/docs/reference/pyflink.datastream/functions.rst
b/flink-python/docs/reference/pyflink.datastream/functions.rst
index 6337688c696..35cc18837dd 100644
--- a/flink-python/docs/reference/pyflink.datastream/functions.rst
+++ b/flink-python/docs/reference/pyflink.datastream/functions.rst
@@ -74,3 +74,4 @@ All user-defined functions.
Partitioner
BroadcastProcessFunction
KeyedBroadcastProcessFunction
+ AsyncFunction
diff --git a/flink-python/docs/reference/pyflink.datastream/index.rst
b/flink-python/docs/reference/pyflink.datastream/index.rst
index f824a64956f..17c68931afe 100644
--- a/flink-python/docs/reference/pyflink.datastream/index.rst
+++ b/flink-python/docs/reference/pyflink.datastream/index.rst
@@ -33,6 +33,7 @@ This page gives an overview of all public PyFlink DataStream
API.
window
checkpoint
sideoutput
+ asyncio
connectors
formats
diff --git a/flink-python/pyflink/datastream/async_retry_predicates.py
b/flink-python/pyflink/datastream/async_retry_predicates.py
new file mode 100644
index 00000000000..eabe366c124
--- /dev/null
+++ b/flink-python/pyflink/datastream/async_retry_predicates.py
@@ -0,0 +1,29 @@
+################################################################################
+# 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.
+################################################################################
+
+
+def empty_result_predicate(result: list):
+ return result is None or len(result) == 0
+
+
+def has_exception_predicate(error):
+ return error is not None
+
+
+def exception_type_predicate(error, expected_error_type):
+ return isinstance(error, expected_error_type)
diff --git a/flink-python/pyflink/datastream/functions.py
b/flink-python/pyflink/datastream/functions.py
index 3edfc6d7bef..db4366390e1 100644
--- a/flink-python/pyflink/datastream/functions.py
+++ b/flink-python/pyflink/datastream/functions.py
@@ -907,6 +907,7 @@ class AsyncRetryPredicate(ABC, Generic[OUT]):
Interface encapsulates an asynchronous retry predicate.
"""
+ @abstractmethod
def result_predicate(self) -> Optional[Callable[[List[OUT]], bool]]:
"""
An optional Python predicate function that defines a condition on
asyncFunction's future
@@ -915,6 +916,7 @@ class AsyncRetryPredicate(ABC, Generic[OUT]):
"""
pass
+ @abstractmethod
def exception_predicate(self) -> Optional[Callable[[Exception], bool]]:
"""
An optional Python predicate function that defines a condition on
asyncFunction's exception
@@ -929,18 +931,21 @@ class AsyncRetryStrategy(ABC, Generic[OUT]):
Interface encapsulates an asynchronous retry strategy.
"""
+ @abstractmethod
def can_retry(self, current_attempts: int) -> bool:
"""
Whether the next attempt can happen.
"""
pass
+ @abstractmethod
def get_backoff_time_millis(self, current_attempts: int) -> int:
"""
The delay time of next attempt.
"""
pass
+ @abstractmethod
def get_retry_predicate(self) -> AsyncRetryPredicate[OUT]:
"""
Returns the defined retry predicate.
