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)"。

判断基準：各コンポーネントはどこから取得するか？

デフォルト：リーダーがチャンクを生成し、レンズで形を整えます。 変換コードを書く前に この手順を確認してください。「手作業で構築する」という考えは、たいてい誤りです。

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 として届き、レンズが必要です（rerun-mcap を参照）。
3. 既存コンポーネントをその場で修正しますか（解像度の入れ替え、色の変更、単位変換など）？ MutateLens と output_mode="forward_unmatched" を使ってください。
4. 新しいコンポーネント／エンティティを導出しますか（FK → /tf、メッセージからスカラーを取得など）？ DeriveLens を使ってください。1 行を N 行に分散させる場合（関節のバッチ → 関節ごとの /tf）は、 2 つのレンズを組み合わせます。まず output_mode="forward_all" でバッチを導出し （関節状態などの元データを保持します）、次に scatter=True と output_mode="drop_unmatched" を指定した 2 つ目の DeriveLens を使います （分散した行だけを出力します）。robot_data_preprocessing の例を参照してください。
5. リーダーやレンズでは生成できない、本当に必要なサイドカーですか（JSON キャリブレーションオフセット、 手動計測した外部パラメーター、外部メタデータなど）？ Chunk.from_columns + from_iter を使ってください。
6. 最後に LazyChunkStream.merge(...) → .collect(optimize=OptimizationProfile.OBJECT_STORE) → write_rrd(application_id, recording_id) を実行してください。

この順序にする理由：パイプラインを遅延実行、列指向、マルチスレッドのまま保ち、 OBJECT_STORE による最適化を可能にするためです。手書きの行ループやレンズの外での pa.array は、その利点をすべて失わせます。意図的に避けているのはこの方法です。

アンチパターン（代わりにリーダー＋レンズを使用）

左側の処理を書いている場合は、そこで止めて右側の方法を使用してください。

• 行／コンポーネントを構築する for ループ → Selector(...).pipe(...) を使ったレンズ PyArrow の計算コールバック。
• 変換のためにメッセージごとに rr.init と rr.log を呼び出す → これは ライブ ロギングです。 取り込みには、リーダーで読み取り、write_rrd を使用してください。
• chunk.to_record_batch() と pc.filter を使い、その後 Chunk.from_columns で再構築する （手作業で行を間引く）→ stream.drop(content=...)、 .split(...)、またはフィルタ済みの pa.array を返す MutateLens を使用してください。
• レンズの外部で pa.array／pa.RecordBatch／np.frombuffer を組み立てる → 変換処理を MutateLens／DeriveLens のセレクターコールバック内に移してください。
• カスタムパーサーから手作業で組み立てた rr.send_columns → 対応する リーダーを使用してください。チャンクが直接生成されます。
• Rerun 以外のライブラリで MCAP／URDF を解析してから再度ログに記録する → McapReader／UrdfTree を使用してください。
• リーダーがすでにデコードするデータに Chunk.from_columns を使う （Pinhole の内部パラメーター、VideoStream、transforms トピックからの Transform3D）→ リーダーストリーム内で処理し、必要に応じて MutateLens で修正してください。

pyarrow.compute の「属性不足」型エラーが大量に発生する場合（pc.filter、 pc.list_element など）、通常は pc.* の呼び出しが Selector.pipe のレンズコールバックではなく、モジュールレベルのヘルパー内にあることを意味します。チェッカーを抑制する前にレンズ内にリファクタリングしてください。これらのエラーは、手作業での構築がそもそも不要であることを示しています。

既存のコンバーターを移植していますか？ 手作業で構築されたコンバーターはデコーダーの改善より前のものであり、正解の基準ではありません。デコーダーの出力を再確認し（手順 2）、コピーする前に、すべての Chunk.from_columns／for ループをこのリストに照らして確認してください。

コアモデル

• LazyChunkStream は遅延パイプライン DAG であり、コレクションではありません。フィルター、レンズ、マップ、分割、マージを構築しても、ソースデータは読み込まれません。
• 実行は終端呼び出し（write_rrd(...)、collect()、to_chunks()、またはストリームの反復処理）で開始されます。
• 実行はストリーミング方式で、マルチスレッドかつほとんどが GIL の影響を受けません。Python の行ループよりも、ストリーム／レンズ操作と PyArrow の計算を優先してください。
• ムーブセマンティクス：ビルダー呼び出し（filter、drop、lenses、map、 flat_map）は入力ストリームを消費します。消費済みのストリームを再利用するとエラーになります。各ステップの後に再代入してください。終端呼び出しはストリームを消費しませんが、呼び出すたびにパイプライン全体が再実行されます。コストが高い場合は、collect() を一度だけ呼び出してください。
• ChunkStore はメモリ上に実体化されます（stream.collect()、 ChunkStore.from_chunks）。LazyStore はマニフェストでインデックス化され、必要に応じてチャンクを読み込みます（RrdReader(path).store()、カタログのセグメントストア）。どちらも schema()、summary()、stream()、write_rrd(...) を備えています。

ストリームの合成

from rerun.chunk import Chunk, LazyChunkStream, OptimizationProfile

