pyflow

Applicability

Product	Supported
Atlas A3 training product/Atlas A3 inference product	√
Atlas A2 training product/Atlas A2 inference product	√
Atlas 200I/500 A2 inference product	x
Atlas inference product	x
Atlas training product	x

Function Description

Enables functions to be run locally or remotely as pipeline tasks. You can use @pyflow to decorate a function to indicate that the function needs to be run in pipeline mode. After a user class or function is decorated with @pyflow, the fnode graph construction method is automatically added. For details about how to use the method, see Examples.

Prototype

@pyflow

Parameters

Parameter

Data Type

Description

num_returns

int

Indicates the number of outputs when a decorator is used. If this parameter is not set, only one value is returned by default. You can either use this parameter to set the number of returned values or use type annotations to set the number and type of returned values.

resources

dict

Indicates the resource information required by the current function, which can be memory, num_cpus, or num_npus. The unit of memory is MB. num_npus indicates the number of NPUs to be used. num_npus is reserved and can only be set to 1. Example: {"memory": 100, "num_cpus": 1, "num_npus": 1}

stream_input

str

Indicates that the input of the current function is a stream input (that is, the input parameter of the function is a queue). Currently, only the queue type is supported. You can obtain data from the input queue.

choice_output

function

Indicates that the output of the current function is optional. Only the output that meets the conditions (the condition is a user-defined function) is returned. For example:

choice_output=lambda e: e is not None

This example indicates that only outputs other than None are returned.

visible_device_enable

bool

After this function is enabled, the UDF process automatically sets ASCEND_RT_VISIBLE_DEVICES based on the configured num_npus and calls the get_running_device_id API to obtain the corresponding logical ID. Currently, num_npus supports only 1. Therefore, the get_running_device_id API returns 0 in this scenario.

env_hook_func

function

Sets up the Python environment or performs the import operation before the Python UDF is initialized.

The hook function does not support inputs or outputs.

Returns

Decorated class or function.

A DfException is thrown upon exceptions. You can catch DfException and retrieve its error_code and message attributes to check the specific error code and error details. For details, see DataFlow Error Codes.

Examples

# current is udf.py
import dataflow as df
@df.pyflow(num_returns=2, resources={"memory": 100, "num_cpus": 1})
def func1(a, b):
    return a + b,a - b

@df.pyflow
def func2(a, b):
    return a + b

@df.pyflow(stream_input='Queue')
def func3(a, b):
    data1 = a.get()
    data2 = a.get()
    data3 = b.get()
    return data1 + data2 + data3

@df.pyflow(choice_output=lambda e: e is not None)
def func4(self, a) -> Tuple[int, int]:
    return None, a  # Non-null values are sent to the corresponding outputs based on the lambda function.

# current is graph.py
from UDF import func2
import dataflow as df

# Graph construction
# Define input.
data0 = df.FlowData()
data1 = df.FlowData()
# Use the fnode method automatically generated by func2 to construct a graph.
func_out = func2.fnode()(data0, data1)

# Construct FlowGraph.
dag = df.FlowGraph([func_out])

Constraints

The cloudpickle package of the corresponding Python version must be installed in the environment.

In the streaming input scenario, the DataFlow framework does not support data alignment and exception handling.

Parent topic: Graph Construction APIs