---
title: DataFlow接口列表
description: "使用DataFlow Python接口构造DataFlow图进行推理。支持定义图处理点，UDF处理点，描述处理点之间的数据流关系；支持导入TensorFlow, ONNX, MindSpore的IR文件作为图处理点计算逻辑定义。"
url: https://www.hiascend.com/document/detail/zh/canncommercial/latest/others/dataflowdevg/dfrefp_23_0001.html
sourcePath: /source/zh/canncommercial/900/others/dataflowdevg/dfrefp_23_0001.html
indexId: e9b88e6e75dc526d5fe160ef9459a677d150b379cbf0fedea89ba82a765ec59269
---
# DataFlow接口列表

使用DataFlow Python接口构造DataFlow图进行推理。支持定义图处理点，UDF处理点，描述处理点之间的数据流关系；支持导入TensorFlow, ONNX, MindSpore的IR文件作为图处理点计算逻辑定义。

#### DataFlow构图接口


**表1 DataFlow构图接口**

| 接口名称 | 简介 |
| --- | --- |
| dataflow.FlowData | DataFlow Graph中的数据节点，每个FlowData对应一个输入。 |
| FlowNode | DataFlow Graph中的计算节点。 |
| add\_process\_point | 给FlowNode添加映射的pp，当前一个FlowNode仅能添加一个pp，添加后会默认将FlowNode的输入输出和pp的输入输出按顺序进行映射。 |
| map\_input | 给FlowNode映射输入，表示将FlowNode的第node\_input\_index个输入给到ProcessPoint的第pp\_input\_index个输入，并且给ProcessPoint的该输入设置上attr里的所有属性，返回映射好的FlowNode节点。该函数可选，不被调用时会默认按顺序去映射FlowNode和ProcessPoint的输入。 |
| map\_output | 给FlowNode映射输出，表示将pp的第pp\_output\_index个输出给到FlowNode的第node\_output\_index个输出，返回映射好的FlowNode节点。 |
| set\_attr | 设置FlowNode的属性。 |
| \_\_call\_\_ | 调用FlowNode进行计算。 |
| set\_balance\_scatter | 设置节点balance scatter属性，具有balance scatter属性的UDF可以使用balance options设置负载均衡输出。 |
| set\_balance\_gather | 设置节点balance gather属性，具有balance gather属性的UDF可以使用balance options设置负载均衡亲和输出。 |
| set\_alias | 设置节点别名，使用option:ge.experiment.data\_flow\_deploy\_info\_path指定节点部署位置时，flow\_node\_list字段可使用别名进行指定。 |
| dataflow.FlowFlag | 设置FlowMsg消息头中的flags。 |
| FlowGraph | DataFlow的graph，由输入节点FlowData和计算节点FlowNode构成。 |
| set\_contains\_n\_mapping\_node | 设置FlowGraph是否包含n\_mapping节点。 |
| set\_inputs\_align\_attrs | 设置FlowGraph中的输入对齐属性。 |
| set\_exception\_catch | 设置用户异常捕获功能是否开启。 |
| dataflow.FlowOutput | 描述FlowNode的输出。 |
| dataflow.Framework | 设置原始网络模型的框架类型。 |
| FuncProcessPoint | FuncProcessPoint的构造函数，返回一个FuncProcessPoint对象。 |
| set\_init\_param | 设置FuncProcessPoint的初始化参数。 |
| add\_invoked\_closure | 添加FuncProcessPoint调用的GraphProcessPoint或者FlowGraphProcessPoint，返回添加好的FuncProcessPoint。 |
| GraphProcessPoint | GraphProcessPoint构造函数，返回一个GraphProcessPoint对象。 |
| fnode | 根据当前的GraphProcessPoint生成一个FlowNode，返回一个FlowNode对象。 |
| dataflow.FlowGraphProcessPoint | GraphProcessPoint构造函数，返回一个GraphProcessPoint对象。 |
| Tensor | Tensor的构造函数。 |
| numpy | 将Tensor转换到numpy的ndarray。 |
| dataflow.TensorDesc | Tensor的描述函数。 |
| dataflow.alloc\_tensor | 根据shape、data type以及对齐大小申请dataflow tensor。 |
| dataflow.utils.generate\_deploy\_template | 根据FlowGraph生成指定部署位置的option:"ge.experiment.data\_flow\_deploy\_info\_path"所需要的文件的模板。 |
| register | 注册自定义类型对应的序列化、反序列化、计算size的函数，可结合feed，fetch接口使用，用于feed/fetch任意Python类型。 |
| registered | 判断消息类型ID是否被注册过。 |
| get\_msg\_type | 根据类型定义获取注册的消息类型ID。 |
| get\_serialize\_func | 根据消息类型ID获取注册的序列化函数。 |
| get\_deserialize\_func | 根据消息类型ID获取注册的反序列化函数。 |
| get\_size\_func | 根据消息类型ID获取注册的计算序列化内存大小的函数。 |
| deserialize\_from\_file | 从序列化的pickle文件进行反序列化恢复Python对象。 |
| pyflow | 支持将函数作为pipeline任务在本地或者远端运行。 |
| method | 对于复杂场景支持将类作为pipeline任务在本地或者远端运行。 |
| npu\_model | 如果UDF部署在host侧，执行时数据需要从device拷贝到本地进行运算。对于PyTorch场景，如果计算全在device侧，输入输出也是在device侧，执行时数据需要从device拷贝到host，执行后PyTorch再将数据搬到device侧，影响执行性能，使用npu\_model可以优化为不搬移数据（即直接下沉到device执行）的方式触发执行。 |
| dataflow.CountBatch | CountBatch功能是指基于UDF为计算处理点将多个数据按batchSize组成batch。该功能应用于dataflow异步场景。 |
| dataflow.TimeBatch | TimeBatch功能是基于UDF为前提的。 正常模型每次处理一个数据，当需要一次处理一批数据时，就需要将这批数据组成一个Batch。最基本的Batch方式是将这批N个数据直接拼接，然后shape前加N，而某些场景需要将某段或者某几段时间数据组成一个batch，并且按特定的维度拼接，则可以通过使用TimeBatch功能来组Batch。 |


