rerun-parquet

Rerun parquet の取り込み

ParquetReader は純粋なリーダーです。フラットなテーブルを Rerun モデルに対応付け、元の列をグループ化された時間インデックス付きの struct コンポーネントと scalar コンポーネントのチャンクに変換します。列名のプレフィックスはエンティティになり、グループ化された列は単一の struct コンポーネントになり、指定された列はタイムラインになります。リーダーは、アーキタイプを組み立てなくなりました。struct のフィールドを型付きの Rerun コンポーネント（Transform3D、Scalars、Points3D）に対応付ける処理は、リーダーの .stream() に対するレンズで行います。リーダーの処理はすべて設定です。まず rerun-data-model の対応付けテーブルを埋め、その内容を stream() で表現してください。.stream() の後のストリーム処理については rerun-chunk-processing を参照してください。

テーブル全体が設定であり、コードではありません。 parquet から Chunk.from_columns を作成したり、先に pandas で加工したりしているなら、いったん止めてください。ほとんどの場合、ParquetReader とレンズで表現できます。リーダーで表現できない処理（行ごとのエンティティの振り分け、派生値、単位変換）は、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() の呼び出しはそれぞれ独立しているため、1 つのリーダーから同じファイルに対して、異なる設定の複数のストリームを生成できます。 設定は stream() の呼び出し時にファイルのスキーマに照らして検証されます（インデックス列や静的列がない場合は、反復処理の途中ではなく、そこで ValueError が発生します）。 column_rules キーワード引数はありません。型付きコンポーネントの組み立てはレンズに移りました（以下を参照）。

リーダーが出力する内容

リーダーはテーブルをチャンクに変換し、グループごとに 1 つのチャンクを生成した後、レンズで対応付けられるよう、データを汎用の struct／scalar コンポーネントとして保持します。パイプラインの残りの部分が参照する重要な要素は名前付けです。

• グループ化された複数列のプレフィックス X → エンティティ /X。単一の struct コンポーネント data を持ちます。struct のフィールド名は、プレフィックス（および "prefix" の場合は区切り文字）を取り除いた列名になります。そのため、プレフィックス A の A_pos_x、A_quat_w は、エンティティ /A 上のフィールド pos_x、quat_w を持つ struct data になります。

• グループに属さない単独の列 → 列名を名前とする独自のエンティティと、列名を名前とする生のコンポーネントになります。data struct にはなりません。たとえば、speed 列はエンティティ /speed、コンポーネント speed になります。

• parquet ファイルのスキーマメタデータから作成される /__properties メタデータチャンク。通常は .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"：各列がそれぞれ独自のチャンク／エンティティになり、列名を名前とする生のコンポーネントを持ちます。プレフィックスが共通する列も含め、struct へのパッキングは一切行いません。この場合、use_structs は無視されます。目的のモデルに合うことはまれで、デバッグ用のベースラインとしてのみ使用してください。

use_structs=True（デフォルト）の場合、"prefix"／"explicit_prefixes" ではグループ内の列が単一の Arrow struct コンポーネント（data フィールド）にパッキングされます。False の場合は列ごとに 1 つのコンポーネントが出力されます（struct 化前のフラットなレイアウトで、クエリからは列が別々に見えます）。

タイムライン：index_columns

各エントリは IndexColumn を使って作成します。IndexColumn.timestamp(name, input_unit=...)（エポックからの時刻）、IndexColumn.duration(name, input_unit=...)（経過時間）、または IndexColumn.sequence(name)（序数の整数）を指定します。

• input_unit は、列の生の整数が何を表すかを示します（デフォルトは "ns"、ほかに "us"、"ms"、"s"）。Rerun は内部でナノ秒に換算します。シーケンスには単位を指定しません。

省略した場合、合成された row_index シーケンスのタイムラインが生成されます。 これは、クエリや整列の基準として使いたいタイムラインであることはほとんどありません。必ず実際の時刻列を指定してください。テーブルに時刻とシーケンスの両方がある場合は、両方のタイムラインを記録してください（複数レートの整列については rerun-data-model を参照）。

静的列：static_columns

ここに列挙した列は、すべての行で一定です。時間データとは別に、単一の静的（時間のない）チャンクとして一度だけ出力されます。列挙した列の値が実際には変化する場合、ストリームの実行時にエラーが発生します。このエラーはデータ品質のシグナルであり、静的列の宣言を削除する理由ではありません。

レンズによる型付きコンポーネント

リーダーのグループ化された出力は、汎用の構造体（data）とスカラー データです。DeriveLens はその構造体のフィールドを読み取り、実際の Rerun コンポーネントにパックしてキャストし、出力エンティティに書き込みます。これは従来の column_rules API が行っていた処理を、ストリームの下流で行うものです。

リーダーの構造体コンポーネント（グループ化されたプレフィックスの場合は "data"、単独または個別の列の場合は列名）を対象にレンズを作成し、1 つ以上の .to_* ビルダーメソッドを追加します。各ビルダーは新しいレンズを返すため、連結できます。

