Video Operators-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Limits

Available only in Realtime Compute Engine VVR 11.8 and later.
Video reading and decoding use PyAV/FFmpeg.
Video URIs must be accessible through the Flink FileSystem.
video_split and video_explode_frames are UDTFs and must be invoked through DataFrame.join_lateral.
This topic uses Video Segment Reference (VIDEO_STRUCT_TYPE), Video Metadata (VIDEO_METADATA_TYPE) and Video Frame Metadata (VIDEO_FRAME_METADATA_TYPE) to refer to the DataFrame API types defined in pyflink.multimodal.types.

Operator Overview

Category	Operator	Description
Metadata	video_metadata	Reads basic video information such as resolution, frame rate, duration, and encoding format. This operator reads only the container and video stream metadata and does not decode video frames.
Video Splitting	video_split	Splits a complete video or an existing video segment into multiple segment references and expands them into one segment reference per row. This operator does not copy video content or read video frames. It only calculates time windows and produces segment references.
Frame Extraction	video_explode_frames	Extracts frames from a complete video or a video segment and expands the results into one frame per row.
Frame Extraction	video_extract_frames	Extracts frames from a complete video or a video segment and collects the results of the same input video into an array in a single row. For long videos or a large number of extracted frames, use `video_explode_frames` to avoid holding too many images in a single row.

Data Type Description

Video Segment Reference (`VIDEO_STRUCT_TYPE`)

The field structure of VIDEO_STRUCT_TYPE is as follows:

DataType.struct({
    "uri": DataType.string(),
    "start_time_ms": DataType.int64(),
    "end_time_ms": DataType.int64()
})

Field	DataFrame API type	Description
`uri`	`DataType.string()`	Video file path or object storage URI.
`start_time_ms`	`DataType.int64()`	Segment start time in milliseconds (inclusive).
`end_time_ms`	`DataType.int64()`	Segment end time in milliseconds (inclusive).

A segment reference only describes the closed interval [start_time_ms, end_time_ms] and does not copy video content.

Video Metadata (`VIDEO_METADATA_TYPE`)

The field structure of VIDEO_METADATA_TYPE is as follows:

DataType.struct({
    "width": DataType.int32(),
    "height": DataType.int32(),
    "fps": DataType.float64(),
    "duration_ms": DataType.int64(),
    "frame_count": DataType.int64(),
    "time_base": DataType.float64(),
    "codec_name": DataType.string(),
    "video_stream_index": DataType.int32()
})

Field	DataFrame API type	Description
`width`	`DataType.int32()`	Video width in pixels.
`height`	`DataType.int32()`	Video height in pixels.
`fps`	`DataType.float64()`	Video frame rate.
`duration_ms`	`DataType.int64()`	Video duration in milliseconds.
`frame_count`	`DataType.int64()`	Total number of video frames.
`time_base`	`DataType.float64()`	Video stream time base, used for calculating frame timestamps.
`codec_name`	`DataType.string()`	Video encoding format name, e.g., `h264`, `hevc`.
`video_stream_index`	`DataType.int32()`	Selected video stream index.

Video Frame Metadata (`VIDEO_FRAME_METADATA_TYPE`)

The field structure of VIDEO_FRAME_METADATA_TYPE is as follows:

DataType.struct({
    "uri": DataType.string(),
    "video_stream_index": DataType.int32(),
    "frame_index": DataType.int64(),
    "pts": DataType.int64(),
    "time_ms": DataType.int64(),
    "key_frame": DataType.boolean(),
    "start_time_ms": DataType.int64(),
    "end_time_ms": DataType.int64(),
})

Field	DataFrame API type	Description
`uri`	`DataType.string()`	Original video file path or object storage URI.
`video_stream_index`	`DataType.int32()`	Video stream index of the output frame.
`frame_index`	`DataType.int64()`	Index of the output frame in the current frame extraction result, starting from 0.
`pts`	`DataType.int64()`	Original frame timestamp in the video file, typically used for alignment with video processing tools. `None` when it cannot be determined.
`time_ms`	`DataType.int64()`	Video time corresponding to the frame in milliseconds. `None` when it cannot be determined.
`key_frame`	`DataType.boolean()`	Whether this is a keyframe.
`start_time_ms`	`DataType.int64()`	Start time of the current frame extraction input segment in milliseconds. `None` when the input is a complete URI.
`end_time_ms`	`DataType.int64()`	End time of the current frame extraction input segment in milliseconds. `None` when the input is a complete URI.

