Multimodal Expressions#

PyFlink DataFrame expressions provide image, video, and audio namespaces for invoking multimodal operations with the expression on the left side.

Examples:

>>> from pyflink.dataframe import col
>>> df = df.with_column(
...     "thumbnail", col("img").image.resize(width=224, height=224))
>>> df = df.with_column(
...     "frames", col("video_uri").video.extract_frames(max_frames=16))
>>> speech_ready = (
...     col("audio_bytes").audio.decode().audio.standardize())
>>> df = df.with_column(
...     "transcript", speech_ready.audio.asr_whisper())

Image Accessor#

Use the image namespace on DataFrame expressions to call image operations.

ImageExpressionAccessor(expression)

Image operations available on a DataFrame expression.

Image Transform#

Image transformation operations.

.image.decode() accepts encoded image bytes and returns a decoded image. Pixel transforms accept decoded images and return decoded images so chained operations can avoid repeated JPEG/PNG encode-decode cycles. .image.encode() accepts decoded images and returns encoded bytes; .image.compress() and .image.convert_format() accept encoded bytes and return encoded bytes.

Typical pipeline:

>>> from pyflink.dataframe import col
>>> df = df.with_column(
...     "jpg",
...     col("image_url").fetch_content()
...     .image.decode()
...     .image.resize(width=512, height=512)
...     .image.blur(radius=2)
...     .image.encode(format="JPEG"))

ImageExpressionAccessor.decode(*[, ...])

Decode image bytes into image values.

ImageExpressionAccessor.encode(*[, format, ...])

Encode image values into bytes or data URLs.

ImageExpressionAccessor.compress(*[, ...])

Compress image values using the requested format and quality.

ImageExpressionAccessor.convert_format(*, format)

Convert image values to a different encoded image format.

ImageExpressionAccessor.to_tensor(*, width, ...)

Convert already prepared image values to fixed-shape tensors.

ImageExpressionAccessor.convert_mode(*, mode)

Convert image values to another image mode.

ImageExpressionAccessor.resize(*, width, height)

Resize image values to a fixed width and height.

ImageExpressionAccessor.rescale(*, scale[, ...])

Scale image values by a ratio.

ImageExpressionAccessor.crop(*[, ...])

Crop image values by coordinates, crop type, ratio, or size.

ImageExpressionAccessor.crop_black_border(*)

Remove black borders from image values.

ImageExpressionAccessor.flip(*[, mode, ...])

Flip image values horizontally or vertically.

ImageExpressionAccessor.blur(*[, radius, ...])

Blur image values.

ImageExpressionAccessor.adjust_color(*[, ...])

Adjust image brightness, contrast, and saturation.

ImageExpressionAccessor.remove_background(*)

Remove image backgrounds and optionally composite a new background.

Image Information#

Image information extraction operations.

Most operations in this section take decoded Image values and return numeric values, strings, or booleans - they do not modify the image. Lightweight metadata and filter operations may also inspect encoded bytes when documented.

Without business parameters:

>>> from pyflink.dataframe import col
>>> df = df.with_column("ratio", col("img").image.aspect_ratio())

With business parameters:

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

The concurrency argument is forwarded to the DataFrame UDF runtime.

ImageExpressionAccessor.metadata(*[, ...])

Extract image metadata.

ImageExpressionAccessor.aspect_ratio(*[, ...])

Compute image aspect ratio.

ImageExpressionAccessor.sharpness(*[, ...])

Estimate image sharpness.

ImageExpressionAccessor.hash(*[, method, ...])

Compute an image perceptual hash.

ImageExpressionAccessor.is_valid(*[, mode, ...])

Check whether image inputs can be decoded as valid images.

ImageExpressionAccessor.size_filter(*[, ...])

Filter images by width and height bounds.

ImageExpressionAccessor.shape_filter(*[, ...])

Filter images by aspect ratio.

ImageExpressionAccessor.file_size_filter(*)

Filter image files by byte size.

Image Detection#

Detection and recognition operations.

Operations in this section detect or recognize structured content within images: objects, text, segments, and subplots.

Model-based operations (detect_objects(), segment(), and ocr()) are registered as pandas batch UDFs, so Flink batches rows automatically. detect_objects() returns axis-aligned boxes x/y/w/h; ocr() returns four-point text polygons; segment() returns indexed PNG mask bytes; and detect_subplot() returns subplot metadata.

Example:

>>> from pyflink.dataframe import col
>>> df = df.with_column(
...     "objects",
...     col("img").image.detect_objects(
...         model="yolov8n", confidence=0.05))
>>> df = df.with_column(
...     "text", col("img").image.ocr(lang=["en", "ch_sim"]))

Model-based pandas operations forward keyword-only concurrency, batch_size, num_gpus, and gpu_type to the DataFrame UDF runtime. detect_subplot() forwards concurrency.

ImageExpressionAccessor.detect_objects(*[, ...])

Detect objects in image values.

ImageExpressionAccessor.segment(*[, model, ...])

Segment objects or regions in image values.

ImageExpressionAccessor.ocr(*[, lang, ...])

Run OCR on image values.

ImageExpressionAccessor.detect_subplot(*[, ...])

Detect subplot regions in image values.

Image Face#

Face processing operations.

Operations in this section detect, count, or blur faces in image inputs with OpenCV Haar cascade classifiers.

Requires: pip install opencv-python-headless. face_blur() also requires pip install Pillow.

The concurrency argument is forwarded to the DataFrame UDF runtime.

Usage:

>>> from pyflink.dataframe import col
>>> df = df.with_column(
...     "faces", col("img").image.face_detect(min_neighbors=3))
>>> df = df.with_column(
...     "face_count", col("img").image.face_count(min_neighbors=3))
>>> df = df.with_column(
...     "blurred",
...     col("img").image.face_blur(radius=15, min_neighbors=3))

