图像属性获取与过滤-阿里云帮助中心

本文介绍 Python DataFrame API 中的图像有效性检查、元数据提取、清晰度计算、感知哈希和过滤算子。

使用前须知

仅实时计算引擎 VVR 11.8 及以上版本支持。
编码图片的校验、元数据和尺寸过滤只读取容器头，上述这些轻量检查不能保证完整像素流一定可以解码。

算子清单

分类	算子	说明
有效性检查	is_valid_image	判断图片是否通过有效性检查。
属性提取	image_metadata	提取图像宽度、高度、通道数、模式等。
	image_aspect_ratio	计算宽高比。
	image_sharpness	使用拉普拉斯方差计算图像清晰度。
	image_hash	计算图像 phash。
数据过滤	image_size_filter	按图像像素尺寸过滤。
	image_shape_filter	按图像宽高比过滤。
	image_file_size_filter	按图像文件大小过滤。

通用 Runtime 参数

函数签名保留各算子实际支持的 Runtime 参数。为避免重复，算子参数表只说明业务参数，Runtime 参数统一说明如下。

参数	类型	默认值	适用范围	说明
`concurrency`	`Optional[int]`	`None`	本页全部算子	UDF 并发度。`None` 表示使用框架默认值。

有效性检查与属性提取

is_valid_image

判断输入图片是否通过轻量级有效性检查，并可检查已解码图片能否转换为指定的 mode。

输入类型： DataType.binary() 或 DataType.image()，编码图片字节列或解码后的图像列。

函数签名：

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

参数	类型	默认值	说明
`mode`	`Optional[str]`	`None`	可选目标模式。设置后检查输入能否转换为该模式。
`pixel_limit`	`Optional[int]`	`None`	最大像素数 `width * height`。

空值、损坏输入、超过 pixel_limit 或无法转换为目标模式时返回 False。

返回类型： DataType.boolean()。

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

result = df.filter(
    is_valid_image(
        col("image_bytes"),
        pixel_limit=4096 * 4096
    )
)

image_metadata

提取图像宽度、高度、通道数、模式和编码格式。

输入类型： DataType.binary() 或 DataType.image()，编码图片字节列或解码后的图像列。

函数签名：

image_metadata(
    *columns,
    concurrency=None
)

编码图片通过容器头轻量读取；空值或损坏输入返回 None。解码图像不保留原文件容器格式，因此 format 返回 "UNKNOWN"。

返回类型：

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

from pyflink.multimodal.operators import image_metadata

result = df.with_column(
    "metadata",
    image_metadata(col("image_bytes"))
)

image_aspect_ratio

计算宽高比 width / height。

输入类型： DataType.binary() 或 DataType.image()，同时支持处理编码图片字节列或解码后的图像列。

函数签名：

image_aspect_ratio(
    *columns,
    concurrency=None
)

无效或损坏输入返回 None。

返回类型： DataType.float64()。

from pyflink.multimodal.operators import image_aspect_ratio

result = df.with_column(
    "aspect_ratio",
    image_aspect_ratio(col("image"))
)

image_sharpness

使用拉普拉斯方差计算图像清晰度，值越高通常表示边缘越清晰、细节越丰富，值越低通常表示图像越模糊。

输入类型： DataType.image()，解码后的图像列。

函数签名：

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

参数	类型	默认值	说明
`max_edge`	`Optional[int]`	`None`	计算前允许的最长边像素数。`None` 表示保持原尺寸。
`allow_upscale`	`bool`	`False`	设置 `max_edge` 后是否允许放大小图。

返回值范围为 [0, +∞)。该结果只表示清晰度，不是综合图像质量分。比较不同图片时，应使用相同的 max_edge 和 allow_upscale 配置。

返回类型： DataType.float64()。

from pyflink.multimodal.operators import image_sharpness

result = df.with_column(
    "sharpness",
    image_sharpness(
        col("image"),
        max_edge=512
    )
)

image_hash

计算图像感知哈希值。

输入类型： DataType.image()，解码后的图像列。

函数签名：

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

参数	类型	默认值	说明
`method`	`str`	`"phash"`	哈希算法。支持 `"phash"`、`"dhash"` 和 `"ahash"`。

返回类型： DataType.string()。

from pyflink.multimodal.operators import image_hash

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

数据过滤

image_size_filter

按图像像素尺寸过滤图像。

输入类型： DataType.binary() 或 DataType.image()，编码图片字节列或解码后的图像列。

函数签名：

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

参数	类型	默认值	说明
`min_w`	`Optional[int]`	`None`	最小宽度，包含端点。
`min_h`	`Optional[int]`	`None`	最小高度，包含端点。
`max_w`	`Optional[int]`	`None`	最大宽度，包含端点。
`max_h`	`Optional[int]`	`None`	最大高度，包含端点。

至少需要设置一个边界。编码图片只读取轻量元数据；无效或损坏输入返回 False。

返回类型： DataType.boolean()。

from pyflink.multimodal.operators import image_size_filter

result = df.filter(
    image_size_filter(
        col("image_bytes"),
        min_w=256,
        min_h=256,
        max_w=8192,
        max_h=8192
    )
)

image_shape_filter

按宽高比过滤图像。

输入类型： DataType.binary() 或 DataType.image()，编码图片字节列或解码后的图像列。

函数签名：

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

参数	类型	默认值	说明
`min_ratio`	`Optional[float]`	`None`	最小宽高比 `width / height`，包含端点。
`max_ratio`	`Optional[float]`	`None`	最大宽高比 `width / height`，包含端点。

至少需要设置一个边界。无效或损坏输入返回 False。

返回类型： DataType.boolean()。

from pyflink.multimodal.operators import image_shape_filter

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

image_file_size_filter

按编码文件大小过滤图像。该算子只接受编码字节，不接受 DataType.image()。

输入类型： DataType.binary()，编码图片字节列。

函数签名：

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

参数	类型	默认值	说明
`min_bytes`	`Optional[int]`	`None`	最小字节数，包含端点。
`max_bytes`	`Optional[int]`	`None`	最大字节数，包含端点。

至少需要设置一个边界。算子只根据原始字节长度判断；图片内容损坏但大小符合条件时仍返回 True。

返回类型： DataType.boolean()。

from pyflink.multimodal.operators import image_file_size_filter

result = df.filter(
    image_file_size_filter(
        col("image_bytes"),
        min_bytes=1024,
        max_bytes=5 * 1024 * 1024
    )
)