rerun-chunk-processing

Rerun 分块处理

原始数据与 RRD 之间的流水线层：读取器生成 Chunk， 流对其进行转换，终端调用负责执行。本技能仅介绍通用机制。首先确定数据模型（rerun-data-model），然后为每种数据源选择相应的导入器技能：

数据源	读取器	技能
MCAP 文件（ROS2、protobuf、Foxglove）	McapReader(path).stream()	rerun-mcap
URDF 机器人模型（+ 关节状态 → FK）	UrdfTree.from_file_path(...).stream()	rerun-urdf
Parquet 表（轨迹、传感器日志）	ParquetReader(path).stream()	rerun-parquet
mp4 摄像头视频	Mp4Reader(path).stream()	rerun-mp4
HDF5 文件（组 → 实体）	Hdf5Reader(path).stream()	暂无
LeRobot 数据集目录	使用内置导入器，然后调用 RrdReader	rerun-lerobot
现有 RRD	RrdReader(path)	本文下文
附属文件（JSON 标定、元数据）	Chunk.from_columns + from_iter	本文下文

API 位于 rerun.chunk；需要了解具体行为时，请检查已安装的接口： python -c "from rerun.chunk import LazyChunkStream; help(LazyChunkStream)"。

决策规则：每个组件从哪里来？

默认情况下：读取器生成分块；lens 对其进行塑形。 编写任何转换代码前，请按此流程判断——多数“手动构建”的直觉都是错的：

1. 读取器支持该数据源？ 使用读取器的 .stream()；绝不要手动解析后重新记录。MCAP→McapReader，URDF→UrdfTree，parquet→ParquetReader，mp4→Mp4Reader，HDF5→Hdf5Reader，RRD→RrdReader，LeRobot 目录→log_file_from_path。
2. 解码器已能输出该原型组件？ Foxglove 可直接提供 Transform3D、Pinhole、VideoStream（真实样本字节）——直接传递，不要重新推导。只有自定义 protobuf 主题会以 <Name>:message 的形式到达，需要使用 lens（参见 rerun-mcap）。
3. 就地修正现有组件（分辨率颠倒、重新着色、单位转换）？使用 MutateLens，并设置 output_mode="forward_unmatched"。
4. 推导新的组件/实体（FK→/tf，从消息中提取标量）？使用 DeriveLens。若要将一行拆分为 N 行（一个关节批次 → 每个关节一个 /tf），请使用双 lens 组合：先用 output_mode="forward_all" 推导批次（保留原始数据，例如关节状态），再使用第二个 DeriveLens，设置 scatter=True 和 output_mode="drop_unmatched"（仅输出拆分后的行）。参见 robot_data_preprocessing 示例。
5. 确实需要附属数据，且读取器或 lens 都无法生成（JSON 标定偏移、手工测量的外参、外部元数据）？使用 Chunk.from_columns + from_iter。
6. 最后调用 LazyChunkStream.merge(...) → .collect(optimize=OptimizationProfile.OBJECT_STORE) → write_rrd(application_id, recording_id)。

为何按此顺序：流水线能够保持惰性、列式、多线程，并可通过 OBJECT_STORE 优化。手动构建逐行循环或在 lens 之外使用 pa.array 会丢弃所有这些优势——我们刻意避免这种做法。

反模式（改用 reader + lens）

如果你正在编写左侧的内容，请停下来，改用右侧的方法：

• 用 for 循环构建行/组件 → 使用带有 Selector(...).pipe(...) 的 lens PyArrow 计算回调。
• 每条消息都调用 rr.init + rr.log 进行转换 → 这是用于实时日志记录的方式； 对于数据摄取，请使用 reader 读取并调用 write_rrd。
• chunk.to_record_batch() + pc.filter，然后通过 Chunk.from_columns 重建（手动稀疏行）→ 使用 stream.drop(content=...)、.split(...)，或返回经过筛选的 pa.array 的 MutateLens。
• 在 lens 外部组装 pa.array / pa.RecordBatch / np.frombuffer → 将转换移入 MutateLens/DeriveLens 选择器回调。
• 根据自定义解析器手动组装 rr.send_columns → 使用匹配的 reader；它会直接生成 chunks。
• 使用非 Rerun 库解析 MCAP/URDF，然后重新记录 → 使用 McapReader / UrdfTree。
• 对 reader 已解码的数据使用 Chunk.from_columns（Pinhole 内参、VideoStream、来自 transforms 主题的 Transform3D）→ 将其保留在 reader 流中；如有需要，使用 MutateLens 修复。

如果出现大量 pyarrow.compute“缺少属性”类型错误（pc.filter、 pc.list_element），通常意味着 pc.* 调用位于模块级辅助函数中，而非 Selector.pipe lens 回调内。请先重构为 lens，再抑制检查器——这些错误表明手动构建本就不应该存在。

