Rerun Lenses 实战指南：在 Rust 中转换、过滤与重塑数据流

Rerun Lenses 实战指南在 Rust 中转换、过滤与重塑数据流【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerunLenses 是 Rerun 提供的一套数据流变换 API用于在日志消息到达底层 sink 之前对其进行提取、转换和重构如过滤、重命名、拆分嵌套结构、构造时间轴。本指南以仓库中的 Rust 示例 examples/rust/lenses 为主体结合 re_lenses 与 re_lenses_core 的源码实现帮助你理解 Lens 的构建方式、Selector 查询语法以及如何通过LensesSink将变换后的数据送入 Rerun 查看器从而实现对多模态机器人数据的按需整形。快速运行示例示例仓库位于examples/rust/lenses它演示了如何使用 lenses 在把日志消息转发给底层 sink 之前对其进行转换。在仓库根目录直接运行cargo run -p lenses该示例包以 workspace 成员方式引用顶层的reruncrate启用了web_viewer、clap、log_setup特性完整依赖声明见 Cargo.toml[dependencies] rerun { path ../../../crates/top/rerun, features [web_viewer, clap, log_setup] } anyhow.workspace true arrow.workspace true运行后示例会依次记录三类数据指令文本、嵌套标量结构、时间戳列并通过一组 Lens 在数据到达 sink 之前完成转换最终在 Rerun 查看器中以转换后的形态呈现。理解 Lens 的核心概念re_lensescrate 对 Lens 的定义是Lenses are an API for extracting, transforming, and restructuring component data.用于提取、转换和重构组件数据的 API它们作用于包含目标组件的 chunkArrow 数据块之上。从源码结构看一个Lens在内部有三种形态LensInner枚举Mutate原地修改输入组件in-placeDeriveSameEntity从输入组件派生新列输出到同一实体DeriveSeparateEntity从输入组件派生新列输出到不同实体由output_entity指定目标路径。无论哪种形态Lens 都工作在 chunk 内的组件列上。由于进入 chunk 的内容是非确定性的、依赖于 batcher 的批处理行为Lens 的设计不假设行与行之间的值有可预测的关系——这保证了变换在流式数据上的稳健性。在 Rust 代码中构造 Lens 通常从三个入口开始Lens::derive(input)返回DeriveLensBuilder用于派生新组件列 / 时间列Lens::scatter(input)返回开启 1:N 行映射scatter/explode的派生 builderLens::mutate(input, selector)返回MutateLensBuilder用于原地修改组件。逐段解析示例四类典型变换示例主程序 构建了四条 Lens覆盖了过滤、重命名、拆分结构与时间轴提取四类典型用法。1. 提取并重命名把指令文本转成 TextDocumentlet instruction Lens::derive(example:Instruction:text) .to_component(TextDocument::descriptor_text(), Selector::parse(.)?) .build()?;输入组件是example:Instruction:textSelector::parse(.)选中整列identity 查询通过to_component把该列输出为TextDocument的标准文本组件TextDocument::descriptor_text()。这一条演示了“重命名 / 映射到标准组件”第三方或自定义组件通过 Lens 变成 Rerun 原生组件下游查看器即可直接渲染。2. 拆分嵌套结构把 Struct 字段映射到不同实体let destructure_a Lens::derive(example:Nested:payload) .output_entity(nested/a) .to_component_with_cast( Scalars::descriptor_scalars(), Selector::parse(.a)?, CastTo::Auto, ) .build()?; let destructure_b Lens::derive(example:Nested:payload) .output_entity(nested/b) .to_component(Scalars::descriptor_scalars(), Selector::parse(.b)?) .build()?;输入example:Nested:payload是一个 ArrowStructArray含字段af32与bf64两条 Lens 分别用Selector::parse(.a)/Selector::parse(.b)选中 struct 内的子字段output_entity(nested/a)把输出写到独立实体从而把一个实体上的嵌套结构拆解为nested/a、nested/b两个实体上的标量列第一条使用to_component_with_cast(..., CastTo::Auto)CastTo::Auto会通过组件描述符反射查找目标组件的规范 Arrow 类型并自动做类型转换此处把 f32 转为Scalars的规范类型第二条输入类型已匹配直接使用to_component。3. 时间轴提取从数据列构造 timeline 与值列let time Lens::derive(my_timestamp) .to_timeline( my_timeline, rerun::time::TimeType::Sequence, Selector::parse(.)?, ) .to_component(ComponentDescriptor::partial(value), Selector::parse(.)?) .build()?;to_timeline从输入列提取数据创建一条名为my_timeline的新时间轴时间类型为TimeType::Sequence序列时间轴同时to_component把同一列输出为组件value注意一个DeriveLensBuilder可以同时声明多个to_component与to_timeline输出它们共享同一个输入组件。4. 组装 Lens 集合与输出模式let lenses Lenses::new(OutputMode::DropUnmatched) .add_lens(instruction) .add_lens(destructure_a) .add_lens(destructure_b) .add_lens(time);Lenses::new(mode)创建一个 Lens 集合OutputMode有三种取值是整个集合的全局行为开关模式行为ForwardAll转发所有原始组件同时附加 Lens 产生的输出ForwardUnmatched转发未被任何 Lens 消费的原始组件同时附加 Lens 产生的输出DropUnmatched只转发 Lens 产生的输出丢弃其余所有组件示例选择DropUnmatched即“过滤”语义未经 Lens 处理的组件会被丢弃最终进入 sink 的只有被四条 Lens 显式转换过的数据。此外还有add_lens_with_filter(filter, lens)可给每条 Lens 附加实体路径过滤器见 ast.rs 中的Lenses::add_lens_with_filter。把 Lens 挂到日志链路上LensesSinkLens 集合本身不会自动生效需要包装进一个 sinklet lenses_sink LensesSink::new(GrpcSink::default(), lenses); let rec rerun::RecordingStreamBuilder::new(rerun_example_lenses).spawn()?; rec.set_sink(Box::new(lenses_sink));LensesSink位于 re_sdk/src/lenses/sink.rs是实现了LogSinktrait 的转换器先对LogMsg中的 chunk 应用 Lenses再把结果转发给底层的LogSink此处是默认的GrpcSink即通过 gRPC 发给本地/远程 viewer。它的关键行为只有与 Lens 匹配的组件会被转发LensesSink::new(sink, lenses)默认采用“尽力而为”策略某条 Lens 出错时仍尽量产出 chunk调用.strict(true)可开启严格模式Lens 出错时不再发送部分 chunk非 Arrow 消息如SetStoreInfo、BlueprintActivationCommand直接透传不做转换。在LogSink::send的实现中ArrowMsg会被还原为Chunk然后通过re_lenses::default_runtime()提供的运行时执行lenses.apply(...)逐条处理转换结果。这也是把 Lens 与整个 Rerun 数据管道re_sdk 的 lenses 模块连接起来的标准做法。记录数据三种输入形态示例程序最后记录了三种不同形态的数据用于喂给上面的 Lens指令文本自定义 DynamicArchetyperec.set_time(tick, TimeCell::from_sequence(1)); rec.log( instructions, DynamicArchetype::new(example:Instruction).with_component_from_data( text, Arc::new(arrow::array::StringArray::from(vec![ This is a nice instruction text., ])), ), )?;用DynamicArchetype动态构造自定义原型example:Instruction并直接写入字符串数组作为组件text。嵌套标量结构StructArraylet struct_array StructArray::from(vec![ (Arc::new(Field::new(a, DataType::Float32, false)), Arc::new(a) as Arcdyn Array), (Arc::new(Field::new(b, DataType::Float64, false)), Arc::new(b) as Arcdyn Array), ]); rec.log(nested, DynamicArchetype::new(example:Nested) .with_component_from_data(payload, Arc::new(struct_array)))?;在0..10的循环里每次构建一个含af32、bf64两个字段的StructArray作为组件payload记录到nested实体——这正是上面两条拆分 Lens 的输入。时间戳列send_columnsrec.send_columns( timestamped, [], [ SerializedComponentColumn { descriptor: rerun::ComponentDescriptor::partial(my_timestamp), list_array: timestamp_list_builder.finish(), }, SerializedComponentColumn { descriptor: rerun::ComponentDescriptor::partial(value), list_array: string_list_builder.finish(), }, ], )?;用send_columns一次性发送两列my_timestampi64 列表与value字符串列表供time这条 Lens 消费。深入底层Selector 查询语法Lens 的灵活性很大程度上来自Selector——一种在 Arrow 数组上执行的、jq 风格的列式查询语言。其完整语法与语义定义在 selector/mod.rs语法含义示例.field访问 struct 中的命名字段.location[]遍历列表中的每个元素.poses[][N]按下标索引列表.[0]?错误抑制 / 可选操作符.field?!断言非空把全空行提升为外层 null.field!\|把表达式输出接到另一个表达式.foo \| .barpack(...)将 1:1 路径打包成FixedSizeListpack(.x, .y, .z)与 jq 的关键区别在于列式而非行式操作作用于整条 Arrow 列而非单个 JSON 值不支持过滤器与算术只有路径导航、迭代与内置函数不支持带引号字段名与字符串插值字段名必须是裸标识符字母数字、-、_。此外.poses[].x等价于.poses[] | .x——段与段之间可以省略显式管道。针对 protobuf 数据的空值处理?与!两个操作符主要是为了处理由 protobuf 消息产生的 Arrow 列proto3 的optional字段带有存在性跟踪字段未设置时对应列是null而非默认值而?在字段完全缺失如 schema 演进时可选列被省略时抑制错误!则把内层全为 null 的行提升为外层 null把[null]折叠为null让下游消费者拿到干净的可空语义。pack构造固定大小列表pack(path, path, ...)会把多条路径打包进一个FixedSizeList是构造Position3D这类组件列的规范方式例如pack(.x, .y, .z)。要求每条路径每行恰好产生一个值、数据类型一致可空性采用条目级 AND模型——任一路径为 null 则整个条目为 null因此可空路径必须用!显式标注如pack(.x, .y!, .z)否则pack会报错该要求由 schema 类型驱动与具体批次是否恰好含 null 无关。内置函数与默认运行时Selector 与 Lens 的执行都依赖一个Runtime。re_lenses 的默认运行时 会注册全部内置函数目前包括string_prefix前缀插入。更多字符串变换定义在 op/string.rsstring_prefix(prefix)给每个字符串值加前缀string_prefix_nonempty(prefix)仅给非空字符串加前缀string_suffix(suffix)给每个字符串值加后缀string_suffix_nonempty(suffix)仅给非空字符串加后缀。这些函数返回impl Fn(ArrayRef) - ResultOptionArrayRef, Error Send Sync可通过Selector::pipe链入查询表达式。内置函数测试 给出了一个可直接对照的用例对[[world], [rerun]]执行string_prefix(hello_)得到[[hello_world], [hello_rerun]]。除此之外op/semantic.rs 还提供了一批语义级数组变换用于真实场景binary_to_list_uint8二进制数组转Listu8复用底层字节缓冲区接近零拷贝timespec_to_nanos把含seconds/nanos或sec/nsec字段的时间戳 struct 转为总纳秒数string_to_video_codec把h264/h265/av1/vp8/vp9等字符串映射为 RerunVideoCodec枚举值大小写不敏感rgba_struct_to_uint32把r/g/b/af32 或 f640..1打包为 RGBA u32。语义化 Lens直接对接第三方 schemare_lenses还内置了一批针对主流机器人数据格式的语义 Lenssemantic/mod.rs把第三方 schema 直接转换成 Rerun 组件与原型ros2msg覆盖sensor_msgs/msg/Image、nav_msgs、geometry_msgs等一系列 ROS 2 消息。例如 ros2msg/image.rs 中的image()Lens从sensor_msgs.msg.Image:message派生CoordinateFrameframe_id 加_image_plane后缀、Image/DepthImage的 format 与 buffer 组件foxglove覆盖 Foxglove 的CameraCalibration、CompressedImage、FrameTransforms、PointCloud、PoseInFrame、RawImage、VoxelGrid等消息类型。这些语义 Lens 与示例中的自定义 Lens 使用完全相同的Lens::derive(...).to_component(...).build()构建方式是 Lens 机制在真实多模态机器人数据管线上配合 re_importer 的 mcap 导入流程的重要落点。组装一个可复制的完整示例将上面所有片段组合起来就是一个完整的 Lens 数据流程序骨架use rerun::lenses::{CastTo, Lens, Lenses, LensesSink, OutputMode, Selector}; use rerun::sink::GrpcSink; use rerun::{ComponentDescriptor, DynamicArchetype, RecordingStream, Scalars, TextDocument, TimeCell}; fn main() - anyhow::Result() { re_log::setup_logging(); // 1. 构建各条 Lens重命名、拆分、时间轴提取 let instruction Lens::derive(example:Instruction:text) .to_component(TextDocument::descriptor_text(), Selector::parse(.)?) .build()?; let destructure_a Lens::derive(example:Nested:payload) .output_entity(nested/a) .to_component_with_cast( Scalars::descriptor_scalars(), Selector::parse(.a)?, CastTo::Auto, ) .build()?; let time Lens::derive(my_timestamp) .to_timeline(my_timeline, rerun::time::TimeType::Sequence, Selector::parse(.)?) .to_component(ComponentDescriptor::partial(value), Selector::parse(.)?) .build()?; // 2. 组装 Lens 集合选择过滤模式 let lenses Lenses::new(OutputMode::DropUnmatched) .add_lens(instruction) .add_lens(destructure_a) .add_lens(time); // 3. 包装 sink 并挂到 RecordingStream 上 let lenses_sink LensesSink::new(GrpSink::default(), lenses); let rec rerun::RecordingStreamBuilder::new(my_lenses_app).spawn()?; rec.set_sink(Box::new(lenses_sink)); // 4. 正常记录数据会被 Lens 转换后转发 rec.set_time(tick, TimeCell::from_sequence(1)); rec.log(instructions, DynamicArchetype::new(example:Instruction) .with_component_from_data(text, Arc::new(arrow::array::StringArray::from(vec![hi]))))?; Ok(()) }小结通过本示例可以掌握 Rerun Lenses 的完整使用链路用Lens::derive/Lens::mutate构建变换借助Selectorjq 子集语法精确选中 Arrow 列中的字段用to_component、to_component_with_cast、to_timeline、output_entity定义输出目标用OutputMode控制原始组件的去留把Lenses集合包进LensesSink替换RecordingStream的底层 sink让所有日志在到达 viewer 前先经过变换管道。这种“记录端整形”的方式尤其适合多模态机器人数据第三方 schemaROS 2 / Foxglove导入后可以用语义 Lens 一次性映射为 Rerun 原生组件再配合自定义 Lens 做拆分、重命名与时间轴重建无需改动上游数据采集代码。【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考