rerun-parquet

Rerun parquet 数据摄取

ParquetReader 是一个纯读取器：它将扁平表映射到 Rerun 模型，把原始列转换为按组划分、按时间索引的 struct 和 scalar 组件。列名前缀会变成实体，分组列会变成单个 struct 组件，指定的列会变成时间线。读取器不再组装 archetype——将 struct 字段映射为带类型的 Rerun 组件（Transform3D、Scalars、Points3D）由读取器 .stream() 上的 lens 完成。读取器的全部工作都是配置；先填写 rerun-data-model 映射表，再通过 stream() 表达映射。.stream() 之后的流处理机制见 rerun-chunk-processing。

整张表都是配置，不是代码。 如果你发现自己在从 parquet 构建 Chunk.from_columns，或先用 pandas 处理数据，请停下来——ParquetReader 加 lens 几乎肯定就能实现你的需求。读取器无法表达的内容（逐行实体路由、派生值、单位转换）应放在下游的 lens 中，而不是在 pandas 预处理中完成；保持流水线为列式处理。

API

from rerun.chunk import DeriveLens, IndexColumn
from rerun.experimental import ParquetReader

reader = ParquetReader(table_path)  # a lightweight handle over the file
stream = reader.stream(
    entity_path_prefix="/world",  # prepended to every entity path
    column_grouping="prefix",  # "prefix" | "individual" | "explicit_prefixes"
    delimiter="_",  # split for column_grouping="prefix"
    prefixes=None,  # required for "explicit_prefixes"
    use_structs=True,  # pack grouped columns into one struct component
    static_columns=["robot_type"],  # constant-per-file values, logged static
    index_columns=[IndexColumn.timestamp("timestamp", input_unit="us"), IndexColumn.sequence("frame_index")],
)

构造函数只接受 path；所有加载选项都在 stream() 上设置，并且全部只能作为关键字参数传入。 每次调用 stream() 都是独立的，因此一个读取器可以对同一文件驱动多个配置不同的流。 配置会在 stream() 时根据文件的 schema 验证（缺少索引列或静态列会在此处引发 ValueError，而不是在迭代过程中引发）。 没有 column_rules 关键字参数——带类型组件的组装已移至 lens（见下文）。

读取器输出的内容

读取器会将表转换为多个 chunk，每组对应一个 chunk，然后保留数据为通用的 struct/scalar 组件，供 lens 映射。命名是流水线其余部分据以处理数据的关键：

• 分组的多列前缀 X → 实体 /X，包含一个名为 data 的 struct 组件。struct 的字段是去掉前缀（以及 "prefix" 情况下的分隔符）后的列名。因此，前缀为 A 时，A_pos_x、A_quat_w 会作为字段 pos_x、quat_w 放入实体 /A 的 struct data 中。

• 未分组的单列 → 以列名命名的独立实体，以及以列名命名的原始组件——不是 data struct。因此，speed 列会变成实体 /speed、组件 speed。

• 根据 parquet 文件的 schema 元数据构建的 /__properties 元数据 chunk。通常会在 .stream() 后立即将其丢弃：

  stream = reader.stream().drop(content="/__properties/**")
  

列分组：哪些列共享一个实体

• "prefix"（默认）：按 delimiter 拆分每个列名，并按第一段分组。gripper_pos_x、gripper_pos_y → 实体 /gripper，struct data{pos_x, pos_y}。
• "explicit_prefixes"：按 prefixes 中的精确字符串分组，按长度从长到短依次尝试；前缀会从每个 struct 字段名中移除（直接匹配字符串，不使用分隔符——foo + a → 字段 a）。不匹配任何前缀的列会单独分组。名称中分隔符含义不明确时使用此选项（observation.state 与 observation.images.top：传入完整前缀）。
• "individual"：每列都作为一个独立的 chunk/实体，包含一个以列名命名的原始组件——完全不打包成 struct，即使列共享前缀也如此。此时会忽略 use_structs。这通常不是你想要的模型；仅在调试时将其作为基线使用。

