read_mcap#

ray.data.read_mcap(paths: str | List[str], *, topics: List[str] | Set[str] | None = None, time_range: Tuple[int, int] | TimeRange | None = None, message_types: List[str] | Set[str] | None = None, include_metadata: bool = True, log_time_order: bool = True, include_row_id: bool = False, filesystem: pyarrow.fs.FileSystem | None = None, parallelism: int = -1, num_cpus: float | None = None, num_gpus: float | None = None, memory: float | None = None, partition_filter: PathPartitionFilter | None = None, partitioning: Partitioning = None, include_paths: bool = False, ignore_missing_paths: bool = False, shuffle: Literal['files'] | FileShuffleConfig | None = None, file_extensions: List[str] | None = None, concurrency: int | None = None, override_num_blocks: int | None = None, label_selector: Dict[str, str] | None = None, fallback_strategy: List[Dict[str, Any]] | None = None, max_calls: int | None = None, resources: Dict[str, float] | None = None, accelerator_type: str | None = None, runtime_env: Dict[str, Any] | None = None, ray_remote_args: Dict[str, Any] | None = None) → Dataset[source]#

Create a Dataset from MCAP (Message Capture) files.

MCAP is a format commonly used in robotics and autonomous systems for storing ROS2 messages and other time-series data. This reader provides predicate pushdown optimization for efficient filtering by topics, time ranges, and message types.

Each row is one message. The data column holds decoded JSON values when every selected topic is JSON-encoded, and the raw payload bytes otherwise. To decode the JSON topics of a recording that also has other encodings, select them with topics or message_types. JSON values are typed from the values each read task holds: a field that only some files have is null in the other rows, and a field whose type differs between files, such as a number in one and a string in another, fails where their rows are combined. On the V2 datasource, Dataset.schema() types them before reading, from the selected messages of the first chunk that holds one. A field that first appears in a later chunk or file is missing from that schema.

By default, read_mcap reads through the V2 datasource (DataContext.use_datasource_v2). Its rows differ from the legacy reader’s in three ways:

  • schema_data and the new channel_metadata column are dictionary-encoded, so a block stores each channel’s values once. pandas shows them as category columns. Cast them to binary and string before you group by them or write blocks from several read tasks into one Arrow IPC file. Cast schema_data before you mix this output with the legacy reader’s in one Parquet folder.

  • A large file is split across read tasks. Its messages keep log_time order across the whole file, and files come out in the order they were listed, only with DataContext.execution_options.preserve_order.

  • override_num_blocks has no effect.

To use the legacy reader, set DataContext.get_current().use_datasource_v2 to False or set RAY_DATA_USE_DATASOURCE_V2=0. This also moves read_parquet back to its legacy reader.

Examples

Noindex:

Read all MCAP files in a directory.

>>> import ray
>>> ds = ray.data.read_mcap("s3://bucket/mcap-data/")
>>> ds.schema()

Read with filtering for specific topics and time range.

>>> from ray.data.datasource import TimeRange
>>> ds = ray.data.read_mcap(
...     "s3://bucket/mcap-data/",
...     topics={"/camera/image_raw", "/lidar/points"},
...     time_range=TimeRange(start_time=1000000000, end_time=5000000000),
...     message_types={"sensor_msgs/Image", "sensor_msgs/PointCloud2"}
... )

Alternatively, use a tuple for time range (backwards compatible).

>>> ds = ray.data.read_mcap(
...     "s3://bucket/mcap-data/",
...     topics={"/camera/image_raw", "/lidar/points"},
...     time_range=(1000000000, 5000000000),
... )

Read multiple local files with include_paths.

>>> ray.data.read_mcap(
...     ["local:///path/to/file1.mcap", "local:///path/to/file2.mcap"],
...     include_paths=True
... )

Read with topic filtering and metadata inclusion.

>>> ds = ray.data.read_mcap(
...     "data.mcap",
...     topics={"/camera/image_raw", "/lidar/points"},
...     include_metadata=True,
...     include_paths=True
... )

Store schema_data as plain bytes, as the legacy reader does.

