pyflink.dataframe.read_video_frames#

read_video_frames(source: Union[str, Sequence[str], pyflink.dataframe.dataframe.DataFrame], video_column: Optional[str] = None, *, frame_selector: str = 'all_frames', sample_interval_ms: Optional[int] = None, max_frames: Optional[int] = None, image_height: Optional[int] = None, image_width: Optional[int] = None, on_error: str = 'raise', concurrency: Optional[int] = None) → pyflink.dataframe.dataframe.DataFrame[source]#

Read video frames from video URI/path strings, glob patterns, or a video column.

source can be a single video path, a sequence of video paths, a directory, a glob pattern, or a DataFrame containing a video path column. File and directory discovery uses Flink’s Java-side FileSystem. A directory path is listed recursively with or without a trailing /.

Supported glob tokens are * for one path segment, ** for recursive path segments, ? for one non-/ character, [] character classes, {a,b} alternatives, and {0..99} numeric ranges. The recursive ** token must be a full path segment, for example videos/**/*.mp4. A path is treated as a glob only when it contains an unescaped *, ?, [, or {; . is always a literal character.

The returned DataFrame keeps the input columns and appends frame and metadata columns. When concurrency is greater than 1, input rows are rebalanced before frame extraction so video files can be distributed across parallel extraction tasks. Empty directories and glob patterns with no matches return an empty DataFrame. Missing non-glob paths raise the underlying FileSystem error.

Parameters

source – A video URI/path string, glob pattern, non-empty sequence of video URI/path strings or glob patterns, or a DataFrame containing a video column.
video_column – Column name to read when source is a DataFrame.
frame_selector – Frame selection mode, "keyframe", "sample", or "all_frames". Defaults to "all_frames".
sample_interval_ms – Sampling interval for frame_selector="sample".
max_frames – Optional safety limit for emitted frames per input row. None emits all selected frames.
image_height – Optional output frame height in pixels. Must be provided together with image_width.
image_width – Optional output frame width in pixels. Must be provided together with image_height.
on_error – Error handling policy. "raise" (default) propagates failures; "skip" emits no rows for inputs that fail before the first frame and preserves rows already emitted before a later decode error.
concurrency – Optional maximum parallelism for reading and extracting video frames. Values greater than 1 also rebalance input rows before frame extraction.

Returns

A DataFrame containing the source columns plus –

frame: decoded video frame as DataType.image().
metadata: per-frame metadata as a Row/struct column, including uri, video_stream_index, frame_index, pts, time_ms, key_frame, start_time_ms, and end_time_ms. Access a field with pf.col("metadata").get("time_ms").

Example::

>>> frames = pf.read_video_frames(
...     "/tmp/video.mp4",
...     image_height=480,
...     image_width=640,
... )
>>> frames = pf.read_video_frames(
...     "file:///tmp/video.mp4",
...     image_height=480,
...     image_width=640,
... )
>>> frames = pf.read_video_frames(
...     ["oss://bucket/a.mp4", "oss://bucket/b.mp4"],
...     image_height=480,
...     image_width=640,
... )
>>> frames = pf.read_video_frames(
...     "oss://bucket/videos/**/*.mp4",
...     image_height=480,
...     image_width=640,
... )
>>> videos = pf.read_parquet("oss://bucket/metadata.parquet",
...                          schema={"video_url": pf.DataType.string()})
>>> frames = pf.read_video_frames(videos, "video_url")