企业官网建设流程全解析

Rerun DataFrame 操作实战：基于 DataFusion 的聚合、Join 与子片段提取

【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun

Rerun 通过 DataFusion 为多模态机器人数据提供了一套标准 Dataframe 接口，使开发者可以用熟悉的join、aggregate、filter、窗口函数等操作直接在 catalog 数据上完成分析。本文以官方示例dataframe_operations为骨架，完整讲解如何搭建本地 catalog server、按实体路径过滤数据、按 episode 聚合时间范围、跨列 Join 找出"迟到"的数据，以及如何利用窗口函数把一段连续录制的机械臂操作切分成独立的子片段——读者学完后可直接在自己的机器人数据流水线中套用这些操作。

背景：Rerun 的 Dataframe 接口与懒执行模型

Dataframe 是现代数据分析工作流的核心抽象。Rerun 通过 DataFusion 为存储在 catalog server 中的数据提供统一的 Dataframe 接口。这意味着你的数据既可以用 Rerun SDK 以可视化的方式浏览，也可以通过 SQL / DataFrame 算子做大规模批量分析。

DataFusion 采用懒执行（lazy execution）模型：上面描述的join、filter、aggregate等操作并不会立即执行，而是先构造一个逻辑执行计划，直到真正需要结果（例如调用count()、.collect()或导出为 pandas/polars/pyarrow）时才触发求值。因此：

• 数据处理阶段应尽量使用 DataFusion 原生的算子，而不是先把数据全量拉回本地再处理；
• DataFusion 同时支持将结果转换为主流工具使用的 DataFrame（pandas、polars、pyarrow），方便后续与既有分析栈衔接。

在 Rerun 仓库中，这一接口的 Rust 侧实现在 re_datafusion crate（"DataFusion interfaces to Rerun gRPC queries"），它负责把 Rerun 的存储/查询能力桥接到 DataFusion 的 TableProvider 体系上；Python 侧的绑定则在 rerun_py/src/catalog 下实现。

本示例依赖统一封装在rerun-sdk[all]中，无需单独安装 DataFusion、PyArrow 等依赖。另外从 docs/snippets/snippets.toml 可以看到，该示例目前仅提供 Python 实现（C++ 与 Rust 标记为 "Not implemented"），因此下文全部使用 Python API。

Setup：连接 Catalog Server 并读取数据

示例首先完成两件事：启动一个本地 catalog server用于演示（实际生产中通常连接云端实例），以及建立客户端并读取数据集。

from __future__ import annotations from pathlib import Path import datafusion as dfn import numpy as np import pyarrow as pa from datafusion import col, lit from datafusion import functions as F import rerun as rr sample_5_path = ( Path(__file__).parents[4] / "tests" / "assets" / "rrd" / "sample_5" ) server = rr.server.Server(datasets={"sample_dataset": sample_5_path}) CATALOG_URL = server.url() client = rr.catalog.CatalogClient(CATALOG_URL) dataset = client.get_dataset(name="sample_dataset") observations = dataset.filter_contents(["/observation/**"]).reader( index="real_time" )

逐行拆解这一环节：

1. 数据源：tests/assets/rrd/sample_5（仓库测试资产，见 tests/assets/rrd）是一个本地 RRD 录制目录，这里通过Path(__file__).parents[4]从示例文件所在目录回溯到仓库根目录后拼接得到。
2. 启动 server：rr.server.Server(datasets={"sample_dataset": sample_5_path})启动一个进程内的开放源码 catalog server（该 server 与 Rerun Hub 托管服务 API 兼容，但把所有数据加载进内存，适合小数据集的快速测试与本地实验，详见 get-data-out.md）。server.url()返回 server 的 gRPC 地址。
3. 客户端与数据集句柄：rr.catalog.CatalogClient(CATALOG_URL)建立客户端，client.get_dataset(name="sample_dataset")拿到数据集对象。对真实部署，只需把CATALOG_URL换成你的云端实例地址即可，其余代码不变。
4. 实体路径过滤：dataset.filter_contents(["/observation/**"])返回一个新的DatasetView，只保留匹配实体路径表达式的内容。从源码看（dataset_view.rs），该方法是叠加式的：如果视图已有内容过滤条件，新表达式会追加合并；它也支持"-/text/**"这类排除语法。
5. 读取 DataFrame：.reader(index="real_time")把视图物化为一个 DataFusion DataFrame。reader的关键参数（见 dataset_view.rs）包括：
   • index：使用的索引时间线，本示例为real_time；
   • fill_latest_at：是否用最近的合法值填充空值（后文 sub-episodes 部分会用到）；
   • include_semantically_empty_columns/include_tombstone_columns：是否包含语义上为空或标记为 tombstone 的列，默认均为False；
   • using_index_values：指定从各 segment 采样的具体索引值。

