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

Rerun DataFrame 操作实战基于 DataFusion 的聚合、Join 与子片段提取【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerunRerun 通过 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 同时支持将结果转换为主流工具使用的 DataFramepandas、polars、pyarrow方便后续与既有分析栈衔接。在 Rerun 仓库中这一接口的 Rust 侧实现在 re_datafusion crateDataFusion 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(namesample_dataset) observations dataset.filter_contents([/observation/**]).reader( indexreal_time )逐行拆解这一环节数据源tests/assets/rrd/sample_5仓库测试资产见 tests/assets/rrd是一个本地 RRD 录制目录这里通过Path(__file__).parents[4]从示例文件所在目录回溯到仓库根目录后拼接得到。启动 serverrr.server.Server(datasets{sample_dataset: sample_5_path})启动一个进程内的开放源码 catalog server该 server 与 Rerun Hub 托管服务 API 兼容但把所有数据加载进内存适合小数据集的快速测试与本地实验详见 get-data-out.md。server.url()返回 server 的 gRPC 地址。客户端与数据集句柄rr.catalog.CatalogClient(CATALOG_URL)建立客户端client.get_dataset(namesample_dataset)拿到数据集对象。对真实部署只需把CATALOG_URL换成你的云端实例地址即可其余代码不变。实体路径过滤dataset.filter_contents([/observation/**])返回一个新的DatasetView只保留匹配实体路径表达式的内容。从源码看dataset_view.rs该方法是叠加式的如果视图已有内容过滤条件新表达式会追加合并它也支持-/text/**这类排除语法。读取 DataFrame.reader(indexreal_time)把视图物化为一个 DataFusion DataFrame。reader的关键参数见 dataset_view.rs包括index使用的索引时间线本示例为real_timefill_latest_at是否用最近的合法值填充空值后文 sub-episodes 部分会用到include_semantically_empty_columns/include_tombstone_columns是否包含语义上为空或标记为 tombstone 的列默认均为Falseusing_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(indexreal_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(indexreal_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_onrerun_segment_id, right_onsegment_id, howleft, ) 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, negatedTrue, ), ) 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 操作列转换与类型 castcol(real_time).cast(pa.int64())把时间戳转为 unix epoch 纳秒整数便于后续做减法with_column新增派生列select精简列集只保留需要的列避免把重列拖进 Join。双流独立聚合再 Joinjoint_min_t与camera_min_t分别是两条实体路径各自按 segment 求min的结果。Join 时把右侧的rerun_segment_id重命名为segment_id以left_on/right_on指定两侧连接键howleft表示左连接保留所有相机 segment即使该 segment 没有关节数据。表达式过滤dfn.Expr.between(..., negatedTrue)等价于不在闭区间[-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(indexreal_time, fill_latest_atTrue) .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_valueFalse, 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_atTrue读取时用最近合法值填充空值保证每行都有夹爪位置可比较列名即实体路径列名是完整的组件路径例如/observation/gripper_position:Scalars:scalars直接以col()引用F.lag窗口函数partition_by[col(rerun_segment_id)]保证跨 segment 不串行order_by[col(real_time)]按时间取上一行default_valueFalse处理每段第一行没有前驱的情况布尔转整数相减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 didnt have a nice min/max utility max_ts pa.scalar(np.iinfo(np.int64).max, typepa.timestamp(ns)) min_ts pa.scalar( np.iinfo(np.int64).min 1_000_000_000, typepa.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_framedfn.expr.WindowFrame(rows, None, 0), order_bycol(real_time), partition_bycol(rerun_segment_id), null_treatmentdfn.common.NullTreatment.IGNORE_NULLS, ) ), ) .fill_null(valuemax_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_framedfn.expr.WindowFrame(rows, None, 0), order_bycol(real_time).sort(ascendingFalse), partition_bycol(rerun_segment_id), null_treatmentdfn.common.NullTreatment.IGNORE_NULLS, ) ), ).fill_null(valuemin_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()})这段代码的机制可以概括为两条互补的向前/向后填充dense_start最后观察到的起点窗口帧rows从窗口起点到当前行WindowFrame(rows, None, 0)order_byreal_time升序F.last_value配合NullTreatment.IGNORE_NULLS得到截至当前行最后一次出现的非空start。因此每个子片段起点之后的每一行都继承该起点。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.pyDataFusion 桥接实现re_datafusionRust 侧 TableProvider / gRPC 查询实现Python 绑定与DatasetViewAPIdataset_view.rsCatalog 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考