#### DataFlow运行接口


**表2 DataFlow运行接口**

| 接口名称 | 简介 |
| --- | --- |
| dataflow.init | 初始化dataflow时的options。 |
| FlowInfo | DataFlow的flow信息。 |
| set\_user\_data | 设置用户信息。 |
| get\_user\_data | 获取用户信息。 |
| user\_data | 获取用户信息。 |
| data\_size | 获取user\_data的长度。 |
| start\_time | 以属性方式读取和设置FlowInfo的开始时间。 |
| end\_time | 以属性方法读取和设置FlowInfo的结束时间。 |
| flow\_flags | 以属性方法读取和设置FlowInfo的flow\_flags。 |
| transaction\_id | 以属性方式读写事务ID。 |
| feed\_data | 将数据输入到Graph。 |
| feed | 将数据输入到Graph，支持可序列化的任意的输入。 |
| fetch\_data | 获取Graph输出数据。 |
| fetch | 获取Graph输出数据。支持可序列化的任意的输出。 |
| dataflow.finalize | 释放dataflow初始化的资源。 |
| dataflow.get\_running\_device\_id | UDF执行时获取当前UDF的运行device\_id, 信息来源和UDF部署位置的配置。 |
| dataflow.get\_running\_instance\_id | UDF执行时获取当前UDF的运行实例ID，该信息来源于data\_flow\_deploy\_info.json中的logic\_device\_list配置。 |
| dataflow.get\_running\_instance\_num | UDF执行时获取当前UDF的运行实例个数，该信息来源于data\_flow\_deploy\_info.json中的logic\_device\_list配置。 |


#### 模块

dataflow module：公共接口的命名空间


#### 类

- class CountBatch：CountBatch属性的类
- class FlowData：输入节点类
- class FlowFlag：数据标记类
- class FlowGraph：dataflow的图类
- class FlowInfo：指定输入输出数据携带的信息类
- class FlowNode：计算节点类
- class FlowOutput：计算节点的输出类
- class Framework：IR文件的框架类型的枚举类
- class FuncProcessPoint：UDF处理点类
- class GraphProcessPoint：图处理点类
- class Tensor：张量数据类
- class TensorDesc：张量的描述类
- class TimeBatch：TimeBatch属性的类


#### 函数

- init(...)：dataflow的资源初始化方法
- finalize(...)：dataflow的资源释放方法


#### 其他成员


**表3 其他成员**

| 名称 | 简介 |
| --- | --- |
| DT\_FLOAT | df.data\_type.DType的对象 32位单精度浮点数 |
| DT\_FLOAT16 | df.data\_type.DType的对象 16位半精度浮点数 |
| DT\_INT8 | df.data\_type.DType的对象 有符号8位整数 |
| DT\_INT16 | df.data\_type.DType的对象 有符号16位整数 |
| DT\_UINT16 | df.data\_type.DType的对象 无符号16位整数 |
| DT\_UINT8 | df.data\_type.DType的对象 无符号8位整数 |
| DT\_INT32 | df.data\_type.DType的对象 有符号32位整数 |
| DT\_INT64 | df.data\_type.DType的对象 有符号64位整数 |
| DT\_UINT32 | df.data\_type.DType的对象 无符号32位整数 |
| DT\_UINT64 | df.data\_type.DType的对象 无符号64位整数 |
| DT\_BOOL | df.data\_type.DType的对象 布尔类型 |
| DT\_DOUBLE | df.data\_type.DType的对象 64位双精度浮点数 |
| DT\_STRING | df.data\_type.DType的对象 字符串类型 |