ImageExpressionAccessor.face_detect(*[, ...])

Detect faces in image values.

ImageExpressionAccessor.face_count(*[, ...])

Count faces in image values.

ImageExpressionAccessor.face_blur(*[, ...])

Blur detected faces in image values.

Image Quality#

Image quality and safety assessment operations.

Operations in this section score images based on quality, safety, and watermark detection. All operations return float scores.

Example:

>>> from pyflink.dataframe import col
>>> df = df.with_column(
...     "quality", col("img").image.quality_score())
>>> df = df.with_column(
...     "nsfw", col("img").image.nsfw_score())

Pandas batch UDFs accept concurrency, batch_size, num_gpus, and gpu_type. Row-level UDFs accept concurrency.

Null image inputs produce None outputs. These operations accept decoded images; use col("raw_bytes").image.decode(on_error="null") when processing untrusted raw bytes. col("raw_bytes").image.is_valid() can be used as a lightweight pre-filter but does not guarantee full pixel materialization.

ImageExpressionAccessor.nsfw_score(*[, ...])

Score images for NSFW content.

ImageExpressionAccessor.aesthetic_score(*[, ...])

Score images for aesthetic quality.

ImageExpressionAccessor.quality_score(*[, ...])

Score general image quality.

ImageExpressionAccessor.watermark_score(*[, ...])

Score images for watermark likelihood.

Image Embedding#

Image embedding and similarity operations.

Operations in this section generate vector embeddings from images or compute image-text similarity scores using CLIP. They use pandas batch UDFs and run CLIP inference once per input batch.

Example:

>>> from pyflink.dataframe import col
>>> df = df.with_column(
...     "vector", col("img").image.embedding(model="ViT-B/32"))
>>> # Fixed text mode - compare all images against one prompt.
>>> df = df.with_column(
...     "score",
...     col("img").image.text_similarity(text="a photo of a cat"))
>>> # Per-row text mode - each row has its own text.
>>> df = df.with_column(
...     "score",
...     col("img").image.text_similarity(text_column=col("text")))

concurrency, batch_size, num_gpus, and gpu_type are forwarded to the DataFrame UDF runtime.

Requires: pip install open_clip_torch torch pillow.

ImageExpressionAccessor.embedding(*[, ...])

Compute image embeddings.

ImageExpressionAccessor.text_similarity(*[, ...])

Compute similarity between image values and text.

Video Accessor#

Use the video namespace on DataFrame expressions to call video operations. These are video frame operations for extraction pipelines.

VideoExpressionAccessor(expression)

Video operations available on a DataFrame expression.

VideoExpressionAccessor.metadata(*[, ...])

Probe video metadata for this expression.

VideoExpressionAccessor.extract_frames(*[, ...])

Extract selected frames from this expression as an array.

Audio Accessor#

Use the audio namespace on DataFrame expressions to call audio operations.

AudioExpressionAccessor(expression)

Audio operations available on a DataFrame expression.

Audio Information#

Audio information, validation, and filter operations.

These operations inspect audio objects and return scalar values or row values. They do not modify waveform content. Encoded inputs are accepted only at the pipeline boundary; waveform-only operations require an explicit .audio.decode() step so that pipeline definitions stay predictable.

AudioExpressionAccessor.is_valid(*[, ...])

Check whether this expression contains valid audio.

AudioExpressionAccessor.metadata(*[, ...])

Probe audio metadata for this expression.

AudioExpressionAccessor.duration(*[, ...])

Return audio duration in seconds for this expression.

AudioExpressionAccessor.duration_filter(*[, ...])

Check whether audio duration is within second bounds.

AudioExpressionAccessor.size_filter(*[, ...])

Check whether encoded audio size is within byte bounds.

AudioExpressionAccessor.silence_detection(*, ...)

Detect low-amplitude silence ranges in waveform values.

AudioExpressionAccessor.detect_speech(*[, ...])

Detect speech activity ranges in waveform values.

Audio Transform#

Audio decode, encode, split, and waveform transform operations.

The transform layer separates encoded boundary inputs from decoded waveform processing. Operations that modify samples accept AUDIO_WAVEFORM only, so a pipeline should usually call .audio.decode() once, transform waveforms, and then call .audio.encode() or a sink-specific writer at the boundary.

AudioExpressionAccessor.decode(*[, ...])

Decode encoded audio into waveform rows.

AudioExpressionAccessor.encode(*[, format, ...])

Encode waveform rows to audio bytes.

AudioExpressionAccessor.convert_format(*[, ...])

Convert encoded audio to another encoded format.

AudioExpressionAccessor.standardize(*[, ...])

Standardize waveform sample rate and channel count.

AudioExpressionAccessor.resample(*, sample_rate)

Resample waveform values to a target sample rate.

AudioExpressionAccessor.split_by_duration(*, ...)

Split audio by fixed duration.

AudioExpressionAccessor.split_by_timestamp(...)

Split audio by explicit timestamp ranges.

AudioExpressionAccessor.split_by_speech(...)

Split audio using caller-provided speech activity ranges.

AudioExpressionAccessor.concat(*[, concurrency])

Concatenate an array of waveform rows.

Audio Speech#

Speech model operations for decoded audio waveforms.

These are model-backed pandas UDFs. They accept standardized AUDIO_WAVEFORM values only: callers should decode boundary inputs and normalize to 16 kHz mono before invoking Whisper-based operations.

AudioExpressionAccessor.asr_whisper(*[, ...])

Run Whisper ASR over waveform values.

AudioExpressionAccessor.detect_language(*[, ...])

Detect likely spoken language candidates for waveform values.