Common Runtime Parameters

The function signatures retain the runtime parameters that each operator actually supports. To avoid repetition, the operator parameter tables describe only business parameters. Runtime parameters are described below.

Parameter	Type	Default	Applies to	Description
`concurrency`	`Optional[int]`	`None`	All operators on this page	Concurrency of the UDF or UDTF. `None` uses the framework default.

Video Metadata and Splitting

video_metadata

Reads basic video information such as resolution, frame rate, duration, and encoding format. This operator reads only the container and video stream metadata and does not decode video frames.

Input type: DataType.string(), the video URI column.

Function signature:

video_metadata(
    *columns,
    on_error="raise",
    container_options=None,
    read_chunk_size=None,
    max_cached_blocks=None,
    read_ahead_blocks=None,
    concurrency=None
)

Parameter	Type	Default	Description
`on_error`	`str`	`"raise"`	`"raise"` throws an exception on error; `"null"` returns null for unreadable input.
`container_options`	`Optional[Mapping[str, str]]`	`None`	Parameters passed to PyAV `av.open`.
`read_chunk_size`	`Optional[int]`	`None`	The number of bytes read from the Java file system bridge at a time. The value must be greater than 0 and cannot exceed 16 MiB.
`max_cached_blocks`	`Optional[int]`	`None`	The maximum number of blocks cached for each file. `0` disables block caching.
`read_ahead_blocks`	`Optional[int]`	`None`	The number of blocks to read ahead after a cache miss. `0` disables read-ahead.

A null URI returns None.

Return type: VIDEO_METADATA_TYPE.

from pyflink.dataframe import col
from pyflink.multimodal.operators import video_metadata

result = df.with_column(
    "video_metadata",
    video_metadata(
        col("uri"),
        on_error="null"
    )
)

video_split

Splits a complete video or an existing video segment into multiple segment references and expands them into one segment reference per row. This operator does not copy video content or read video frames. It only calculates time windows.

Input type: DataType.string() or VIDEO_STRUCT_TYPE, optionally followed by VIDEO_METADATA_TYPE or DataType.int64(). The first input is a video URI or an existing segment reference. The second input can provide video metadata or the duration in milliseconds to avoid repeated probing. A null video input outputs no rows.

Function signature:

video_split(
    *columns,
    segment_duration_ms=None,
    num_segments=None,
    video_duration_ms=None,
    max_segments=1024,
    on_error="raise",
    container_options=None,
    read_chunk_size=None,
    max_cached_blocks=None,
    read_ahead_blocks=None,
    concurrency=None
)

Parameter	Type	Default	Description
`segment_duration_ms`	`Optional[int]`	`None`	The fixed segment length, in milliseconds. The value must be greater than 0. Exactly one of this parameter and `num_segments` must be set.
`num_segments`	`Optional[int]`	`None`	The target number of segments of approximately equal length. The value must be greater than 0. Exactly one of this parameter and `segment_duration_ms` must be set.
`video_duration_ms`	`Optional[int]`	`None`	The explicit video duration, in milliseconds. It is used as a fallback when the input has no end time and the second input does not provide a valid duration.
`max_segments`	`int`	`1024`	The maximum number of segments output in a single row. The value must be greater than 0. Remaining segments are not output after the limit is reached.
`on_error`	`str`	`"raise"`	`"raise"` throws an exception; `"skip"` outputs no segments for failed inputs.
`container_options`	`Optional[Mapping[str, str]]`	`None`	Parameters passed to PyAV `av.open` when the operator probes metadata by itself.
`read_chunk_size`	`Optional[int]`	`None`	The number of bytes read at a time when the operator probes metadata by itself. The value cannot exceed 16 MiB.
`max_cached_blocks`	`Optional[int]`	`None`	The maximum number of blocks cached for each file. `0` disables block caching.
`read_ahead_blocks`	`Optional[int]`	`None`	The number of blocks to read ahead after a cache miss. `0` disables read-ahead.

