Multimodal Operators

Invocation Methods

Each operator supports two invocation styles:

Factory Function Call

Configure operator parameters first, then apply the operator to one or more columns. Suitable for reusing the same configuration across multiple columns or steps.

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

resize = image_resize(width=512, height=512)
df = df.with_column("resized", resize(col("image")))

Direct Call

Pass input columns directly to the operator and set parameters in the same step.

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

df = df.with_column(
    "resized",
    image_resize(col("image"), width=512, height=512),
)

Expanding Multi-Row Results

video_split and video_explode_frames expand a single input row into multiple output rows. Use join_lateral to invoke them.

from pyflink.dataframe import col
from pyflink.multimodal.operators import video_explode_frames, video_split

segments = df.join_lateral(
    video_split(segment_duration_ms=1000)(col("uri")).alias("segment")
)

frames = segments.join_lateral(
    video_explode_frames(frame_selector="sample", sample_interval_ms=1000)(
        col("segment")
    ).alias("frame", "metadata")
)

Common Parameters

All operators support concurrency to control parallel processing. Some model-based image operators also support batch size and GPU resource parameters.

model_sharing supports the following values:

Image Input/Output

Video Segment Reference

Definition:

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

This struct only describes the video segment position without copying video content. video_split returns this struct, and it can be used as input to video_explode_frames or video_extract_frames .

Video Metadata

Definition:

DataType.struct(
  {
    "width": DataType.int32(),
    "height": DataType.int32(),
    "fps": DataType.float64(),
    "duration_ms": DataType.int64(),
    "frame_count": DataType.int64(),

    "codec_name": DataType.string(),
    "video_stream_index": DataType.int32()
  }
)

video_metadata returns this struct. It is commonly used to filter videos by resolution, duration, or encoding format, and can also be passed to video_split to avoid repeated duration probing.

Video Frame Metadata

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()
  }
)

video_explode_frames returns DataType.image and this metadata struct. Retaining this struct enables tracking which video, segment, and timestamp each frame originates from.

Image Transformation and Generation

Image Detection and Recognition

Image Embedding and Similarity

Face Processing

Image Attributes and Filtering

Image Quality Scoring

Video Metadata, Splitting, and Frame Extraction

image_decode

Decode raw image bytes to DataType.image (decoded image).

image_decode(*columns, on_error="raise", mode=None, pixel_limit=None, concurrency=None)

Input type: DataType.binary

Return type: DataType.image

df = df.with_column("img", image_decode(col("raw_bytes")))

image_encode

Encode an image to compressed bytes.

image_encode(*columns, format=None, quality=85, output="bytes", concurrency=None)

Input type: DataType.image

Return type: DataType.binary or DataType.string

df = df.with_column("jpeg", image_encode(col("image"), format="JPEG", quality=90))

image_compress

Compress an encoded image to encoded bytes.

image_compress(*columns, quality=85, format="JPEG", concurrency=None)

Input type: DataType.binary

Return type: DataType.binary

df = df.with_column("compressed", image_compress(col("img"), quality=60))

image_convert_format

Convert the encoded image format.

image_convert_format(*columns, format, concurrency=None)

Input type: DataType.binary

Return type: DataType.binary

df = df.with_column("png", image_convert_format(col("img"), format="PNG"))

image_convert_mode

Convert an image to the target color mode.

image_convert_mode(*columns, mode, concurrency=None)

Supported modes:

Input type: DataType.image

Return type: DataType.image

df = df.with_column("gray", image_convert_mode(col("img"), mode="L"))

image_to_tensor

Convert an image to a fixed-shape float32 tensor (resizing + normalization to [0, 1]).

image_to_tensor(*columns, width, height, mode="RGB", layout="CHW", concurrency=None)

Input type: DataType.image

Return type: DataType.tensor

df = df.with_column("tensor", image_to_tensor(col("img"), width=224, height=224))

image_resize

Resize an image to specified dimensions.