正在移植现有转换器？ 手动构建的转换器早于解码器改进，因此不能作为事实依据。重新验证解码器输出（步骤 2），并在复制前检查所有 Chunk.from_columns / for 循环是否符合此列表。

核心模型

• LazyChunkStream 是一个惰性流水线 DAG，而非集合。构建筛选器、lens、映射、拆分和合并操作不会读取源数据。
• 执行从终端调用开始：write_rrd(...)、collect()、 to_chunks()，或遍历流。
• 执行采用流式、多线程方式，且基本不受 GIL 限制。优先使用 stream/lens 操作和 PyArrow 计算，而非 Python 行循环。
• 移动语义：构建器调用（filter、drop、lenses、map、 flat_map）会消耗输入流；重用已消耗的流会报错。每一步之后都要重新赋值。终端调用不会消耗流，但每次终端调用都会重新执行整个流水线；如果代价太高，请只调用一次 collect()。
• ChunkStore 会在内存中物化（stream.collect()、 ChunkStore.from_chunks）。LazyStore 由清单索引，按需加载 chunks （RrdReader(path).store()、目录片段存储）。两者都有 schema()、summary()、stream() 和 write_rrd(...)。

流组合

from rerun.chunk import Chunk, LazyChunkStream, OptimizationProfile

• stream.filter(content=, has_timeline=, is_static=, components=) 保留每个 chunk 中匹配的部分；stream.drop(...) 则返回其补集，使用相同的关键字筛选条件。content 接受实体路径 glob 或其列表。
• stream.map(fn) 应用 Chunk -> Chunk；stream.flat_map(fn) 应用 Chunk -> Iterable[Chunk]。这是用于 chunk 级 Python 逻辑的逃生口；列式处理优先使用 lens。
• stream.split(content=, ...) 返回 (matching, non_matching)；两个分支共享同一上游。
• LazyChunkStream.merge(*streams) 汇入任意数量的源。
• LazyChunkStream.from_iter(chunks) 包装手动构建的 chunks。

stream = source_stream()  # any importer skill
stream = stream.drop(content="/video_raw/**")
stream = stream.lenses(fix_lens, content="/cam/**", output_mode="forward_unmatched")
merged = LazyChunkStream.merge(stream, sidecar_stream)
merged.write_rrd(out_path, application_id="my_app", recording_id=recording_id)

手动构建分块——仅用于 sidecar

仅当没有 reader 或 lens 能生成数据时，才使用 Chunk.from_columns——例如 JSON/CSV 校准数据、帧偏移量、外部元数据。如果某个 reader （McapReader/UrdfTree/ParquetReader/Hdf5Reader）可以解码该主题，或某个 lens 可以推导出它，那么这才是惯用做法；不要在这里手动组装。 在 robot_data_preprocessing 示例中，唯一手动构建的分块是 JSON 偏移量 sidecar；相机修正、FK→/tf、网格和重新着色都由 reader 和 lens 完成。

Chunk.from_columns(entity_path, indexes, columns) 与 rr.send_columns(...) 相对应，并接受相同的 archetype .columns(...) 辅助方法。 indexes 为空表示静态数据。

chunk = Chunk.from_columns(
    "/tf_static/robot_offsets",
    indexes=[],  # static
    columns=rr.Transform3D.columns(
        translation=translations,
        quaternion=quaternions_xyzw,
        parent_frame=parents,
        child_frame=children,
    ),
)
sidecar_stream = LazyChunkStream.from_iter([chunk])

rr.AnyValues.columns(...) 涵盖非标准元数据字段。为便于检查，Chunk 提供 entity_path、num_rows、is_static、 timeline_names、to_record_batch() 和 format()（人类可读表格）。

Recording 属性

Recording 属性是唯一的例外：不要手动构建。 Chunk.from_property(name, values) 与 rr.send_property 相对应，并在 /__properties/<name> 上生成静态分块。

这样做：

chunk = Chunk.from_property(
    "episode",
    rr.AnyValues(task="pick", duration_sec=12.4),
)

而不是手动组装相同的分块：

Chunk.from_columns(
    f"/__properties/{name}",
    indexes=[],
    columns=rr.AnyValues.columns(**{name: [value]}),  # type: ignore[arg-type]
)

• /__properties/ 是 Rerun 的内部布局；第二种写法将其硬编码了。
• 动态字段名称会导致 AnyValues.columns() 需要添加 # type: ignore[arg-type]。这种抑制提示你应该改用其他方法。

Lenses

Lenses 无需遍历行即可重塑、修正或推导组件。使用 stream.lenses(lenses, output_mode=..., content=...) 应用。

