DORA Python Echo 示例详解用三节点数据流验证端到端数据传递与校验【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora导读Python Echo 是 DORADataflow-Oriented Robotic Architecture仓库中最精简的 Python 数据流示例仅用三个节点便完整演示了一条数据从定时触发、跨节点透明转发、到下游校验的完整链路。本文以 examples/python-echo/README.md 为骨架结合三个节点源码、dataflow.yml编排文件以及 Python 节点 API 的底层实现讲解 DORA 数据流图中数据如何产生、如何流动、如何被验证的核心机制。读完本文你将掌握dora/timer/millis/N定时器输入的用法、PyArrow 数组的序列化与反序列化、事件元数据metadata的透传以及如何用dora run一键启动并验证一个多节点管道。示例整体架构python-echo是一个最小化的三节点管道用于演示 DORA 中端到端的数据传递与校验。其数据流拓扑如下timer (500ms) -- sender -- data -- echo -- data -- checker整条链路由四部分构成一个内置定时器timer周期性唤醒上游节点随后数据依次经过sender、echo、checker三个节点。完整的编排声明位于 examples/python-echo/dataflow.yml对应三个 Python 文件分别位于 examples/python-echo/sender.py、examples/python-echo/echo.py 与 examples/python-echo/checker.py。节点职责逐一看sender由内置定时器驱动的数据源sender每 500 ms 被内置定时器触发一次向下游发送一个固定的 PyArrow 数组[1, 2, 3, 4, 5]并且原样转发事件的元数据。核心代码如下Sender node: emits a fixed PyArrow array every 500ms. import pyarrow as pa from dora import Node def main(): node Node() for event in node: if event[type] INPUT: node.send_output(data, pa.array([1, 2, 3, 4, 5]), event[metadata]) elif event[type] STOP: break if __name__ __main__: main()值得注意的细节节点通过for event in node:持续消费事件流这是 DORA Python 节点 API 的标准写法事件按event[type]区分INPUT表示收到输入消息STOP表示数据流终止收到STOP后循环退出节点自然结束node.send_output(data, pa.array([1, 2, 3, 4, 5]), event[metadata])将pa.array()构造的 Arrow 数组发往名为data的输出并透传event[metadata]。echo验证数据跨节点无损的透明中继echo节点不做任何处理把收到的值value和元数据metadata原样重新发送充当透明中继角色用于验证数据经过一个节点跳转后没有被修改Echo node: forwards any incoming value and metadata unchanged. from dora import Node def main(): node Node() for event in node: if event[type] INPUT: node.send_output(data, event[value], event[metadata]) elif event[type] STOP: break if __name__ __main__: main()这里event[value]直接作为send_output的数据参数再次发出没有经过任何构造或转换从源码结构看这正是值 元数据均不变语义的实现基础。checker下游数据校验与结果输出checker接收 echo 转发来的数组与期望值[1, 2, 3, 4, 5]逐一比较每条消息打印[PASS]或[FAIL]并在数据流停止时打印最终通过计数Checker node: validates that each received array matches the expected value. from dora import Node EXPECTED [1, 2, 3, 4, 5] def main(): node Node() count 0 for event in node: if event[type] INPUT: received event[value].to_pylist() if received EXPECTED: count 1 print(f[PASS] #{count} data matches: {received}) else: print(f[FAIL] expected {EXPECTED}, got {received}) elif event[type] STOP: break print(fTotal PASS: {count}) if __name__ __main__: main()event[value].to_pylist()是读取 Arrow 数组的关键操作它把 PyArrow Array 转回 Python 原生列表从而可以与期望的 Python 列表直接比较。此外checker 维护了一个count计数器通过[PASS] #N的序号和最终的Total PASS: N可直观确认数据是否稳定、持续地在管道中流动。数据流编排文件剖析三个节点的连接关系完全由 examples/python-echo/dataflow.yml 声明nodes: - id: sender path: sender.py inputs: tick: dora/timer/millis/500 outputs: - data - id: echo path: echo.py inputs: data: sender/data outputs: - data - id: checker path: checker.py inputs: data: echo/data逐项说明nodes列表每个元素声明一个节点id是节点在管道内的唯一标识path指向节点程序Python 文件路径inputs映射键是节点内部使用的输入名值是上游数据源地址。格式为节点id/输出名例如sender/data表示取 sender 节点的 data 输出outputs列表声明节点对外暴露的输出名sender与echo都暴露名为data的输出tick: dora/timer/millis/500这是 DORA 内置定时器输入的地址语法表示每 500 毫秒向该输入注入一次触发事件。该语法同样出现在仓库核心代码与测试用例中例如 libraries/core/src/manifest/inject.rs 中的tick: dora/timer/millis/100以及 binaries/daemon/src/running_dataflow.rs 中动态拼接节点配置时生成的dora/timer/millis/100可见dora/timer/millis/N是 daemon 与核心库共同支持的通用定时触发机制N为毫秒数可按需替换为任意正整数如dora/timer/millis/100即 100 ms。从编排文件可以看出sender 没有任何上游输入、仅由定时器驱动因此它是数据流的源头echo 将 sender 的data输出转发为新的data输出checker 消费 echo 的data输出完成校验。环境准备与安装注意事项示例依赖 Python 节点 API 与 PyArrow安装命令如下pip install dora-rs pyarrow这里有一个极易踩坑的点README 特别用 Note 强调注意Python 中的导入名是dora即from dora import Node但 PyPI 上的包名是dora-rs。如果执行pip install dora安装到的是一个无关的包运行时会报ImportError: cannot import name Node。这一点与仓库中的 API 声明完全一致apis/python/node/dora/init.py 明确写明 You can install it viapip install dora-rs并在顶部导出了Node类同时条件性地尝试导入start_runtime避免在只装节点 API 而未装完整 CLI 时报错。因此安装包名与导入名的对应关系是PyPI 包dora-rs→ Python 模块dora。此外还需要可用的dora命令行工具提供dora run等命令在编译并安装仓库的 CLI 后Python 节点即可通过dora命令调度运行。运行与预期输出在examples/python-echo目录下执行dora run dataflow.yml预期输出如下[PASS] #1 data matches: [1, 2, 3, 4, 5] [PASS] #2 data matches: [1, 2, 3, 4, 5] ... Total PASS: N其中N表示数据流运行期间通过校验的消息总数。当用户终止数据流如 Ctrl-C 触发停止时checker 收到STOP事件后退出循环并打印Total PASS: N。本示例演示的 DORA 能力清单README 将本示例覆盖的功能点归纳如下表功能特性演示位置定时器触发节点dora/timer/millis/NSenderpa.array()数据序列化Sender元数据透传event[metadata]Sender、Echo透明中继节点Echo用event[value].to_pylist()读取数据Checker跨节点数据校验Checker对照仓库源码可以进一步印证这些能力背后的实现send_output的签名Python 绑定定义在 apis/python/node/src/lib.rs其形式为send_output(output_id, data, metadataNone)其中data为pyarrow.Arraymetadata为可选的Dict。README 中 sender/echo 的node.send_output(data, ..., event[metadata])正是该签名的直接使用元数据的作用metadata 是随消息传递的键值信息如示例代码注释中出现的{open_telemetry_context: 7632e76}在 sender→echo→checker 的链路中被逐跳原样携带这为跨节点传递上下文信息提供了统一通道数据以 Arrow 数组为载体消息内容通过 PyArrow Array 进行序列化接收端使用to_pylist()还原为 Python 列表保证了数据在节点间传输时类型与内容的确定性。进一步探索如果希望基于本示例继续深入可以在仓库中找到更丰富的参考资料更多 Python 数据流示例从最简单的三节点 examples/python-dataflow 到包含动态增删节点、并发读写、多数组、异步接收等场景的 examples/python-echo 同级目录下的其他示例可对照学习Python API 的完整用法阅读 apis/python/node/README.md 与 apis/python/node/dora/init.py、apis/python/node/dora/init.pyi 的类型声明可了解Node的全部方法包括send_output、send_output_raw零拷贝发送、next()拉取事件等数据流编排语法docs/yaml-spec.md 系统性地介绍了dataflow.yml的全部字段节点、输入输出、定时器、动态增删、容错策略等内置定时器实现在 binaries/daemon/src 与 libraries/core/src 中搜索dora/timer/millis可以看到定时器输入在 daemon 调度与核心描述符校验两个层面的处理逻辑。总之python-echo 虽然只有数十行代码却完整覆盖了 DORA 数据流中最核心的定时触发、Arrow 数据传递、元数据透传、跨节点校验四大机制是理解 DORA 多节点管道工作方式的最佳入门示例。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考