ビルダー	生成されるもの	引数の順序
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	1 つ以上のフィールド名
to_packed_component(component, *fields)	指定されたコンポーネント	ディスクリプター、続いてフィールド名
to_component(component, selector)	指定されたコンポーネント	ディスクリプター、続いて Selector
to_timeline(name, type, selector)	タイムライン（コンポーネントではありません）	name、"sequence"／"duration_ns"／"timestamp_ns"、selector

to_packed_component は、名前付きの構造体フィールド（順序どおり、最低 1 つ必須）を、コンポーネントが想定する固定サイズリストにパックします。また、コンポーネントの型に合わせるため、デフォルトで f64→f32 に自動キャストします。to_translation、to_quaternion、to_scale ヘルパーはこのメソッドを使いやすくしたラッパーであり、すべて自動キャストします。to_rotation_axis_angle は Struct{axis, angle} を作成し、内部で axis と angle を f32 にハードキャストします。to_scalars にフィールドを 1 つ指定すると、行ごとに通常のスカラー（1 要素のリストではありません）を出力します。複数のフィールドを指定すると、同じエンティティに各フィールドのスカラー系列を 1 つずつ出力します。

.stream().lenses([lens], content="/A", output_mode="drop_unmatched") でレンズを適用します。

• content は、ソースエンティティパスに対する事前フィルターです。レンズが処理できるチャンクの範囲を指定します。範囲外のチャンクは変更されずに通過します。リーダーのグループ化されたエンティティ（例："/A"）を設定してください。
• output_mode は、範囲内にあるものの一致しないチャンクの扱いを決定します。"drop_unmatched"（デフォルト。一致したレンズ出力のみを保持）、"forward_unmatched"（出力が一致したものを置き換え、その他の元データは保持）、または "forward_all"（出力とすべての元データを出力）を指定できます。
• レンズ自身の output_entity= は、content とは独立して、出力先エンティティを設定します。content は入力側を制御します。

Transform3D のエンドツーエンドの例です。リーダーは A_* 列を /A の data 構造体にグループ化します。レンズはプレフィックスを除いたフィールド名（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()
)

1 つのレンズで共通の output_entity を指定して複数の .to_* を連結すると、そのエンティティの同じアーキタイプに複数のコンポーネント列が蓄積されます。上の例では、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"
)

セレクター

レンズのフィールドパスには Selector を使用します。これは Arrow 列に対する jq 風の文法で、構造体フィールドにアクセスする .field、リストを反復する []、インデックスを指定する [N]、フィールドが存在しない場合のエラーを抑制する ?、非 null を保証する !、パイプを使う |、パスを結合して固定サイズリストにする pack(.x, .y, .z) を使用できます。to_* ヘルパーはこれらのセレクターを作成します。カスタムのフィールドパスが必要な場合は、to_component(component, Selector(".x")) を使用してください。フィールドパスでは、プレフィックスを除いた構造体フィールド名を参照します。レンズから見えるのは A_pos_x ではなく pos_x です。

注意点

1. index_columns がない場合、合成された row_index のタイムラインのみが使われます。タイムスタンプのタイムラインを前提とするクエリでは、何も見つかりません。
2. unit は目的の出力単位ではなく、元の列の単位です。マイクロ秒の列を "ns" として宣言すると、1000 倍過去の時刻になります。
3. 指定した列の値が実際には変化する場合、static_columns はエラーを発生させます。このエラーはデータ品質を示すシグナルであり、静的であるという宣言を削除する理由ではありません。一様性はデータに依存するため、ストリームの実行時に遅延して検証されます。指定した列が存在しない場合は、stream() の呼び出し時に即座にエラーが発生します。
4. グループ化されたプレフィックスの struct コンポーネント名は data です。これは DeriveLens が照合する input_component 文字列です。単独の列または "individual" 列は、列名を付けた raw コンポーネントになります。
5. セレクターのフィールドパスでは、プレフィックスを除いた struct フィールド名（gripper_pos_x ではなく pos_x）を指定します。
6. リーダーが Parquet スキーマのメタデータから出力する /__properties メタデータチャンクを削除してください。.stream().drop(content="/__properties/**") とします。
7. to_quaternion のクォータニオンの列順は x、y、z、w です。フィールドを接続する前に、元データの規約を確認してください。
8. to_packed_component と、それを利用する変換ヘルパーは、コンポーネントの型に合わせて f64 を f32 に自動キャストします。Parquet の double 列では、通常これが望ましい動作です。
9. リーダーで表現できない処理（行ごとのエンティティへのルーティング、派生値、単位変換）は、Pandas で前処理するのではなく、後続のレンズで行います。パイプラインは列指向のままにしてください（rerun-chunk-processing）。

参考資料

• 詳細な docstring を含むレンズビルダーのソース：インストール済みの 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 のレンズフロー）と、rerun_py/tests/integration/test_lazy_chunk_stream.py（レンズの適用、content／output_mode、セレクター）。
• rerun-lerobot — LeRobot データセットはエピソードを Parquet として保存します。このスキルでは、組み込みのインポーターを使う方法と、このリーダーで Parquet を直接読み込む方法を解説します。
• rerun-data-model（マッピングの判断）、rerun-chunk-processing（.stream() の呼び出し後のストリーム処理）