DORA 与 C++ 的 Apache Arrow 数据交换实战：基于 c++-arrow-dataflow 示例深入解析

DORA 与 C 的 Apache Arrow 数据交换实战基于 c-arrow-dataflow 示例深入解析【免费下载链接】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本指南围绕 DORA 仓库中的 examples/c-arrow-dataflow 示例展开讲解如何让 C 节点与 DORA 基于 Rust 的运行时高效交换结构化数据。你将掌握event_as_arrow_input()/event_as_arrow_input_with_info()与send_arrow_output()的完整用法、跨语言传输的底层原理Apache Arrow C Data Interface、可复用的Metadata对象全量 API以及从安装 Apache Arrow C 到一键编译运行整个 dataflow 的完整实操路径。示例概览一个双 C 节点的数据流c-arrow-dataflow示例演示的是C 节点通过dora-node-api.h头文件暴露的函数接收和发送 Arrow 数组数据在 DORA 的 Rust 运行时与 C 之间以 Apache Arrow 的内存高效序列化格式流动从而无缝跨越语言边界。示例的 dataflow 声明位于 examples/c-arrow-dataflow/dataflow.ymlnodes: - id: cxx-node-rust-api path: build/node_rust_api inputs: tick: dora/timer/millis/300 outputs: - counter - id: cxx-node-rust2-api path: build/node_rust_api inputs: tick: cxx-node-rust-api/counter outputs: - counter两个节点使用同一个C 二进制build/node_rust_api源码为 node-rust-api/main.cccxx-node-rust-api由 DORA 内置定时器dora/timer/millis/300每 300 毫秒触发一次向counter输出端发送一个Int32Arrow 数组值 10、100、1000并附带包含producer、iteration、successful、multiples、weights、notes等键的元数据cxx-node-rust2-api订阅上游的counter输出收到事件后通过 Arrow 输入 API 解包数组与元数据并逐项打印然后同样回发一个输出。节点主循环只运行 10 次迭代while (counter10)随后收到Stop或AllInputsClosed事件即退出——因此数据流会自动结束无需手动干预。接收 Arrow 输入从事件中解出数组与元数据在 main.cc 中receive_and_print_input展示了完整的接收路径。首先调用event_as_arrow_input_with_info它会填充两个由 C 侧声明的 Arrow 结构体并额外返回输入 ID 与元数据struct ArrowArray c_array; struct ArrowSchema c_schema; auto input_info event_as_arrow_input_with_info( std::move(event), reinterpret_castuint8_t*(c_array), reinterpret_castuint8_t*(c_schema) ); if (!input_info.error.empty()) { std::cerr Error getting Arrow array: std::string(input_info.error) std::endl; return nullptr; } // 打印输入 ID 与元数据时间戳 std::cout Input ID: std::string(input_info.id) std::endl; auto metadata std::move(input_info.metadata); std::cout Metadata timestamp: metadata-timestamp() std::endl;随后把 C Data Interface 结构体导入为标准arrow::Arrayauto result2 arrow::ImportArray(c_array, c_schema); std::shared_ptrarrow::Array input_array result2.ValueOrDie(); std::cout Received Arrow array: input_array-ToString() std::endl; std::cout Array details: type input_array-type()-ToString() , length input_array-length() std::endl;从 C Node API 的 FFI 定义apis/c/node/src/lib.rs可以看到接收侧有两个函数可选函数返回内容适用场景event_as_arrow_input(event, out_array, out_schema)仅导出 Arrow 数组数据原始版本只关心 payload、不需要输入 ID 与元数据event_as_arrow_input_with_info(event, out_array, out_schema)返回ArrowInputInfo { id, metadata, error }需要输入 ID、消息元数据以及错误信息的完整场景底层实现中Rust 侧会把事件里的 Arrow 数据通过arrow::ffi::to_ffi转换为FFI_ArrowArray与FFI_ArrowSchema结构体并写入调用者提供的指针lib.rs#L777-L819。这正是 Apache Arrow C Data Interface 的核心价值零拷贝地把 Rust 的 Arrow 数组内存布局暴露给 CC 侧用arrow::ImportArray还原为std::shared_ptrarrow::Array即可直接处理。发送 Arrow 输出导出数组并附带元数据发送端的关键函数是send_arrow_output。在 main.cc#L127-L177 的send_output函数中先通过arrow::ExportArray把std::shared_ptrarrow::Array导出为 C Data Interface 结构体再连同元数据一起发送struct ArrowArray out_c_array; struct ArrowSchema out_c_schema; auto export_status arrow::ExportArray(*output_array, out_c_array, out_c_schema); if (!export_status.ok()) { std::cerr Failed to export Arrow array: export_status.ToString() std::endl; return false; } auto metadata new_metadata(); metadata-set_string(producer, cpp-node); metadata-set_int(iteration, counter); metadata-set_bool(successful, true); rust::Vecint64_t doubled_values; doubled_values.push_back(counter); doubled_values.push_back(counter * 2); doubled_values.push_back(counter * 3); metadata-set_list_int(multiples, std::move(doubled_values)); rust::Vecdouble weights; weights.push_back(0.1); weights.push_back(0.2); metadata-set_list_float(weights, std::move(weights)); rust::Vecrust::String notes; notes.push_back(generated); notes.push_back(from_cpp); metadata-set_list_string(notes, std::move(notes)); auto send_result send_arrow_output( dora_node.send_output, counter, reinterpret_castuint8_t*(out_c_array), reinterpret_castuint8_t*(out_c_schema), std::move(metadata) ); if (!send_result.error.empty()) { std::cerr Error sending Arrow array: send_result.error std::endl; return false; } return true;五参数重载与四参数重载C 头文件为send_arrow_output提供两个重载lib.rs#L398-L412五参数版本send_arrow_output(sender, id, array_ptr, schema_ptr, metadata)把new_metadata()构造的元数据随消息一起发出上文示例即此版本四参数版本不需要发送元数据时调用省略 metadata 参数即可。所有权语义结构体被消费在 Rust 侧的send_arrow_output_impllib.rs#L1548-L1601中有一个容易踩坑的细节array_ptr与schema_ptr指向的ArrowArray/ArrowSchema结构体会被std::ptr::read读取并转移所有权随后原地写入std::mem::zeroed()清零。也就是说调用后C 侧那两个结构体变量不可再复用否则读取被清零的内存会导致未定义行为每个待发送的数组需要重新arrow::ExportArray生成新的结构体这也是该函数在头文件中被声明为unsafe的原因——调用方必须保证指针有效且结构体生命周期符合约定。Metadata 对象为 Arrow 消息附加结构化参数C Node API 暴露了一个可复用的Metadata对象用于在发送前填充、在接收后读取消息参数。创建方式为new_metadata()返回一个rust::BoxMetadata智能指针。支持的参数类型与 setter类型SetterC 参数类型读取方法布尔set_bool(key, value)boolget_bool(key)整数set_int(key, value)int64_tget_int(key)浮点set_float(key, value)doubleget_float(key)字符串set_string(key, value)rust::Stringget_str(key)整数列表set_list_int(key, value)rust::Vecint64_tget_list_int(key)浮点列表set_list_float(key, value)rust::Vecdoubleget_list_float(key)字符串列表set_list_string(key, value)rust::Vecrust::Stringget_list_string(key)时间戳set_timestamp(key, value)int64_tUnix 纪元以来的纳秒get_timestamp(key)这些 setter/getter 在 lib.rs#L426-L448 的 FFI 声明中定义Rust 侧把值存入内部的BTreeMapString, DoraParameterlib.rs#L821-L1100。值得注意的实现细节时间戳以i64纳秒数跨 FFI 边界传递。set_timestamp内部用欧几里得除法div_euclid/rem_euclid把纳秒拆成(secs, subsec_nanos)这样1970 年之前的负时间戳也能正确往返不会因整除截断产生负余数而被拒绝lib.rs#L1033-L1051每个 setter 都返回Result()写入失败如时间戳超出可表示范围会以错误形式返回不会直接崩溃。读取时先查类型再取值MetadataValueType枚举覆盖全部八种类型Bool、Integer、Float、String、ListInt、ListFloat、ListString、Timestamp。推荐用metadata-type(key)先确认类型、再调用对应的 getterlist_keys()可以枚举全部键。示例中 main.cc#L33-L112 完整演示了针对每种类型的switch分支打印逻辑并捕获std::exception以处理键缺失或类型不匹配的情况。JSON 序列化to_json()把整个元数据对象序列化为 JSON 字符串结构为{ timestamp: 1730000000000000000, parameters: { producer: { String: cpp-node }, ...: ... } }其中timestamp是消息时间戳来自 DORA 的混合逻辑时钟 HLCparameters是自定义键值对集合。get_json(key)可单独取出某个键的 JSON 表示。示例运行后会在控制台打印Metadata JSON: ...以便核对。系统依赖Apache Arrow C 19.0.1本示例的唯一硬性系统依赖是Apache Arrow C 库版本 19.0.1 或更高。这是arrow::Array、arrow::ExportArray、arrow::ImportArray以及arrow/c/bridge.h中 C Data Interface 桥接 API 的来源。Ubuntu 安装sudo apt-get update sudo apt-get install -y -V ca-certificates lsb-release wget wget https://apache.jfrog.io/artifactory/arrow/$(lsb_release --id --short | tr A-Z a-z)/apache-arrow-apt-source-latest-$(lsb_release --codename --short).deb sudo apt-get install -y -V ./apache-arrow-apt-source-latest-$(lsb_release --codename --short).deb sudo apt-get update sudo apt-get install -y -V libarrow-dev libarrow-glib-dev第一条wget命令动态解析发行版 ID 与代号如ubuntu/noble自动下载对应的 Arrow APT 源安装包随后安装libarrow-devC 核心库与libarrow-glib-devGLib 绑定示例本身并不直接依赖安装它可满足部分系统工具的传递依赖。macOS 安装brew update brew install apache-arrow安装完成后请确认pkg-config能解析arrow包见下文自动构建流程必要时将 Arrow 的lib/pkgconfig目录加入PKG_CONFIG_PATH。编译与运行一键自动构建示例提供了自动化构建入口 run.rs它会完成全部构建步骤并启动 dataflow。在仓库根目录执行cargo run --example cxx-arrow-dataflowrun.rs 的自动构建流程探测 Arrow 配置通过pkg-config --cflags arrow与pkg-config --libs arrow获取编译与链接参数run.rs#L65-L91。若pkg-config找不到arrow会给出明确报错提示检查安装与PKG_CONFIG_PATH构建 C 绑定执行cargo build --package dora-node-api-cxx在target/cxxbridge/dora-node-api-cxx/install下生成dora-node-api.h头文件与dora-node-api.cc桥接实现run.rs#L32-L36编译 C 节点用clangC20 标准把node-rust-api/main.cc与生成的dora-node-api.cc一起编译链接dora_node_api_cxx静态库、Arrow 库及平台系统库输出到build/node_rust_apirun.rs#L38-L53。Linux 上会附加-lm -lrt -ldl -lz -pthread等链接参数运行 dataflow调用RunCommand以dataflow.yml启动run.rs#L58-L60。stop_after被设为 120 秒——这是针对节点卡死场景的兜底超时正常运行时节点 10 次迭代后自行退出不会触发。手动构建参考若希望手动编译可参考 examples/c-dataflow/README.md 中记录的cxx-dataflow示例构建步骤先cargo build -p dora-node-api-cxx生成头文件与.cc桥接源文件再使用clang将main.cc与dora-node-api.cc一并编译链接-stdc20最后用构建出的doraCLI 执行cargo build -p dora-cli --release ../../target/release/dora run dataflow.yml注意当前c-arrow-dataflow在 Windows 上存在链接器错误run.rs检测到cfg!(windows)时会直接提示并退出run.rs#L11-L16因此推荐在 Linux 或 macOS 上运行本示例。源码级原理cxx 桥接与 Arrow C Data Interface整个示例的技术底座由 apis/c/node/src/lib.rs 中的#[cxx::bridge]定义。跨语言调用的关键路径如下初始化init_dora_node()在 Rust 侧调用dora_node_api::DoraNode::init_from_env()返回同时持有events事件流与send_output输出发送器的DoraNode句柄lib.rs#L470-L479事件类型DoraEventType枚举在桥接层映射 Rust 的Event变体包括Stop、Input、InputClosed、AllInputsClosed以及故障恢复相关的NodeFailed、NodeRestarted、Reload等lib.rs#L44-L76。示例主循环正是依据这些类型决定退出还是处理输入字节输入限制普通的event_as_input仅支持UInt8原始字节载荷遇到Int32等其他 Arrow 类型会返回错误见 lib.rs#L679-L693 的input_bytes实现。因此跨语言交换类型化数据必须走event_as_arrow_input/event_as_arrow_input_with_info路径发送路径send_arrow_output最终调用node.send_output(output_id, parameters, arrow_array)其中arrow_array由arrow::ffi::from_ffi从 C Data Interface 结构体还原而来lib.rs#L1572-L1587元数据参数则来自传入的Metadata。在 apis/c/node/README.md 中还能找到更完整的 C API 文档包括接收原始字节输入的event_as_input、读取元数据的event_as_input_with_metadata、Service/Action 请求-响应模式send_service_request/recv_service_response/recv_action_result以及 ROS2 桥接dora-ros2-bindings.h的用法是进一步开发 C DORA 节点的重要参考。小结通过c-arrow-dataflow示例可以看到DORA 为 C 节点提供了一条完整的类型化数据通道C 侧用arrow::ExportArray导出、Rust 侧用arrow::ffi导入反之亦然全程基于 Apache Arrow C Data Interface 的稳定 ABI避免了跨语言序列化的拷贝开销。配合Metadata对象的八种参数类型与 JSON 序列化能力C 节点可以像 Rust、Python 节点一样在 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），仅供参考