• MutateLens(component, selector, keep_row_ids=False) 原地修改现有组件。
• DeriveLens(component, output_entity=None, scatter=False) 创建新列，也可以将其放到其他实体中。对每个输出，链式调用 .to_component(descriptor, selector)；.to_timeline(name, "sequence" | "duration_ns" | "timestamp_ns", selector) 从数据本身提取时间列。 scatter=True 会将一条输入行展开为 N 条输出行（每个列表元素对应一条）。
• 如果多个实体下存在同名组件，请使用 content= 指定范围。

输出模式，默认值为 drop_unmatched：

• drop_unmatched（默认）：仅保留 lens 输出。适用于仅执行推导的中间流；若广泛应用，会静默丢弃其他所有内容。
• forward_unmatched：保留 lens 输出，以及没有被 lens 消耗的原始组件。适用于有针对性的修正，同时保留流的其余部分。
• forward_all：保留 lens 输出和所有原始组件，包括 DeriveLens 的输入。对 mutate lens 而言，与 forward_unmatched 相同。

原地修正（保持 Arrow 类型和长度不变）：

stream = stream.lenses(
    MutateLens(
        "Pinhole:resolution",
        Selector(".").pipe(
            lambda res: pa.array(
                [(h, w) for w, h in res.to_pylist()],
                type=res.type,
            )
        ),
    ),
    content=["/external/cam_low", "/external/cam_high"],
    output_mode="forward_unmatched",
)

使用单位换算进行推导（使用 PyArrow 计算，不进行 Python 循环）：

DeriveLens("schemas.proto.JointState:message", output_entity="/joints_deg/waist").to_component(
    rr.Scalars.descriptor_scalars(),
    Selector(".joint_positions").pipe(lambda arr: pc.multiply(pc.list_element(arr, 0), 180.0 / math.pi)),
)

Selector 语法

Selector("<query>") 以 jq 风格遍历嵌套的 Arrow 数据：

• . 当前值；.field 结构体字段
• [] 遍历列表元素；[N] 索引列表
• ? 抑制错误／跳过缺失的可选值；! 断言非空
• | 将一个表达式传入另一个表达式

.pipe(fn) 链式调用 Python/PyArrow 转换（或另一个 Selector）。 .execute(array) 会立即执行；.execute_per_row(array) 可确保输出行数与输入相同（在必须保持行对齐的 lens 回调中使用）。

写入 RRD

• stream.write_rrd(path, application_id=..., recording_id=...) 会在一次流式处理中执行并写入。
• stream.collect(optimize=OptimizationProfile.OBJECT_STORE).write_rrd(...) 会先物化数据、优化分块布局，再写入。内存占用随物化的分块数量增长。
• 配置文件：OBJECT_STORE（大分块，适用于存储／查询／目录）和 LIVE（小分块，适用于低延迟查看器）。
• 多个物理 RRD 只要共享同一个 recording_id，就会构成一个逻辑 recording；可用此方式分离基础数据、模型/URDF 数据和各图层。

如果 RRD 将用于 Rerun catalog 或 Hub，请始终使用 OptimizationProfile.OBJECT_STORE，除非另有明确要求。

分块 API 与日志 API

• 日志记录（rr.log、rr.send_columns、RecordingStream）用于从用户代码实时记录日志；分块处理则用于摄取、转换和后处理已有录制数据。
• 日志记录 → 分块：写入 RRD，再用 RrdReader 读回。RrdReader(path) 会列出 recordings() / blueprints()（每个都是包含 kind、application_id、recording_id 的 StoreEntry）；.stream(store=entry) 用于顺序遍历，.store(store=entry) 用于索引访问。
• 分块 → 日志记录：rerun.send_chunks(chunks, recording=...) 接受 Chunk、LazyChunkStream、LazyStore、ChunkStore 或任意可迭代的分块集合。源存储区的 application_id / recording_id 不会保留；以活动录制数据的标识为准。

常见问题

• 默认镜头的 output_mode 为 drop_unmatched；针对性修复时忘记设置 forward_unmatched，会静默丢弃流中的其余内容。
• 不要重复使用已消费的 LazyChunkStream；请重新赋值或有意调用 split。
• 使用 content= 限定镜头范围；同一个组件名称通常存在于多个实体下。
• 在 MutateLens 转换中保留 Arrow 数组的类型和长度。
• 对于目录层，图层的 recording_id 必须与片段 ID 相同。
• 这是 rerun.chunk；升级时请固定检查签名。

参考资料

• 端到端示例（MCAP + URDF + JSON sidecar、镜头、合并、优化）： https://github.com/rerun-io/rerun/tree/main/examples/python/robot_data_preprocessing
• 文档：https://rerun.io/docs/concepts/logging-and-ingestion/chunk-processing-api、 https://rerun.io/docs/concepts/query-and-transform/lenses