一个值得注意的列是rerun_segment_id：它标识数据所属的 segment（对应一次录制/一个 episode），是本示例所有聚合与 Join 的"分组键"。

Group-by / 聚合：追踪每个 episode 的时间范围

拿到observationsDataFrame 后，示例先做一个典型的 group-by 聚合——按rerun_segment_id分组，用first_value/last_value求出每个 episode 的起始与结束时间：

first_last = observations.aggregate( col("rerun_segment_id"), [ F.first_value(col("real_time")).alias("start"), F.last_value(col("real_time")).alias("end"), ], ) # Sort for consistency here first_last = first_last.sort("start") pa.table(first_last)["start"][0]

要点说明：

• aggregate的第一个参数是分组列（rerun_segment_id），第二个参数是聚合表达式列表；F.first_value/F.last_value分别取组内第一个 / 最后一个非空值，alias为结果列命名。
• 聚合结果顺序不确定，示例用.sort("start")显式排序保证一致性。
• pa.table(first_last)演示了如何把 DataFusion DataFrame 直接转为PyArrow Table与 pandas / polars 生态互操作。

Join 与查询：找出"迟到"的相机数据

现实录制的不同数据流往往不是同时开始的——关节位姿可能先于相机开始采集。下一段代码的目标就是量化这种时间差：对每个 episode 求出相机第一帧时间与关节第一位姿时间的差值，并找出差值超过阈值的"离群"episode。

joints = dataset.filter_contents(["/observation/joint_positions"]) # Find the earliest joint position in each episode (cast to unix epoch # nanoseconds for easier math later) joint_min_t = ( joints .reader(index="real_time") .with_column("joint_epoch_ns", col("real_time").cast(pa.int64())) .select("rerun_segment_id", "joint_epoch_ns") .aggregate( col("rerun_segment_id"), F.min(col("joint_epoch_ns")).alias("joint_min_t"), ) ) cameras = dataset.filter_contents(["/camera/**"]) # Find the earliest camera frame in each episode (cast to unix epoch # nanoseconds for easier math later) camera_min_t = ( cameras .reader(index="real_time") .with_column("camera_epoch_ns", col("real_time").cast(pa.int64())) .select( "rerun_segment_id", col("real_time").cast(pa.int64()).alias("camera_epoch_ns"), ) .aggregate( col("rerun_segment_id"), F.min(col("camera_epoch_ns")).alias("camera_min_t"), ) ) # Join the two dataframes min_t = camera_min_t.join( joint_min_t.with_column_renamed("rerun_segment_id", "segment_id"), left_on="rerun_segment_id", right_on="segment_id", how="left", ) delta_t = min_t.select( col("rerun_segment_id"), (col("camera_min_t") - col("joint_min_t")).alias("start_delta_t"), ) THRESHOLD_S = 1 NANO_S = 1_000_000_000 outliers = delta_t.filter( dfn.Expr.between( col("start_delta_t"), -THRESHOLD_S * NANO_S, THRESHOLD_S * NANO_S, negated=True, ), ) outliers = outliers.with_column( "start_delta_t_s", col("start_delta_t") / 1_000_000_000.0 ) print( f"{outliers.count()=}\n", f"{joint_min_t.count()=}\n", f"{camera_min_t.count()=}", sep="", )

这段代码展示了几个高频 DataFrame 操作：

