2.7 Transform Stream Operator
The TransformStreamAbsOperator
is used to transform the streaming data to another
streaming data.
There are one way to use the TransformStreamAbsOperator
:
Implement a custom TransformStreamAbsOperator
Just override the transform
method to return a new async iterable.
from typing import AsyncIterator
from dbgpt.core.awel import DAG, TransformStreamAbsOperator
class NumberDoubleOperator(TransformStreamAbsOperator[int, int]):
async def transform_stream(self, it: AsyncIterator) -> AsyncIterator[int]:
async for i in it:
# Double the number
yield i * 2
with DAG("numbers_dag") as dag:
task = NumberDoubleOperator()
Examples
Double The Numbers
Create a new file named transform_stream_operator_double_numbers.py
in the awel_tutorial
directory and add the following code:
import asyncio
from typing import AsyncIterator
from dbgpt.core.awel import DAG, TransformStreamAbsOperator, StreamifyAbsOperator
class NumberProducerOperator(StreamifyAbsOperator[int, int]):
"""Create a stream of numbers from 0 to `n-1`"""
async def streamify(self, n: int) -> AsyncIterator[int]:
for i in range(n):
yield i
class NumberDoubleOperator(TransformStreamAbsOperator[int, int]):
async def transform_stream(self, it: AsyncIterator) -> AsyncIterator[int]:
async for i in it:
# Double the number
yield i * 2
with DAG("numbers_dag") as dag:
task = NumberProducerOperator()
double_task = NumberDoubleOperator()
task >> double_task
async def print_stream(t, n: int):
# Call the streaming operator by `call_stream` method
async for i in await t.call_stream(call_data=n):
print(i)
asyncio.run(print_stream(double_task, 10))
And run the following command to execute the code:
poetry run python awel_tutorial/transform_stream_operator_double_numbers.py
And you will see the following output printed to the console.
0
2
4
6
8
10
12
14
16
18