>>> import pyarrow as pa
>>> def plain_schema_data(t: pa.Table) -> pa.Table:
...     i = t.schema.get_field_index("schema_data")
...     return t.set_column(i, "schema_data", t["schema_data"].cast(pa.binary()))
>>> ds = ds.map_batches(plain_schema_data, batch_format="pyarrow")
Parameters:

  • paths (str | List[str]) – A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories.

  • topics (List[str] | Set[str] | None) – Optional list or set of topic names to include. If specified, only messages from these topics will be read.

  • time_range (Tuple[int, int] | TimeRange | None) – Optional time range for filtering messages by timestamp. Can be either a tuple of (start_time, end_time) in nanoseconds (for backwards compatibility) or a TimeRange object. Both values must be non-negative and start_time < end_time.

  • message_types (List[str] | Set[str] | None) – Optional list or set of message type names (schema names) to include. Only messages with matching schema names will be read.

  • include_metadata (bool) – Whether to include MCAP metadata fields in the output. Defaults to True. When True, includes schema, channel, and message metadata. On the V2 datasource, channel_metadata holds each channel’s metadata map as JSON text with sorted keys, such as ROS 2’s offered_qos_profiles; read it with json.loads. It and schema_data are dictionary-encoded, so a block stores each of a channel’s values once.

  • log_time_order (bool) – Whether each file’s messages come out in ascending log_time order. Defaults to True. Messages with equal log times keep the order they were written in. If False, messages come in the order they were written. False requires the V2 datasource. On the V2 datasource (DataContext.use_datasource_v2), a read task holds about RAY_DATA_MCAP_BIN_PACKING_BYTES of estimated read memory (128 MiB by default). A large file is split across several tasks, and small files can share one. Each task returns each of its files in order, but tasks finish in any order, so a split file is in order only within each task. With DataContext.execution_options.preserve_order, files come out in the order they were listed, and with log_time_order they are not split: each file is read by one task, so it comes out in order from start to end. That task holds the chunks that overlap in time together, so a file whose chunks all overlap is held whole. A file without a chunk index is always read by one task, which holds its selected messages in memory to sort them.

  • include_row_id (bool) – If True, add a row_id column with a deterministic id per message, built from the file path and the message’s position in the file. A message gets the same id on every run, however the read is split into tasks, so the column can serve as the id_column of a CheckpointConfig. Turned on automatically when the current checkpoint config’s id_column is row_id. Requires the V2 datasource (DataContext.use_datasource_v2). read_mcap refuses CheckpointConfig(generated_id_column=...), which only read_parquet supports.

  • filesystem (pyarrow.fs.FileSystem | None) – The PyArrow filesystem implementation to read from.

  • parallelism (int) – This argument is deprecated. Use override_num_blocks argument.

  • num_cpus (float | None) – The number of CPUs to reserve for each parallel read worker.

  • num_gpus (float | None) – The number of GPUs to reserve for each parallel read worker. For example, specify num_gpus=1 to request 1 GPU for each parallel read worker.

  • memory (float | None) – The heap memory in bytes to reserve for each parallel read worker.

  • partition_filter (PathPartitionFilter | None) – A PathPartitionFilter. Use with a custom callback to read only selected partitions of a dataset.

  • partitioning (Partitioning) – A Partitioning object that describes how paths are organized. Defaults to None.

  • include_paths (bool) – If True, include the path to each file. File paths are stored in the 'path' column.

  • ignore_missing_paths (bool) – If True, ignores any file paths in paths that are not found. Defaults to False.

  • shuffle (Literal['files'] | ~ray.data.datasource.file_based_datasource.FileShuffleConfig | None) – If setting to “files”, randomly shuffle input files order before read. If setting to FileShuffleConfig, you can pass a seed to shuffle the input files. Defaults to not shuffle with None.

  • file_extensions (List[str] | None) – A list of file extensions to filter files by. Defaults to ["mcap"].

  • concurrency (int | None) – The maximum number of Ray tasks to run concurrently. Set this to control number of tasks to run concurrently. This doesn’t change the total number of tasks run or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources.

  • override_num_blocks (int | None) – Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn’t manually set this value in most cases. The V2 datasource ignores it and logs a warning: it sizes each read task by bytes instead, about 128 MiB by default (RAY_DATA_MCAP_BIN_PACKING_BYTES).

  • label_selector (Dict[str, str] | None) – Labels required on the node where each read task runs.

  • fallback_strategy (List[Dict[str, Any]] | None) – Alternative label requirements that Ray tries in order if label_selector can’t be satisfied.

  • max_calls (int | None) – The maximum number of read tasks a worker runs before exiting.

  • resources (Dict[str, float] | None) – Custom resources to reserve for each read task, expressed as a mapping from resource name to quantity.

  • accelerator_type (str | None) – The accelerator type required for each read task.

  • runtime_env (Dict[str, Any] | None) – The runtime environment to use for each read task.

  • ray_remote_args (Dict[str, Any] | None) – Additional options passed to ray.remote() for each read task. This argument is deprecated and will be removed in Ray 2.64. Use the named remote parameters instead.

Returns:

Dataset producing records read from the specified MCAP files.

Return type:

Dataset

PublicAPI (alpha): This API is in alpha and may change before becoming stable.