• stream.filter(content=, has_timeline=, is_static=, components=) は、各チャンクのうち条件に一致する部分を保持します。stream.drop(...) はその補集合で、同じキーワードフィルターを使用します。content にはエンティティパスの glob またはそのリストを指定します。
• stream.map(fn) は Chunk -> Chunk を適用し、stream.flat_map(fn) は Chunk -> Iterable[Chunk] を適用します。チャンク単位の Python 処理に使う回避手段ですが、列指向の処理にはレンズを優先してください。
• stream.split(content=, ...) は (matching, non_matching) を返します。両方の分岐で上流を共有します。
• LazyChunkStream.merge(*streams) は任意の数のソースを集約します。
• LazyChunkStream.from_iter(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)

手作業で作成したチャンク — サイドカー専用

Chunk.from_columns を使うのは、リーダーやレンズでは出力できないデータ（JSON／CSV のキャリブレーション、フレームオフセット、外部メタデータ）に限ってください。リーダー（McapReader／UrdfTree／ParquetReader／Hdf5Reader）でトピックをデコードできる場合、またはレンズで導出できる場合は、それが標準的な方法です。ここで手作業で組み立てないでください。 robot_data_preprocessing の例で手作業で作成するチャンクは、JSON オフセットのサイドカーだけです。カメラの修正、FK→/tf、メッシュ、色の再設定はすべてリーダーとレンズで行います。

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] が必要になります。この抑制が切り替えの合図です。

レンズ

レンズは、行を反復せずにコンポーネントの形状を変更したり、修正したり、導出したりします。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 にすると、入力の 1 行を N 行の出力（リスト要素ごとに 1 行）に展開します。
• 同じコンポーネント名が複数のエンティティに存在する場合は、content= で対象を絞り込みます。

出力モード。デフォルトは drop_unmatched です。

• drop_unmatched（デフォルト）：レンズの出力だけが残ります。導出のみを行う中間ストリームに適しています。広範囲に適用すると、ほかのデータはすべて暗黙に破棄されます。
• forward_unmatched：レンズの出力に、どのレンズでも消費されなかった元のコンポーネントを加えたものです。ストリームの残りを維持しながら特定箇所を修正するのに適しています。
• forward_all：レンズの出力に、DeriveLens の入力も含め、元のコンポーネントをすべて加えたものです。mutate レンズでは 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 の compute を使用し、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("<query>") は jq 形式で、ネストされた Arrow データをたどります。

• . は現在の値、.field は struct フィールドです。
• [] はリスト要素を反復し、[N] はリストのインデックスを指定します。
• ? はエラーを抑制して欠落したオプショナル値をスキップし、! は非 null を保証します。
• | は式から次の式へ値を渡します。

.pipe(fn) は Python／PyArrow の変換処理（または別の Selector）をつなげます。 .execute(array) は式を即時実行します。.execute_per_row(array) は出力行数が入力と一致することを保証します（行の対応関係を維持する必要があるレンズのコールバック内で使用します）。

RRD の書き込み

• stream.write_rrd(path, application_id=..., recording_id=...) は、ストリーミングによる 1 回のパスで実行と書き込みを行います。
• stream.collect(optimize=OptimizationProfile.OBJECT_STORE).write_rrd(...) は、データをメモリに展開してチャンクのレイアウトを最適化した後、書き込みます。メモリ使用量は、メモリに展開したチャンクに応じて増加します。
• プロファイル：OBJECT_STORE（チャンクが大きく、ストレージ／クエリ／カタログ向け）と LIVE（チャンクが小さく、ビューアーの低レイテンシー向け）。
• 複数の物理 RRD は、同じ recording_id を共有すると、1 つの論理 Recording を構成します。これを使って、ベースデータ、モデル／URDF データ、レイヤーを分けてください。

RRD を Rerun カタログまたは Hub に送る場合は、明示的に別の指定がない限り、必ず OptimizationProfile.OBJECT_STORE を使用してください。

Chunk API と logging API の比較

• Logging（rr.log、rr.send_columns、RecordingStream）はユーザーコードからのライブロギング向けです。チャンク処理は、既存の記録データの取り込み、変換、後処理向けです。
• Logging → チャンク：RRD を書き込み、RrdReader で読み戻します。RrdReader(path) は recordings() / blueprints()（それぞれ kind、application_id、recording_id を持つ StoreEntry）を列挙します。順次処理には .stream(store=entry)、インデックスアクセスには .store(store=entry) を使います。
• チャンク → logging：rerun.send_chunks(chunks, recording=...) は、Chunk、LazyChunkStream、LazyStore、ChunkStore、またはチャンクの任意のイテラブルを受け取ります。ソースストアの application_id / recording_id は引き継がれません。アクティブな記録データの ID が優先されます。

よくある注意点

• デフォルトのレンズの output_mode は drop_unmatched です。対象を絞った修正で forward_unmatched の設定を忘れると、ストリームの残りが警告なく破棄されます。
• 消費済みの LazyChunkStream を再利用しないでください。意図的に再割り当てするか、split してください。
• レンズは content= でスコープを指定してください。同じコンポーネント名が複数のエンティティに存在することはよくあります。
• MutateLens の変換では、Arrow 配列の型と長さを保持してください。
• カタログレイヤーでは、レイヤーの recording_id がセグメント ID と一致している必要があります。
• これは rerun.chunk です。アップグレード時にはシグネチャを確認してください。

参考資料

• エンドツーエンドの例（MCAP + URDF + JSON サイドカー、レンズ、マージ、最適化）： 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