• 列转换与类型 cast：col("real_time").cast(pa.int64())把时间戳转为 unix epoch 纳秒整数，便于后续做减法；with_column新增派生列，select精简列集（只保留需要的列，避免把重列拖进 Join）。
• 双流独立聚合再 Join：joint_min_t与camera_min_t分别是两条实体路径各自按 segment 求min的结果。Join 时把右侧的rerun_segment_id重命名为segment_id，以left_on/right_on指定两侧连接键，how="left"表示左连接（保留所有相机 segment，即使该 segment 没有关节数据）。
• 表达式过滤：dfn.Expr.between(..., negated=True)等价于"不在闭区间[-1s, +1s]内"，即筛选出相机与关节起始时间差超过 ±1 秒的 segment。随后又用with_column把差值换算成秒（除以1e9）便于人工阅读。
• 触发求值：count()是触发 DataFusion 懒执行计划真正运行的算子之一，print输出三个计数用于验证。

⚠️性能警告（来自官方文档）尽管 DataFusion 是懒执行拉取数据的，但 Rerun 目前没有把数据负载与其时间线解耦。例如本例为了检查相机数据的 min/max 时间戳，就必须把完整的相机数据拉取进来。当数据已在本地内存中时这很快，但在云端大规模数据上可能成为瓶颈。因此做此类分析前，建议先评估目标列的数据量，尽量用select只保留必要列（示例中camera_min_t就只 select 了rerun_segment_id与时间列）。

提取子片段：用窗口函数把一次录制切成多个 episode

很多录制的单条 recording 实际包含多个 episode——例如一个机械臂多次抓放物体，每次抓取可视为一个独立 episode。示例利用"夹爪开合"信号把连续时间范围切分成子片段：夹爪打开（gripper_open由 False 变 True）作为子片段起点，夹爪关闭作为终点，供下游流程逐段处理。

# Grab a dataframe all_data = ( dataset .filter_contents(["/action/**", "/observation/**"]) .reader(index="real_time", fill_latest_at=True) .filter( col( "/observation/joint_positions:Scalars:scalars" ).is_not_null() # filter out rows where there is no observation ) ) # Drop heavy columns for performance light_slice = all_data.select( "rerun_segment_id", "real_time", "/observation/gripper_position:Scalars:scalars", ) # Define criteria for sub-episode start/end THRESHOLD = 0.1 light_slice = light_slice.with_column( "gripper_open", col("/observation/gripper_position:Scalars:scalars") > [THRESHOLD], ) # Find start and end light_slice = light_slice.with_column( "prev_gripper_open", F.lag( col("gripper_open"), default_value=False, partition_by=[col("rerun_segment_id")], order_by=[col("real_time")], ), ) light_slice = light_slice.with_column( "gripper_change", col("gripper_open").cast(pa.int8()) - col("prev_gripper_open").cast(pa.int8()), ) slice_times = light_slice.with_column( "start", F .case(col("gripper_change")) .when(lit(1), col("real_time")) .otherwise(lit(None)), ).with_column( "end", F .case(col("gripper_change")) .when(lit(-1), col("real_time")) .otherwise(lit(None)), )

这一步的思路是：用gripper_change = gripper_open - prev_gripper_open捕捉状态跳变——1表示夹爪从关到开（子片段起点），-1表示从开到关（子片段终点）；然后用CASE WHEN把跳变点对应的时间戳写入start/end列，非跳变点置为NULL。

其中值得展开的细节：

• fill_latest_at=True：读取时用最近合法值填充空值，保证每行都有夹爪位置可比较；
• 列名即实体路径：列名是完整的组件路径，例如"/observation/gripper_position:Scalars:scalars"，直接以col()引用；
• F.lag窗口函数：partition_by=[col("rerun_segment_id")]保证跨 segment 不串行，order_by=[col("real_time")]按时间取上一行，default_value=False处理每段第一行没有前驱的情况；
• 布尔转整数相减：cast(pa.int8())后相减得到差分信号；
• .select瘦身：light_slice只保留rerun_segment_id、real_time和夹爪位置三列，丢弃"/action/**"等重列，是控制内存与传输量的关键实践。

接下来是核心难点——把稀疏的start/end跳变点"铺满"每一行，从而用real_time BETWEEN dense_start AND dense_end圈定子片段范围：