use_structs=True（默认）会将一组列打包成一个 Arrow struct 组件（data 字段），适用于 "prefix"/"explicit_prefixes"；False 则会为每列输出一个组件（打包成 struct 之前的扁平布局，查询会将其视为独立列）。

时间线：index_columns

每个条目都通过 IndexColumn 构建：IndexColumn.timestamp(name, input_unit=...)（自纪元起算）、IndexColumn.duration(name, input_unit=...)（已经过时间）或 IndexColumn.sequence(name)（序数整数）。

• input_unit 描述列中原始整数的单位（默认 "ns"， 也可以是 "us"、"ms"、"s"）；Rerun 会在内部将其换算为 ns。序列不需要单位。

如果省略，则会生成一个合成的 row_index 序列时间线。 这几乎从来不是你想要查询或对齐的时间线；请始终指定真实的时间列。如果表同时包含时间戳和序列时间线，请同时标记两者（多速率对齐，参见 rerun-data-model）。

静态列：static_columns

列出的列在所有行中都是常量；它们会作为单个静态（无时间）数据块单独发出，与时序数据分开。流运行时，如果列出的列实际存在变化，则会报错——此错误是数据质量信号，而不是删除静态声明的理由。

通过 lens 转换为类型化组件

reader 的分组输出是通用结构体（data）和标量数据。DeriveLens 会读取该结构体的字段，将它们打包并转换为真正的 Rerun 组件，然后写入输出实体——这正是旧版 column_rules API 的功能，现在改为在流的下游执行。

针对 reader 的结构体组件创建 lens（分组前缀对应 "data"，单独列对应列名），然后添加一个或多个 .to_* 构建器方法。每个构建器都会返回一个新的 lens，因此可以链式调用。

构建器	生成内容	参数顺序
to_translation(x, y, z)	Transform3D:translation	x、y、z
to_quaternion(x, y, z, w)	Transform3D:quaternion	x、y、z、w（xyzw）
to_scale(x, y, z)	Transform3D:scale	x、y、z
to_rotation_axis_angle(axis_x, axis_y, axis_z, angle)	Transform3D:rotation_axis_angle	axis_x、axis_y、axis_z、angle（弧度）
to_scalars(*fields)	Scalars:scalars	一个或多个字段名
to_packed_component(component, *fields)	指定的组件	组件描述符，然后是字段名
to_component(component, selector)	指定的组件	组件描述符，然后是一个 Selector
to_timeline(name, type, selector)	时间线（非组件）	名称、"sequence"/"duration_ns"/"timestamp_ns"、selector

to_packed_component 会按顺序将命名的结构体字段（至少需要一个）打包为组件所需的定长列表，并默认自动将 f64 转换为 f32，以匹配组件类型。to_translation、to_quaternion、to_scale 助手方法是它的便捷封装，因此都会自动转换。to_rotation_axis_angle 会构建一个 Struct{axis, angle}，并在内部将 axis 和 angle 强制转换为 f32。to_scalars 仅有一个字段时，会为每行输出一个普通标量（而非含 1 个元素的列表）；有多个字段时，会在同一实体上为每个字段输出一个标量序列。

使用 .stream().lenses([lens], content="/A", output_mode="drop_unmatched") 应用 lens：

• content 是针对源实体路径的预筛选条件，用于限定 lens 可以处理哪些块。范围外的块会原样透传。将其设为 reader 的分组实体（例如 "/A"）。
• output_mode 决定如何处理范围内但未匹配的块："drop_unmatched"（默认，仅保留 lens 输出）、"forward_unmatched"（输出替换匹配项，其他原始内容保留），或 "forward_all"（输出与所有原始内容均保留）。
• lens 自身的 output_entity= 用于设置目标实体，与控制输入端的 content 无关。

Transform3D 端到端示例。reader 会将 A_* 列分组到 /A 的 data 结构体中；lens 会读取去掉前缀后的字段名（pos_x、quat_w），对其打包并转换，然后将完整的 Transform3D 写入 /pose：

from rerun.chunk import DeriveLens, IndexColumn
from rerun.experimental import ParquetReader

