Skip to main content

Video Frames

Store complete encoded videos and read their logical frame rows for training. First create a multimodal conn as shown in Multimodal Tables. Provide your own video files and frame metadata for the examples below.

Video Frame Storage​

For a frame table, set video-frame-field to one or more comma-separated scalar BLOB columns. Paimon stores frame ordinals in VideoFrameDescriptor and compact .video indexes, so the normal data file does not need a frame_index or frame_timestamp column:

import pyarrow as pa
import pypaimon.multimodal as pm

frames = conn.create_table(
    "video_frames",
    schema=pa.schema([
        pa.field("episode_id", pa.int64()),
        pa.field("state", pa.list_(pa.float32())),
        pa.field("action", pa.list_(pa.float32())),
        pa.field("video", pa.large_binary()),
    ]),
    options={
        "video-frame-field": "video",
    },
)

add_video accepts a complete encoded-video source plus the ordinary columns for its logical frame rows. Blob.from_local is descriptor-backed and streams the source; the MP4 is not first loaded into Python memory:

video = pm.Blob.from_local("/data/episode-42.mp4")
frames.add_video(video, frame_rows, first_frame=0)

Use add_videos to keep one writer and one commit open across several source videos. This is the path that lets one .video object pack multiple complete MP4 payloads and reduces object/manifest count:

frames.add_videos([
    (pm.Blob.from_local("/data/episode-42.mp4"), episode_42_rows),
    (pm.Blob.from_local("/data/episode-43.mp4"), episode_43_rows),
])

To encode an ordered sequence of image bytes or Pillow images before storing it, use Python 3.10 or newer, install the optional video dependencies, and call add_images_as_video:

pip install 'pypaimon[video]'
frames.add_images_as_video(
    image_frames,
    episode_44_rows,
    fps=30,
    codec="libx264",
    pixel_format="yuv420p",
    gop_size=2,
    codec_options={"crf": "18", "preset": "fast"},
)

The image and row counts must match, and every image must have the same dimensions. The container is MP4; codec-specific options are passed to PyAV. Encoding may be lossy and the original image payloads are not retained. A smaller GOP improves random frame access but usually increases file size. After encoding, Paimon copies the MP4 bytes into .video storage without transcoding them again.

Each item may also be (video, frame_rows, first_frame). Frame ordinals are generated consecutively from first_frame. If application semantics require PTS, wall-clock time, or a non-unit sampling map, retain that value in a normal column; .video version 1 addresses presentation-order frames with stride one.

Remote sources can use an explicit descriptor. Its URI must be readable with the table's configured FileIO credentials:

video = pm.BlobDescriptor(
    "oss://source-bucket/episode-43.mp4",
    offset=0,
    length=video_size,
)
frames.add_video(video, episode_43_rows)

With PyAV, PyPaimon builds keyframe indexes for supported ISO BMFF videos (such as MP4); other decodable videos use the scan fallback.

The writer deduplicates exact payload descriptor identity inside each .video file. Its video grouping policy coordinates normal, BLOB, and vector rolling at payload boundaries. A file may exceed its target before the next boundary. The normal .blob format is unchanged.

Update frame rows and replace a video​

Frame rows use the ordinary table update API for non-video columns. This writes only the changed data-evolution columns and keeps the existing .video object and frame descriptors unchanged:

frames.update(
    where="episode_id = 42",
    values={"state": corrected_states},
)

Writing a complete encoded video uses the specialized API. replace_video selects the matching logical rows in _ROW_ID order, assigns consecutive frame ordinals, and writes a video-column delta without rewriting the normal data file:

frames.replace_video(
    where="episode_id = 42",
    video=pm.Blob.from_local("/data/episode-42-corrected.mp4"),
    first_frame=0,
)

The generic update() API rejects assignments to the configured video-frame-field; use replace_video() so Paimon can preserve the complete video payload and its embedded row-to-frame mapping.

Read with PyTorch DataLoader​

Install the PyTorch extra, then convert the multimodal scan directly. Unlike a regular BLOB materialization, ScanQuery.to_torch always returns serialized descriptors for BLOB columns. DataLoader workers receive Paimon splits and open the selected video ranges themselves:

pip install 'pypaimon[torch]'
from torch.utils.data import DataLoader

dataset = (
    frames.scan()
    .select(["episode_id", "state", "action", "video"])
    .to_torch(streaming=True)
)

loader = DataLoader(
    dataset,
    batch_size=32,
    num_workers=4,
    shuffle=False,
)

The video value is serialized VideoFrameDescriptor bytes, not repeated MP4 bytes. VideoFrameCollator owns a bounded, process-local LRU cache keyed by the physical video range, so different frame descriptors reuse one decoder session in each DataLoader worker. decoder_factory(stream) opens any codec library; decode_fn(decoder, frame_index, row) receives the embedded frame ordinal.

import av
import torch
import pypaimon.multimodal as pm
from torch.utils.data import DataLoader


class SequentialPyAvDecoder:
    def __init__(self, stream):
        self.container = av.open(stream)
        self._reset()

    def _reset(self):
        self.frames = iter(self.container.decode(video=0))
        self.next_index = 0

    def frame(self, index):
        if index < self.next_index:
            self.container.seek(0)
            self._reset()
        while self.next_index <= index:
            frame = next(self.frames)
            self.next_index += 1
        array = frame.to_ndarray(format="rgb24")
        return torch.from_numpy(array).permute(2, 0, 1)

    def close(self):
        self.container.close()


def decode_frame(decoder, frame_index, row):
    return decoder.frame(frame_index)


collator = pm.VideoFrameCollator(
    frames,
    video_column="video",
    decoder_factory=SequentialPyAvDecoder,
    decode_fn=decode_frame,
    output_column="frame",
    max_open_videos=4,
)

loader = DataLoader(
    dataset,
    batch_size=32,
    num_workers=4,
    shuffle=False,
    collate_fn=collator,
    persistent_workers=True,
)

for batch in loader:
    # batch["frame"] is [N, C, H, W].
    train(batch["frame"], batch["episode_id"])

The example decoder additionally requires pip install av; PyAV is not a PyPaimon dependency.

Define the decoder class and function at module scope when DataLoader uses the spawn multiprocessing method. For sequential video decoding, keep shuffle=False; row-buffer shuffle can turn monotonic frame access into seeks and require more open decoder sessions. Paimon's worker sharding is split-level, so it does not duplicate a split across workers.