# Helper because pyarrow timestamps didn't have a nice min/max utility max_ts = pa.scalar(np.iinfo(np.int64).max, type=pa.timestamp("ns")) min_ts = pa.scalar( np.iinfo(np.int64).min + 1_000_000_000, type=pa.timestamp("ns") ) # This generates the column for the last observed start time slice_dense_times = ( slice_times .select("rerun_segment_id", "real_time", "start", "end") .with_column( "dense_start", F.last_value(col("start")).over( dfn.expr.Window( window_frame=dfn.expr.WindowFrame("rows", None, 0), order_by=col("real_time"), partition_by=col("rerun_segment_id"), null_treatment=dfn.common.NullTreatment.IGNORE_NULLS, ) ), ) .fill_null(value=max_ts, subset=["dense_start"]) ) # This generates the column for the next observed end time (by finding the # last_value in reversed order) slice_dense_times = slice_dense_times.with_column( "dense_end", F.last_value(col("end")).over( dfn.expr.Window( window_frame=dfn.expr.WindowFrame("rows", None, 0), order_by=col("real_time").sort(ascending=False), partition_by=col("rerun_segment_id"), null_treatment=dfn.common.NullTreatment.IGNORE_NULLS, ) ), ).fill_null(value=min_ts, subset=["dense_end"]) slice_dense_times = slice_dense_times.select( "rerun_segment_id", "real_time", "dense_start", "dense_end" ) sub_episodes = slice_dense_times.filter( dfn.Expr.between(col("real_time"), col("dense_start"), col("dense_end")), ) print(f"{sub_episodes.count()=}")

这段代码的机制可以概括为两条互补的"向前/向后填充"：

1. dense_start（最后观察到的起点）：窗口帧rows从窗口起点到当前行（WindowFrame("rows", None, 0)），order_by=real_time升序，F.last_value配合NullTreatment.IGNORE_NULLS得到"截至当前行最后一次出现的非空start"。因此每个子片段起点之后的每一行都继承该起点。
2. dense_end（下一个观察到的终点）：把real_time按降序排列后再取last_value，等价于"从当前行往后第一个非空end"——每个子片段终点之前的每一行都继承该终点。

随后fill_null兜底：dense_start为空（起点之前的行）填max_ts（时间戳能表示的最大值），dense_end为空（终点之后的行）填min_ts（最小值附近）。这样每条记录的real_time要么落在某个[dense_start, dense_end]区间内（属于某个子片段），要么被兜底值排除在BETWEEN之外。最后用dfn.Expr.between过滤即得到sub_episodes。

这种"窗口函数 + 空值处理 + BETWEEN 过滤"的切片手法，同样适用于状态机类信号（如电池充电开始/结束、阀门开闭）驱动的任何子序列切分任务，是值得沉淀为通用工具的模式。

性能注意事项小结

综合官方文档的警告与本示例的代码实践，在使用 Rerun Dataframe 接口做大规模分析时应注意：

• 懒执行不是"零成本"：计划只有被count()/ collect / 导出等操作触发时才执行，但 Rerun 目前不将数据负载与时间线解耦，涉及时间统计（如 min/max 时间戳）时仍要拉取对应列的全部数据；
• 尽早select瘦身：凡是后续用不到的列（尤其是相机图像等重列），应在 Join / 聚合之前用select剔除，示例中light_slice与camera_min_t都体现了这一点；
• 本地 vs 云端：本地内存中的小数据集上这些操作很快；云端大规模数据下时间类聚合可能成为瓶颈，需要结合数据规模规划查询；
• 按 segment 隔离计算：凡是跨 episode 的操作（lag、last_value、aggregate）都要记得以rerun_segment_id作为partition_by/ 分组列，避免不同 episode 的数据互相污染。

参考资料

• 本文示例完整源码：docs/snippets/all/howto/dataframe_operations.py
• DataFusion 桥接实现：re_datafusion（Rust 侧 TableProvider / gRPC 查询实现）
• Python 绑定与DatasetViewAPI：dataset_view.rs
• Catalog server 启动与连接方式：get-data-out.md
• 数据重采样相关操作：dataset_resampling.md

【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun