多模态算子

调用方式

每个算子支持两种调用风格：

工厂函数调用

先配置算子参数，再把算子应用到一个或多个列上。适合在多个列或多个步骤中复用同一配置。

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")))

直接调用

直接把输入列传给算子，并在同一步中设置参数。

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),
)

展开多行结果

video_split 和 video_explode_frames 会把一行输入展开为多行结果，需要通过 join_lateral 使用。

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")
)

通用参数

所有算子都可以通过 concurrency 控制并发处理数。部分模型类图像算子还支持批量大小和 GPU 资源参数。

model_sharing 支持以下取值：

图像输入输出

视频片段引用

定义：

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

该结构只描述视频片段位置，不复制视频内容。video_split 返回该结构，且它可以作为 video_explode_frames 或 video_extract_frames 的输入。

视频元数据

定义：

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

video_metadata 返回该结构。该结构常用于按分辨率、时长、编码格式过滤视频，也可传给 video_split，避免重复探测视频时长。

视频帧元数据

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 返回 DataType.image 和该元数据结构。保留该结构可以追踪每一帧来自哪个视频、哪个片段、哪个时间点。

图像变换与生成

图像检测与识别

图像向量与相似度

人脸处理

图像属性与过滤

图像质量评分

视频元数据、切分与抽帧

image_decode

将原始图像字节解码为DataType.image（Decoded image）。

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

输入类型：DataType.binary

返回类型：DataType.image

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

image_encode

将图像编码为压缩字节。

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

输入类型：DataType.image

返回类型：DataType.binary 或 DataType.string

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

image_compress

将编码图像压缩为编码字节。

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

输入类型：DataType.binary

返回类型：DataType.binary

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

image_convert_format

转换编码图像格式。

image_convert_format(*columns, format, concurrency=None)

输入类型：DataType.binary

返回类型：DataType.binary

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

image_convert_mode

将图像转换为目标色彩模式。

image_convert_mode(*columns, mode, concurrency=None)

支持的模式：

输入类型：DataType.image

返回类型：DataType.image

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

image_to_tensor

将图像转换为固定形状的 float32 张量（缩放 + 归一化到 [0, 1]）。

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

输入类型：DataType.image

返回类型： DataType.tensor

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

image_resize

将图像缩放到指定尺寸。

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

输入类型：DataType.image

返回类型：DataType.image

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

image_rescale

按比例缩放图像。

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

输入类型： DataType.image

返回类型： DataType.image

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

image_crop

裁剪图像区域。

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

输入类型：DataType.image

返回类型：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

检测并裁剪图像黑边。

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

输入类型：DataType.image

返回类型：DataType.image

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

image_flip

水平或垂直翻转图像。

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

输入类型：DataType.image

返回类型：DataType.image

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

image_blur

对图像应用模糊处理。

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

输入类型：DataType.image

返回类型：DataType.image

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

image_adjust_color

调整亮度、对比度和饱和度。

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

输入类型：DataType.image

返回类型：DataType.image

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

image_remove_background

移除图像背景

依赖： 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)

输入类型：DataType.image

返回类型：DataType.image

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

image_detect_objects

使用 YOLO 检测目标物体。

依赖： 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)

输入类型：DataType.image

返回类型：

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

字段说明：

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

image_segment

使用 FastSAM 生成语义分割掩码。

依赖： 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)

输入类型：DataType.image

返回类型：DataType.binary（PNG 编码的分割掩码，每个唯一像素值代表一个分割区域 ID）

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

image_ocr

使用 EasyOCR 从图像中提取文字。

依赖： pip install easyocr

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

lang支持以下语言代码：

多语言识别需要使用同一组兼容语言。en可以和多数语言一起使用；不要在同一个 lang中混用多个非英文语言组，例如不要同时设置中文和日文。

输入类型：DataType.image

返回类型：

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

字段说明：

每个元素代表一个检测到的文字区域。

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

image_detect_subplot

判断图像是否为拼接图、宫格图或 subplot，并返回估计子图数量。

依赖： pip install opencv-python

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

任意边小于 20 像素的图像始终返回 (False, 1)。

输入类型：DataType.image

返回类型：

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

字段说明：

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

image_embedding

从图像生成 CLIP 嵌入向量。

依赖： 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)

输入类型：DataType.image

返回类型：DataType.list(DataType.float32())（维度取决于模型，ViT-B/32 为 512 维）

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

image_text_similarity

计算图像与文本之间的 CLIP 余弦相似度。支持两种模式：

• 固定文本模式（text="..."）：所有图像与同一文本比较（单列 UDF）。

• 逐行文本模式（text=None）：每行图像与对应文本列配对（双列 UDF）。

依赖： 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)

输入类型：DataType.image

返回类型：DataType.float64（余弦相似度分数）

# 固定文本模式
sim = image_text_similarity(text="一张猫的照片")
df = df.with_column("score", sim(col("img")))

# 逐行文本模式
sim = image_text_similarity()
df = df.with_column("score", sim(col("img"), col("caption")))