When splitting by fixed duration, the operator generates closed-interval segments. For example, a 2-second video split by 1000 milliseconds produces [0, 999] and [1000, 1999]. When splitting by the target number of segments, the segment duration is rounded up, so the actual number of segments may be less than the target value.

Return type: A single UDTF column segment of type VIDEO_STRUCT_TYPE.

from pyflink.multimodal.operators import video_split

segments = df.join_lateral(
    video_split(
        col("uri"),
        segment_duration_ms=10_000
    ).alias("segment")
)

If you already have the results of video_metadata, pass them as a second input column:

with_metadata = df.with_column(
    "metadata",
    video_metadata(col("uri")),
)

segments = with_metadata.join_lateral(
    video_split(
        col("uri"),
        col("metadata"),
        segment_duration_ms=10_000
    ).alias("segment")
)

Frame Extraction

video_explode_frames

Extracts frames from a complete video or a video segment and expands the results into one frame per row.

Input type: DataType.string() or VIDEO_STRUCT_TYPE: the video URI column or the video segment reference column returned by video_split. A null input outputs no rows.

Function signature:

video_explode_frames(
    *columns,
    frame_selector="all_frames",
    sample_interval_ms=None,
    max_frames=None,
    image_height=None,
    image_width=None,
    on_error="raise",
    container_options=None,
    read_chunk_size=None,
    max_cached_blocks=None,
    read_ahead_blocks=None,
    concurrency=None
)

Parameter	Type	Default	Description
`frame_selector`	`str`	`"all_frames"`	The frame extraction strategy. Supported values: `"all_frames"`, `"keyframe"`, `"sample"`.
`sample_interval_ms`	`Optional[int]`	`None`	The sampling interval in `"sample"` mode, in milliseconds. The value must be greater than 0. `None` indicates 1000 milliseconds.
`max_frames`	`Optional[int]`	`None`	The maximum number of frames output for each input. The value must be greater than 0. `None` outputs all selected frames.
`image_height`	`Optional[int]`	`None`	The output frame height, in pixels. This parameter must be set together with `image_width` and must be greater than 0.
`image_width`	`Optional[int]`	`None`	The output frame width, in pixels. This parameter must be set together with `image_height` and must be greater than 0.
`on_error`	`str`	`"raise"`	`"raise"` throws an exception; `"skip"` outputs no rows for failed inputs. If an error occurs after some frames have been output, the rows already output are retained.
`container_options`	`Optional[Mapping[str, str]]`	`None`	Parameters passed to PyAV `av.open`.
`read_chunk_size`	`Optional[int]`	`None`	The number of bytes read from the Java file system bridge at a time. The value cannot exceed 16 MiB.
`max_cached_blocks`	`Optional[int]`	`None`	The maximum number of blocks cached for each file. `0` disables block caching.
`read_ahead_blocks`	`Optional[int]`	`None`	The number of blocks to read ahead after a cache miss. `0` disables read-ahead.

When the input is a video segment reference, the operator selects frames with known times within the closed interval [start_time_ms, end_time_ms]. "sample" establishes sampling time points from the start of the segment and selects the first frame whose time is not earlier than each sampling point.

Return type: Two UDTF columns: frame of type DataType.image() (the frame image) and frame_metadata of type VIDEO_FRAME_METADATA_TYPE.

from pyflink.multimodal.operators import video_explode_frames

frames = df.join_lateral(
    video_explode_frames(
        col("uri"),
        frame_selector="sample",
        sample_interval_ms=1_000,
        max_frames=300,
        image_height=360,
        image_width=640,
        on_error="skip"
    ).alias("frame", "frame_metadata")
)

video_extract_frames

Extracts frames from a complete video or a video segment and collects the results of the same input video into an array in a single row.For long videos or a large number of extracted frames, use video_explode_frames to avoid holding too many images in a single row.

Input type: DataType.string() (video URI column) or VIDEO_STRUCT_TYPE (the video segment reference column returned by video_split).

Function signature:

video_extract_frames(
    *columns,
    frame_selector="all_frames",
    sample_interval_ms=None,
    max_frames=None,
    image_height=None,
    image_width=None,
    on_error="raise",
    container_options=None,
    read_chunk_size=None,
    max_cached_blocks=None,
    read_ahead_blocks=None,
    concurrency=None
)

Parameter	Type	Default	Description
`frame_selector`	`str`	`"all_frames"`	The frame extraction strategy. Supported values: `"all_frames"`, `"keyframe"`, `"sample"`.
`sample_interval_ms`	`Optional[int]`	`None`	The sampling interval in `"sample"` mode, in milliseconds. The value must be greater than 0. `None` indicates 1000 milliseconds.
`max_frames`	`Optional[int]`	`None`	The maximum number of frames retained in the same output row. The value must be greater than 0. `None` retains all selected frames.
`image_height`	`Optional[int]`	`None`	The output image height. This parameter must be set together with `image_width` and must be greater than 0.
`image_width`	`Optional[int]`	`None`	The output image width. This parameter must be set together with `image_height` and must be greater than 0.
`on_error`	`str`	`"raise"`	`"raise"` throws an exception on error; `"null"` returns null for unreadable input.
`container_options`	`Optional[Mapping[str, str]]`	`None`	Parameters passed to PyAV `av.open`.
`read_chunk_size`	`Optional[int]`	`None`	The number of bytes read from the Java file system bridge at a time. The value cannot exceed 16 MiB.
`max_cached_blocks`	`Optional[int]`	`None`	The maximum number of blocks cached for each file. `0` disables block caching.
`read_ahead_blocks`	`Optional[int]`	`None`	The number of blocks to read ahead after a cache miss. `0` disables read-ahead.

If the input is null or processing fails under on_error="null", the operator returns None. If processing succeeds but no frame is selected, an empty array is returned.

Return type: DataType.list(DataType.image()).

from pyflink.multimodal.operators import video_extract_frames

# 1. Read directly from the URI and extract frames.
result = df.with_column(
    "sampled_frames",
    video_extract_frames(
        col("uri"),
        frame_selector="sample",
        sample_interval_ms=1_000,
        max_frames=60
    )
)

# 2. Build CLIP refs first, 10 seconds per clip.
clips = df.join_lateral(
    video_split(
        col("uri"),
        col("metadata"),
        segment_duration_ms=10_000,
        max_segments=1024,
        on_error="skip"
    ).alias("clip")
)

# Extract frames by segment using the CLIP refs.
result = clips.with_column(
    "sampled_frames",
    video_extract_frames(
        col("clip"),
        frame_selector="sample",
        sample_interval_ms=1_000,
        max_frames=60
    )
)

Complete Pipeline Example

The following example first reads the metadata, splits the video into 10-second segments, and then extracts one frame per second from each segment.

from pyflink.dataframe import col
from pyflink.multimodal.operators import (
    video_explode_frames,
    video_metadata,
    video_split
)

with_metadata = df.with_column(
    "metadata",
    video_metadata(
        col("uri"),
        on_error="null"
    )
).filter(col("metadata").is_not_null)

segments = with_metadata.join_lateral(
    video_split(
        col("uri"),
        col("metadata"),
        segment_duration_ms=10_000,
        max_segments=1024,
        on_error="skip"
    ).alias("segment")
)

frames = segments.join_lateral(
    video_explode_frames(
        col("segment"),
        frame_selector="sample",
        sample_interval_ms=1_000,
        max_frames=10,
        on_error="skip"
    ).alias("frame", "frame_metadata")
)

Limits

Operator Overview

Data Type Description

Video Segment Reference (VIDEO_STRUCT_TYPE)

Video Metadata (VIDEO_METADATA_TYPE)

Video Frame Metadata (VIDEO_FRAME_METADATA_TYPE)

Common Runtime Parameters

Video Metadata and Splitting

video_metadata

video_split

Frame Extraction

video_explode_frames

video_extract_frames

Complete Pipeline Example

Video Segment Reference (`VIDEO_STRUCT_TYPE`)

Video Metadata (`VIDEO_METADATA_TYPE`)

Video Frame Metadata (`VIDEO_FRAME_METADATA_TYPE`)