image_resize(*columns, width, height, method="lanczos", concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("thumb", image_resize(col("img"), width=256, height=256))

image_rescale

Rescale an image proportionally.

image_rescale(*columns, scale, method="lanczos", concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("half", image_rescale(col("img"), scale=0.5))

image_crop

Crop an image region.

image_crop(*columns, crop_coords=None, crop_type="center", crop_ratio=(0.8, 0.8),
           crop_size=None, concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("cropped", image_crop(col("img"), crop_ratio=(0.9, 0.9)))
df = df.with_column("roi", image_crop(col("img"), crop_coords=(10, 10, 200, 200)))

image_crop_black_border

Detect and crop black borders from an image.

image_crop_black_border(*columns, threshold=None, detect_algorithm="auto",
                        black_threshold=None, edge_sensitivity=1.0,
                        min_border_size=1, concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("clean", image_crop_black_border(col("img")))

image_flip

Flip an image horizontally or vertically.

image_flip(*columns, mode="horizontal", concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("flipped", image_flip(col("img"), mode="vertical"))

image_blur

Apply blur processing to an image.

image_blur(*columns, radius=2, blur_type="gaussian", concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("blurred", image_blur(col("img"), radius=5))

image_adjust_color

Adjust brightness, contrast, and saturation.

image_adjust_color(*columns, brightness=1.0, contrast=1.0,saturation=1.0, concurrency=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("bright", image_adjust_color(col("img"), brightness=1.5))

image_remove_background

Remove image background

Dependency: pip install rembg

image_remove_background(*columns, alpha_matting=False,
                        alpha_matting_foreground_threshold=240,
                        alpha_matting_background_threshold=10,
                        alpha_matting_erode_size=10,
                        bgcolor=None, model_sharing=None,
                        concurrency=None, batch_size=None,
                        num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.image

df = df.with_column("fg", image_remove_background(col("image")))

image_detect_objects

Detect objects using YOLO.

Dependency: pip install ultralytics

image_detect_objects(*columns, model="yolov8n", confidence=0.05, imgsz=640,
                     iou=0.5, model_sharing=None, concurrency=None,
                     batch_size=None, num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type:

DataType.list(
    DataType.struct(
    {
        "label": DataType.string(),
        "x": DataType.float64(),
        "y": DataType.float64(),
        "w": DataType.float64(),
        "h": DataType.float64(),
        "confidence": DataType.float64(),
    }
)

Field description:

objs = image_detect_objects(confidence=0.25)
df = df.with_column("objects", objs(col("img")))

image_segment

Generate semantic segmentation masks using FastSAM.

Dependency: pip install ultralytics

image_segment(*columns, model="FastSAM-x.pt", imgsz=1024, confidence=0.05,
              iou=0.5, model_sharing=None, concurrency=None,
              batch_size=None, num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.binary (PNG-encoded segmentation mask where each unique pixel value represents a segment ID)

df = df.with_column("mask", image_segment(col("img")))

image_ocr

Extract text from images using EasyOCR.

Dependency: pip install easyocr

image_ocr(*columns, lang=None, model_sharing=None, concurrency=None,
          batch_size=None, num_gpus=None, gpu_type=None)

lang supports the following language codes:

Multilingual recognition requires compatible languages from the same group. en can be used with most languages. Do not mix multiple non-English language groups in the same lang, e.g., do not set both Chinese and Japanese simultaneously.

Input type: DataType.image

Return type:

DataType.list(
  DataType.struct(
    {
      "text": DataType.string(),
      "confidence": DataType.float64(),
      "bbox": DataType.list(DataType.list(DataType.float64())),
    }
  )
)

Field description:

Each element represents a detected text region.

ocr = image_ocr(lang=["en", "ch_sim"])
df = df.with_column("text", ocr(col("img")))

image_detect_subplot

Determine whether an image is a collage, grid, or subplot, and return the estimated number of sub-images.

Dependency: pip install opencv-python

image_detect_subplot(*columns, threshold=0.5, concurrency=None)

Images with any side smaller than 20 pixels always return (False, 1).

Input type: DataType.image

Return type:

DataType.struct(
  {
    "is_subplot": DataType.boolean(),
    "count": DataType.int32(),
  }
)

Field description:

df = df.with_column("subplot", image_detect_subplot(col("img")))
df = df.filter(image_detect_subplot(col("img"))["is_subplot"])

image_embedding

Generate CLIP embeddings from images.

Dependency: pip install open_clip_torch torch

image_embedding(*columns, model="ViT-B/32", pretrained="openai",
                model_sharing=None, concurrency=None, batch_size=None,
                num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.list(DataType.float32()) (dimensions depend on the model;ViT-B/32 produces 512 dimensions)

df = df.with_column("vector", image_embedding(col("img")))

image_text_similarity

Compute CLIP cosine similarity between images and text. Supports two modes:

• Fixed text mode (text="..."): All images are compared against the same text (single-column UDF).

• Per-row text mode (text=None): Each image is paired with the corresponding text column (two-column UDF).

Dependency: pip install open_clip_torch torch

image_text_similarity(*columns, text=None, model="ViT-B/32", pretrained="openai",
                      model_sharing=None, concurrency=None, batch_size=None,
                      num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.float64 (cosine similarity score)

# Fixed text mode
sim = image_text_similarity(text="a photo of a cat")
df = df.with_column("score", sim(col("img")))

# Per-row text mode
sim = image_text_similarity()
df = df.with_column("score", sim(col("img"), col("caption")))

image_face_detect

Detect face bounding boxes in an image.

Dependency: pip install opencv-python

image_face_detect(*columns, cv_classifier="haarcascade_frontalface_alt.xml",
                  confidence=0.5, scale_factor=1.1, min_size=None,
                  max_size=None, concurrency=None)

Input type: DataType.image

Return type:

DataType.list(
  DataType.struct(
    {
      "x": DataType.int32(),
      "y": DataType.int32(),
      "w": DataType.int32(),
      "h": DataType.int32(),
      "confidence": DataType.float64(),
    }
  )
)

Field description:

df = df.with_column("faces", image_face_detect(confidence=0.6)(col("img")))

image_face_count

Count the number of faces in an image.

image_face_count(*columns, cv_classifier="haarcascade_frontalface_alt.xml",
                 confidence=0.5, scale_factor=1.1, min_size=None,
                 max_size=None, concurrency=None)

Parameters: Same as image_face_detect.

Input type: DataType.image

Return type: DataType.int64 (number of detected faces)

df = df.with_column("n_faces", image_face_count()(col("img")))
df = df.filter(image_face_count(col("img")) > 0)

image_face_blur

Detect faces and apply blur processing to them.

Dependency: pip install opencv-python

image_face_blur(*columns, cv_classifier="haarcascade_frontalface_alt.xml",
                blur_type="gaussian", radius=2,
                scale_factor=1.1, min_size=None, max_size=None,
                concurrency=None)

Remaining parameters are the same as image_face_detect .

Input type: DataType.image (raw image)

Return type: DataType.image (raw image with detected face regions blurred)

df = df.with_column("safe", image_face_blur(col("img"), radius=15))

is_valid_image

Check whether an image can be successfully decoded.

is_valid_image(*columns, mode=None, pixel_limit=None, concurrency=None)

Input type: DataType.binary(raw image bytes) or DataType.image

Return type: DataType.bool

df = df.with_column("ok", is_valid_image()(col("img")))
df = df.filter(is_valid_image(col("img"), pixel_limit=4096*4096))

image_metadata

Extract image metadata such as width, height, channels, and encoding format.

image_metadata(*columns, concurrency=None)

Input type: DataType.image or DataType.binary

Return type:

DataType.struct(
  {
    "width": DataType.int32(),
    "height": DataType.int32(),
    "channels": DataType.int32(),
    "mode": DataType.string(),
    "format": DataType.string(),
  }
)

Field description:

df = df.with_column("meta", image_metadata(col("img")))

image_aspect_ratio

Compute aspect ratio (width / height).

image_aspect_ratio(*columns, concurrency=None)

Input type: DataType.image or DataType.binary

Return type: DataType.float64

df = df.with_column("ratio", image_aspect_ratio(col("img")))

image_sharpness

Compute image sharpness using Laplacian variance. Higher values indicate sharper images, lower values indicate blurrier images. Commonly used in data cleaning to filter blurry images.

Difference from image_quality_score: image_quality_score returns a normalized [0, 1] composite score (sharpness 50% + contrast 30% + color richness 20%), while image_sharpness returns the raw Laplacian variance (unbounded) whose magnitude depends on image content.

image_sharpness(*columns, max_edge=None, allow_upscale=False, concurrency=None)

Input type: DataType.image

Return type: DataType.float64 (Laplacian variance, unbounded, higher values indicate sharper images)

df = df.filter(image_sharpness(col("img"), max_edge=512) > 100)

image_hash

Compute the hash value of an image.

image_hash(*columns, method="phash", concurrency=None)

Input type: DataType.image

Return type: DataType.string (hex-encoded hash value)

df = df.with_column("hash", image_hash(col("image"), method="phash"))

image_size_filter

Filter images by pixel dimensions.

image_size_filter(*columns, min_w=None, min_h=None, max_w=None, max_h=None, concurrency=None)

Input type: DataType.image or DataType.binary

Return type: DataType.boolean (returns True when the image satisfies all constraints)

df = df.filter(image_size_filter(col("img"), min_w=256, min_h=256))

image_shape_filter

Filter images by aspect ratio (width / height).

image_shape_filter(*columns, min_ratio=None, max_ratio=None, concurrency=None)

Input type: DataType.image or DataType.binary

Return type: DataType.boolean

df = df.filter(image_shape_filter(col("img"), min_ratio=0.5, max_ratio=2.0))

image_file_size_filter

Filter images by encoded file size (bytes).

image_file_size_filter(*columns, min_bytes=None, max_bytes=None, concurrency=None)

Input type: DataType.binary

Return type: DataType.boolean

df = df.filter(image_file_size_filter(col("img"), max_bytes=5*1024*1024))

image_quality_score

Compute a composite quality score combining sharpness, contrast, and color richness.

Algorithm: weighted combination of Laplacian variance (sharpness, weight 50%) + grayscale standard deviation (contrast, weight 30%) + channel standard deviation (color richness, weight 20%).

image_quality_score(*columns, concurrency=None)

Input type: DataType.image

Return type: DataType.float64 (score range [0, 1], higher values indicate better quality)

df = df.with_column("quality", image_quality_score(col("img")))
df = df.filter(image_quality_score(col("img")) > 0.3)

image_nsfw_score

Assess NSFW risk using an image classification model.

Dependency: pip install transformers torch

image_nsfw_score(*columns, hf_nsfw_model="Falconsai/nsfw_image_detection",
                 model_sharing=None, concurrency=None, batch_size=None,
                 num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.float64 (probability range [0, 1], higher values indicate higher NSFW likelihood)

df = df.with_column("nsfw", image_nsfw_score(col("img")))
df = df.filter(image_nsfw_score(col("img")) < 0.5)

image_aesthetic_score

Assess image aesthetic quality using an aesthetic scoring model.

Dependency: pip install transformers torch

image_aesthetic_score(*columns,
                      hf_scorer_model="shunk031/aesthetics-predictor-v2-sac-logos-ava1-l14-linearMSE",
                      model_sharing=None, concurrency=None, batch_size=None,
                      num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.float64

df = df.with_column("aesthetic", image_aesthetic_score(col("img")))

image_watermark_score

Assess watermark risk using an image classification model.

Dependency: pip install transformers torch

image_watermark_score(*columns, hf_watermark_model="amrul-hzz/watermark_detector",
                      model_sharing=None, concurrency=None, batch_size=None,
                      num_gpus=None, gpu_type=None)

Input type: DataType.image

Return type: DataType.float64 (probability range [0, 1], higher values indicate higher watermark likelihood)

df = df.with_column("wm", image_watermark_score()(col("img")))
df = df.filter(image_watermark_score(col("img")) < 0.8)

video_metadata

Read basic video information such as resolution, frame rate, duration, and encoding format.

video_metadata(*columns, on_error="raise", container_options=None)

Input type: DataType.string(video URI)

Return type:

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()
  }
)

For field definitions, see Video Metadata

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

video_split

Split a complete video or existing video segment into multiple segment references, expanding to one segment reference per row (UDTF). This operator does not actually read video frames; it only computes time windows.

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)

segment_duration_ms and num_segments are mutually exclusive, representing two splitting strategies: split by time interval, or split by equal segment count.

Input type: DataType.string(video URI) / video segment reference

Return type: Video segment reference

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

For field definitions, see Video Segment Reference

The operator automatically probes video duration. If metadata has already been obtained via video_metadata, pass the video_metadata result as a second column to avoid redundant probing.

segments = df.join_lateral(
    video_split(segment_duration_ms=1000)(
        col("uri"),
        col("video_meta"), # optional column
    ).alias("segment")
)

video_explode_frames

Extract frames from a complete video or video segment, expanding results to one frame per row (UDTF).

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)

Input type: DataType.string or video segment reference

Returns two columns: The first column is DataType.image (decoded RGB video frame); the second column contains video frame metadata 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()
  }
)

For field definitions, see Video Frame Metadata

frames = segments.join_lateral(
    video_explode_frames(
        frame_selector="sample",
        sample_interval_ms=1000,
        max_frames=1,
        image_width=512,
        image_height=512,
    )(col("segment")).alias("frame", "metadata")
)

video_extract_frames

Extract frames from a complete video or video segment, collecting results from the same input video into arrays within a single row.

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)

Parameters are the same as video_explode_frames.

Input type: DataType.string / video segment reference

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

Suitable when downstream logic requires all frames from the same video at once. For long videos or large frame counts, use video_explode_frames to avoid holding too many images in a single row.

df = df.with_column(
    "frames",
    video_extract_frames(
        col("uri"),
        frame_selector="sample",
        sample_interval_ms=1000,
        max_frames=8,
        image_width=512,
        image_height=512,
    )
)