image_face_detect

检测图像中的人脸边界框。

依赖： 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)

输入类型：DataType.image

返回类型：

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

字段说明：

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

image_face_count

统计图像中的人脸数量。

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)

参数：与 image_face_detect 相同。

输入类型：DataType.image

返回类型：DataType.int64（表示检测到的人脸数量）

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

image_face_blur

检测人脸并对其进行模糊处理。

依赖： 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)

其余参数与 image_face_detect 相同。

输入类型：DataType.image（原始图像）

返回类型：DataType.image（检测到的人脸区域被模糊处理后的原始图像）

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

is_valid_image

检查图像是否可以成功解码。

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

输入类型：DataType.binary（原始图像字节）或 DataType.image

返回类型：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

提取图像的宽度、高度、通道数和编码格式等元数据。

image_metadata(*columns, concurrency=None)

输入类型：DataType.image 或 DataType.binary

返回类型：

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

字段说明：

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

image_aspect_ratio

计算宽高比（width / height）。

image_aspect_ratio(*columns, concurrency=None)

输入类型：DataType.image 或 DataType.binary

返回类型：DataType.float64

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

image_sharpness

使用拉普拉斯方差计算图像清晰度。值越高表示图像越清晰，值越低表示图像越模糊。常用于数据清洗中过滤模糊图片。

与 image_quality_score 的区别：image_quality_score 返回归一化的 [0, 1] 综合评分（清晰度 50% + 对比度 30% + 色彩丰富度 20%），而 image_sharpness 返回原始拉普拉斯方差（无上界），数值大小取决于图像内容。

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

输入类型：DataType.image

返回类型：DataType.float64（拉普拉斯方差，无上界，值越高越清晰）

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

image_hash

计算图像的哈希值。

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

输入类型：DataType.image

返回类型：DataType.string（十六进制编码的哈希值）

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

image_size_filter

按像素尺寸过滤图像。

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

输入类型：DataType.image 或 DataType.binary

返回类型：DataType.boolean（图像满足所有约束时返回 True）

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

image_shape_filter

按宽高比（width / height）过滤图像。

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

输入类型：DataType.image 或 DataType.binary

返回类型：DataType.boolean

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

image_file_size_filter

按编码文件大小（字节）过滤图像。

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

输入类型：DataType.binary

返回类型：DataType.boolean

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

image_quality_score

计算综合质量评分，融合清晰度、对比度和色彩丰富度。

算法：拉普拉斯方差（清晰度，权重 50%）+ 灰度标准差（对比度，权重 30%）+ 通道标准差（色彩丰富度，权重 20%）的加权组合。

image_quality_score(*columns, concurrency=None)

输入类型：DataType.image

返回类型：DataType.float64（分数范围 [0, 1]，值越高质量越好）

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

image_nsfw_score

使用图像分类模型评估 NSFW 风险。

依赖： 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)

输入类型：DataType.image

返回类型：DataType.float64（概率范围 [0, 1]，值越高 NSFW 可能性越大）

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

image_aesthetic_score

使用美学评分模型评估图像美学质量。

依赖： 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)

输入类型：DataType.image

返回类型：DataType.float64

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

image_watermark_score

使用图像分类模型评估水印风险。

依赖： 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)

输入类型：DataType.image

返回类型：DataType.float64（概率范围 [0, 1]，值越高越可能含水印）

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

video_metadata

读取视频基本信息，例如分辨率、帧率、时长和编码格式。

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

输入类型：DataType.string（视频 URI）

返回类型：

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

字段含义见：视频元数据

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

video_split

将完整视频或已有视频片段切分为多个片段引用，展开为一行一个片段引用（UDTF）。该算子本身不实际读取视频帧，只计算时间窗口。

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 和 num_segments 二选一，表示两种不同的切分方式：按时间区间切分、按数量切分。

输入类型：DataType.string（视频 URI）/ 视频片段引用

返回类型：视频片段引用

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

字段含义见：视频片段引用

算子会自动探测视频时长。如果已经通过 video_metadata 获取到了元数据，为了避免额外探测，可以把 video_metadata 的结果作为第二列传入。

segments = df.join_lateral(
    video_split(segment_duration_ms=1000)(
        col("uri"),
        col("video_meta"), # 可选列
    ).alias("segment")
)

video_explode_frames

从完整视频或视频片段中抽帧，并将结果展开为一帧一行（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)

输入类型：DataType.string或视频片段引用

返回两列： 第一列DataType.image（解码后的 RGB 视频帧），第二列视频帧元数据如下

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

字段含义见：视频帧元数据

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

从完整视频或视频片段中抽帧，并把同一个输入视频的结果收集到同一行的数组中。

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)

参数含义与 video_explode_frames 相同。

输入类型：DataType.string / 视频片段引用

返回类型：DataType.array(DataType.image())

适合下游逻辑需要一次性拿到同一视频的多个帧。若视频较长或抽帧数量较多，推荐使用 video_explode_frames，避免单行持有过多图像。

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,
    )
)