lens = (
    DeriveLens("data", output_entity="/pose")
    .to_translation("pos_x", "pos_y", "pos_z")
    .to_quaternion("quat_x", "quat_y", "quat_z", "quat_w")
)

chunks = (
    ParquetReader(table_path)
    .stream(index_columns=[IndexColumn.sequence("frame_index")])
    .lenses([lens], content="/A", output_mode="drop_unmatched")
    .to_chunks()
)

在一个共用 output_entity 的 lens 上链式调用多个 .to_*，会将多个组件列累积到该实体的同一个 archetype 中——上例中，Transform3D:translation 和 Transform3D:quaternion 都会写入 /pose，组成完整的 Transform3D。对于通用的定长列表组件，将组件描述符传给 to_packed_component：

import rerun as rr
from rerun.chunk import DeriveLens
from rerun.experimental import ParquetReader

lens = DeriveLens("data", output_entity="/points").to_packed_component(
    rr.Points3D.descriptor_positions(), "x", "y", "z"
)

选择器

lens 字段路径使用 Selector，这是一种用于 Arrow 列的类 jq 语法（.field 用于访问结构体字段，[] 用于遍历列表，[N] 用于索引，? 用于忽略缺失字段的错误，! 用于断言非空，| 用于管道操作，pack(.x, .y, .z) 用于将多个路径打包为定长列表）。to_* 助手方法会为你构建这些 selector；需要自定义字段路径时，可使用 to_component(component, Selector(".x"))。字段路径引用的是去掉前缀后的结构体字段名——lens 看到的是 pos_x，而不是 A_pos_x。

注意事项

1. 没有 index_columns → 只有合成的 row_index 时间线。预期存在时间戳时间线的查询将无法找到任何内容。
2. unit 是原始列的单位，而不是期望的输出单位；声明为 "ns" 的微秒列会落在过去 1000 倍的位置。
3. 如果列出的列实际存在变化，static_columns 会引发错误；这个错误是数据质量信号，而不是删除静态声明的理由。是否统一取决于数据，因此只有在数据流运行时才会延迟引发错误；列出的列若不存在，则会在调用 stream() 时立即引发错误。
4. 分组前缀的结构体组件名为 data——这是 DeriveLens 匹配的 input_component 字符串。单独的列或 "individual" 列则是一个以列名命名的原始组件。
5. 选择器字段路径引用的是去掉前缀后的结构体字段名（pos_x，而不是 gripper_pos_x）。
6. 丢弃读取器从 parquet 架构元数据中生成的 /__properties 元数据块：.stream().drop(content="/__properties/**")。
7. to_quaternion 中四元数列的顺序是 x、y、z、w；连接字段前请检查源数据的约定。
8. to_packed_component（以及基于它构建的转换辅助函数）会自动将 f64 转换为 f32，以匹配组件类型；对于 parquet 的双精度列，这通常正是你想要的。
9. 读取器无法表达的内容（逐行实体路由、派生值、单位转换）应放在下游的 lens 中，而不是放在 pandas 预处理里；保持流水线按列处理（rerun-chunk-processing）。

参考资料

• 包含完整文档字符串的 Lens 构建器源码：已安装的 rerun-sdk 包中的 rerun/chunk/_lens.py（to_translation、to_quaternion、to_scale、to_rotation_axis_angle、to_scalars、to_packed_component、to_component、to_timeline）。
• 读取器源码：rerun/experimental/_parquet_reader.py，或运行 python -c "from rerun.experimental import ParquetReader; help(ParquetReader)"
• 权威的完整示例：集成测试 rerun_py/tests/integration/test_parquet_reader.py（分组、索引/静态列，以及 Transform3D / Points3D / Scalars lens 流程）和 rerun_py/tests/integration/test_lazy_chunk_stream.py（应用 lens、content/output_mode、选择器）。
• rerun-lerobot — LeRobot 数据集将片段存储为 parquet；该技能涵盖内置导入器路径与使用此读取器直接读取 parquet 的方式。
• rerun-data-model（映射决策）、rerun-chunk-processing（.stream() 之后的流机制）