EMR Serverless Spark 多模态算子市场：算子清单与参数参考-开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心

EMR Serverless Spark 的多模态算子市场提供丰富的开箱可用算子，覆盖图片、视频、音频与文本的处理、打分和 AI 理解。您在 Daft DataFrame 中通过 emr_udf() 调用这些算子，无需自行封装模型服务或维护处理脚本。本文介绍算子的调用方式与参数分类，给出完整算子清单，并提供常用算子的参数明细与代码示例。

功能概述

在智能驾驶等多模态数据处理场景中，通常需要将视频转码、视频抽帧、图片裁剪与编码、模型推理和内容理解等能力串联起来。传统方式往往需要分别搭建和维护 FFmpeg、本地推理程序及模型服务，系统组合复杂，开发和运维成本较高。

多模态算子市场将上述能力统一封装为可复用的 Daft 算子。您可以在同一条 DataFrame 数据处理链路中，按需完成多媒体数据解析与预处理、内容打分、模型推理及向量化等任务。例如，对于一段智能驾驶视频，可以先进行转码和抽帧，再完成图片编码处理，最后调用模型服务进行目标识别、场景分析或其他内容理解，无需在多个系统之间切换和编排。

使用方式

算子随 EMR Serverless Spark 引擎镜像提供，您在工作空间中创建 Notebook 即可直接调用，无需额外安装依赖。创建 Notebook 时需选择包含目标算子的引擎版本，详情请参见Notebook开发和管理运行环境。

所有算子都是算子类，通过 emr_udf() 包装为 Daft Expression 后在 with_column() 中调用。以视频转码算子 VideoConvert 为例：

from daft.emr.functions import emr_udf, VideoConvert
from daft import col

ds = ds.with_column(
    "result",
    emr_udf(
        VideoConvert,
        construct_args={"output_format": "mov",
                        "extra_params": ["-c:v", "libx265", "-b:v", "2M"]},
        num_cpus=1, num_gpus=0, concurrency=1, batch_size=1,
    )(col("input_path"), col("output_path")),
)

调用时需要区分三组入参：

算子类：emr_udf() 的第一个参数，例如 VideoConvert。
业务参数：通过 construct_args 传入，决定算子的处理逻辑。
输入数据列：包装后的表达式的调用参数，接收 DataFrame 的列。

资源与并发参数（num_cpus、num_gpus、concurrency、batch_size）直接传给 emr_udf()，取值说明见附录 B：emr_udf 包装参数。

AI 算子的调用方式相同，模型服务由 EMR Serverless Spark 内置的 Model Manager 提供，您无需配置服务地址或鉴权信息。AI 算子的输入数据列通常使用命名参数传入，以 QwenVLVideoUnderstanding 为例：

from daft.emr.functions import emr_udf, QwenVLVideoUnderstanding
from daft import col

ds = ds.with_column(
    "result",
    emr_udf(
        QwenVLVideoUnderstanding,
        construct_args={
            "model": "qwen3.6-plus",
            "prompt": "请给出这段视频的详细描述。",
        },
        num_cpus=1, num_gpus=0, concurrency=4, batch_size=1,
    )(
        videos=col("video_url"),
        user_prompts=col("prompt"),
    ),
)

说明

为便于阅读，下文各算子的使用示例沿用以上两段代码的导入语句，仅展示 with_column() 调用部分。直接复制运行时，请补上 from daft.emr.functions import emr_udf, <算子类> 和 from daft import col，并确保 df 为待处理的 DataFrame。

参数分类

算子参数分为四类，与控制台算子表单的各区域一一对应。了解这个对应关系，可以帮助您在表单配置和代码调用之间切换。

控制台表单区域	文档中的名称	说明
业务场景	章节分组	文档与文本、图片、视频、音频
算子与标签	算子标题	基础算子在计算节点本地执行，不调用大模型；AI 算子通过 EMR 内置 Model Manager 调用视觉语言模型、大语言模型或内容安全能力
输入类型、OSS 路径	输入数据列（`__call__` 的参数）	支持 OSS 路径、挂载的本地路径和 DataFrame 列
算子参数	业务参数（`construct_args` 的成员）	表单中带星号的参数为必选
由框架处理，表单不展示	执行参数与基础设施参数	性能调优参数和 OSS 凭证等，详见本文附录
输出类型	输出（RETURN_DTYPE）	打印或写回等行为由框架的 sink 决定

算子清单

算子按业务场景分为图片、视频、音频、文档与文本四类。常用算子在本文图片算子及之后的章节提供完整的参数说明和使用示例，其余算子在本节提供参数索引。此外还提供表达式级快捷 API，详见表达式快捷 API。

下表的「类型」列区分两类算子：基础算子在计算节点本地执行，依赖镜像内的 FFmpeg 与本地模型；AI 算子通过 EMR Serverless Spark 内置的 Model Manager 调用视觉语言模型、大语言模型和内容安全能力，您无需配置服务地址、鉴权密钥或路由开关。

下表列出提供完整参数说明的算子。

场景	算子	类型	说明	输出
图片	ImageCompress	基础算子	图像压缩、缩放与格式转换	路径 string
图片	ImageResample	基础算子	图像重采样，调整尺寸与 DPI	struct
图片	ImageSharpness	基础算子	图像清晰度评估	float64
图片	ImageQualityScore	基础算子	基于 CLIP-IQA 的图像质量评分，取值范围 [0,1]	float64
图片	ImageEmbedding	基础算子	使用镜像内 CLIP 模型生成图像向量	list[float64]
视频	VideoConvert	基础算子	视频格式与编码转换	路径 string
视频	VideoCompress	基础算子	自适应压缩到目标文件大小	路径 string
视频	VideoClip	基础算子	按时间戳剪裁视频片段	struct
视频	VideoConcat	基础算子	多个视频按顺序拼接	路径 string
视频	VideoCrop	基础算子	裁剪视频画面区域	路径 string
视频	VideoFaceBlur	基础算子	检测人脸并做模糊处理	路径 string
视频	VideoBlackBorderCrop	基础算子	检测并裁剪视频黑边	路径 string
视频	VideoMotionScore	基础算子	基于光流计算运动评分	struct
视频	VideoKeyframeExtract	基础算子	提取关键帧，支持 CLIP 与直方图两种策略	struct
视频	VisionUnderstanding	AI 算子	通用多模态理解，支持图片、视频与文本输入	struct
视频	VideoUnderstanding	AI 算子	视频内容理解	struct
视频	VideoFineUnderstanding	AI 算子	精细化视频理解	struct
视频	QwenVLVideoUnderstanding	AI 算子	基于 Qwen-VL 的视频理解	struct
视频	VideoSceneSeg	AI 算子	视频场景分割与人物识别	struct
视频	VideoInpaint	AI 算子	去除水印与字幕的视频修复	struct
视频	VideoRiskRec	AI 算子	视频内容安全审核	struct
视频	VideoSmartEdit	AI 算子	智能剪辑高价值片段	struct
文档与文本	TextGeneration	AI 算子	纯文本生成	struct

说明

VisionUnderstanding 支持图片、视频和文本输入，按主要用途归入视频场景。

下表按场景列出其余算子的输入列、核心业务参数与输出类型。阅读时请注意以下约定：

参数值标注为必填的参数没有默认值，必须在 construct_args 中显式传入，否则调用失败。
参数值标注为 代码默认 的参数由算子实现指定默认值，通常无需修改。
输入列标注 =None 表示该列可省略。
下表省略了通用存储和 IO 参数，这部分参数对以下算子同样生效，详见附录 A：通用存储/IO 参数。

图片与视频

算子	输入列	核心 construct_args（默认值）	输出
ClipEmbedding	content	content_type=image；batch_size=16；model_path=None；model_name=openai/clip-vit-base-patch32；device=None	list[float64]
ImageBlackBorderCrop	images、output_basenames	image_format=jpeg；output_dir=""；image_src_type=image_url；detect_algorithm=auto；black_threshold=10；crop_sides=None；target_dpi=None；quality=85；on_io_error=skip	struct
ImageCrop	images、output_basenames	image_format=jpeg；output_dir=""；image_src_type=image_url；crop_type=center；crop_coords/crop_ratio/crop_size/crop_margins=None；quality=85；method=lanczos；on_io_error=skip	struct
ImageViTEmbedding	images	image_src_type=image_url；dtype=float32；batch_size=32；model_path=/opt/las/models；model_name=facebook/dinov2-large；use_cls_token_embedding=True；rank=0	list[float32]
VideoHasAudio	video_paths	timeout=None	bool
VideoMetadataExtract	video_paths	timeout=None	struct
VideoRemoveAudio	input_paths、output_paths	output_format=mp4；timeout=None	string
VideoResolutionAdjust	input_paths、output_paths	min/max_width=1280/2560；min/max_height=1280/2560；force_original_aspect_ratio_type=decrease；force_divisible_by=2；crf=23；preset=medium；output_format=mp4	string
VideoSplitByKeyframe	input_paths、output_path_templates	method=I_frame；threshold=0；keyframes_cnt=10；seconds_per_frame=-1；output_format=mp4；timeout=None	list[string]

音频

音频场景的算子分为两类能力：

基础音频能力：语音识别（ASR）、语言识别、语音活动检测（VAD）、CTC 对齐、音频事件分类、说话人验证，以及音质、语音质量和信噪比评分；此外还包括转码、拼接、按时长切分、标准化和静音、时长、大小检测。
本地大模型能力：QwenOmniAudioUnderstanding 使用镜像内的 Qwen2.5-Omni 模型进行音频理解，无需调用外部模型服务。

算子	输入列	核心 construct_args（默认值）	输出
AudioAsrLidWhisper	audios	model_path=None；model_name/model_version=代码默认；punc_model_path=None；return_language_only=False	struct
AudioAsrWhisper	audios、languages=None	model_path=None；batch_size=10；source_language=None；translate_to_english=False；temperature=0.5；dtype=bfloat16	struct
AudioBeatsClassifier	audios	model_path=None；top_k=代码默认；precision=None；chunk_seconds=代码默认	list[struct]
AudioCTCAligner	audios、texts、langs	model_path=None	list[struct]
AudioConcat	audio_paths_list、output_paths	output_format=mp3；sample_rate=16000；timeout=None；extra_params=None	string
AudioConvert	input_paths、output_paths	output_format=mp3；timeout=None；extra_params=None	string
AudioConvertToMp3	input_paths、output_paths	bitrate=None；sample_rate=None；quality=None；timeout=None	string
AudioDuration	audio_inputs	timeout=None	float64
AudioFFMPEGWrapped	audio_paths	filter_name=必填；filter_kwargs=None；output_dir=None；output_format=wav；timeout=None	string
AudioLidWhisper	audios	model_path=None；model_version=代码默认	struct
AudioMetascore	paths	model_path=None	struct
AudioSNR	audio_paths	n_components=8；max_iter=200；flatness_threshold=0.3；n_fft=2048；target_sr=16000；max_duration=None	float64
AudioSilenceDetect	audio_paths	silence_threshold=-60.0；timeout=None	bool
AudioSize	audio_paths	无业务构造参数	int64
AudioSpeakerVerificationEres2net	speaker_a_audios、speaker_b_audios	model_path=None；model_version=代码默认	float32
AudioSpeechScore	audio_paths、reference_audio_paths=None	model_path=None；metrics=None	struct
AudioSplitByDuration	input_paths、output_path_templates	split_duration=必填；output_format=wav；min_segment_duration=0；timeout=None	list[string]
AudioStandardization	audios	target_sr=None；target_channels=None；target_dbfs=None；target_gain_range=None；output_format=wav	binary
AudioVadFsmn	audios	model_path=None；model_version=代码默认；batch_size_s=代码默认	list[struct]
QwenOmniAudioUnderstanding	audio_inputs、prompts=None	model_path=None；model_name=代码默认；default_prompt=代码默认；system_prompt=None；max_new_tokens=512；batch_size=4；device/dtype=None	string

Model Manager 服务算子

算子	输入列	核心 construct_args（默认值）	输出
BailianThinkingVision	images=None、texts=None	model=None；version=None；source_type=url；enable_thinking=True；thinking_budget=None；max_tokens=None；max_concurrency=100；request_timeout=1200	struct
BailianVideoFaceBlur	video_paths、output_basenames=None	business_policy=必填；output_dir=""；model=None；keep_threshold=0.9；model_path=/opt/emr/models；det_thresh=0.3；keep_audio=True；request_timeout=300	struct
MultilingualTextTranslate	contents	model=None；source_language=Chinese；target_language=English；max_tokens=None；max_concurrency=32；request_timeout=600	string
TextSafetyScorer	text	model=None；max_concurrency=32；request_timeout=600	struct

文档解析

算子	输入列	核心 construct_args（默认值）	输出
PDFVisionParse	pdf_url=None	start_page=0；num_pages=-1；output_oss_path=""；model=None；pages_per_request=1；request_concurrency=4；render_scale=2；request_timeout=600	struct
XlsxParse	xlsx_col=None	save_type=markdown；output_oss_path=""	struct

文本处理

算子	输入列	核心 construct_args（默认值）	输出
AlphanumericRatioCalculator	texts	tokenization=False；model_path=/opt/emr/models；model_name=pythia-6.9b-deduped	float64
BgeEmbedding	texts	dtype=float32；batch_size=512；model_path=/opt/las/models；model_name=代码默认；rank=None；use_query_instruction=False	list[float32]
BulletLineRatioCalculator	texts	无业务构造参数	float64
CommonCrawlContentExtractor	warc_files	warc_src_type=必填；extractor_type=trafilatura；max_records=None	list[struct]
CopyrightCleaner	texts	无业务构造参数	string
EnTextQualityScorer	texts	batch_size=32；model_path=/opt/emr/models；model_name=代码默认	float64
HtmlTagRemover	texts	separator=""；strip=True	string
MaximumWordLengthCalculator	texts	无业务构造参数	int64
PerplexityCalculator	texts	lang=zh；model_path=/opt/emr/models；model_name=kenlm/wikipedia	float64
RegexReplacer	texts	patterns=必填；replacements=必填	string
RepeatedLinesCalculator	texts	无业务构造参数	float64
TextLengthCalculator	texts	无业务构造参数	int64
UrlRatioCalculator	texts	无业务构造参数	float64
WhitespaceNormalizer	texts	无业务构造参数	string
WordRepetitionCalculator	texts	repetition=5；lang=zh；tokenization=True；model_path=/opt/emr/models；model_name=kenlm/wikipedia	float64

通用

算子	输入列	核心 construct_args（默认值）	输出
TimestampMerge	timestamp_ranges	pre_merge_gap_seconds=0；max_span_seconds=None	list[struct]

表达式快捷 API

除算子类之外，还提供以下表达式级快捷 API。

API	作用	说明
ai_query	文本与多模态模型查询	表达式级快捷入口
ai_embedding	文本向量化	表达式级快捷入口
ai_embedding_multimodal	多模态向量化	表达式级快捷入口
emr_udf	将算子类包装为 Daft Expression	同时接收资源、批大小与服务批处理参数，取值说明见附录 B：emr_udf 包装参数

输出中包含 prompt_tokens 和 completion_tokens 的算子，会在作业结束后于 Driver 日志中聚合 Token 用量。

图片算子

本节列出图片算子的输入数据列、业务参数与使用示例。执行参数与基础设施参数统一见附录。

ImageCompress（图像压缩）

图像压缩/缩放，支持格式转换、质量调整、元数据保留。

输入类型：OSS 路径 / 挂载路径 / DataFrame 输出（RETURN_DTYPE）：string()（输出文件路径，失败为 None）

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
图像路径数组	`image_paths`	与 image_binaries 二选一	list[str]
图像二进制数组	`image_binaries`	与 image_paths 二选一	list[bytes]
图像格式提示	`image_formats`	否	list[str]（仅二进制输入用）
输出文件基础名	`output_basenames`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出目录	`output_dir`	否	""	目录路径（空=写临时文件）
输出格式	`image_format`	否	None	jpeg / png / webp（None=保留源格式）
压缩质量	`quality`	否	85	整数 [1, 100]
最大宽度	`max_width`	否	None	正整数（按比例缩放）
最大高度	`max_height`	否	None	正整数（按比例缩放）
缩放插值算法	`method`	否	lanczos	nearest / bilinear / bicubic / lanczos
启用格式优化	`optimize`	否	True	bool
保留 EXIF/ICC	`keep_metadata`	否	False	bool

基础设施参数：io_config（见附录 A）

使用示例

df = df.with_column("compressed_path", emr_udf(
    ImageCompress,
    construct_args={"image_format": "webp", "quality": 80, "max_width": 1920,
                    "output_dir": "oss://bucket/compressed/"},
)(image_paths=col("image_path")))

ImageResample（图像重采样）

按目标尺寸/DPI 重采样，返回 Base64 + 可选输出路径。

输出（RETURN_DTYPE）：struct({base64: string, image_path: string})

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
图像数据数组	`images`	是	list[Any]（URL/Base64/二进制）
输出文件基础名	`output_basenames`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输入源类型	`image_src_type`	否	image_url	image_url / image_base64 / image_binary
图像格式提示	`image_format`	否	png	png / jpg 等
输出目录	`output_dir`	否	""	目录路径（空=仅返回 Base64）
目标尺寸	`target_size`	否	None	[宽, 高] 正整数
目标 DPI	`target_dpi`	否	None	整数或 [x, y]
缩放插值算法	`method`	否	bicubic	nearest / bilinear / bicubic / lanczos
IO 错误策略	`on_io_error`	否	skip	skip / raise

基础设施参数：io_config（见附录 A）

使用示例

df = df.with_column("resampled", emr_udf(
    ImageResample,
    construct_args={"target_size": [512, 512], "method": "bicubic",
                    "output_dir": "oss://bucket/resampled/"},
)(images=col("image_url")))

ImageSharpness（图像清晰度评估）

计算图像清晰度分数（4 种算法）。

输出（RETURN_DTYPE）：float64()（清晰度分数，失败为 None）

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
图像数据数组	`images`	是	list[Any]
（未使用，仅保持 API 一致）	`output_basenames`	否	忽略

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输入源类型	`image_src_type`	否	image_url	image_url / image_base64 / image_binary
清晰度算法	`method`	否	laplacian	laplacian / tenengrad / brenner / fft_highfreq

基础设施参数：io_config（见附录 A）

使用示例

df = df.with_column("sharpness", emr_udf(
    ImageSharpness,
    construct_args={"method": "laplacian"},
)(images=col("image_url")))

ImageQualityScore（图像质量评分）

使用 CLIP-IQA 模型评估图像质量，输出 [0.0, 1.0]。

输出（RETURN_DTYPE）：float64()（质量分数 [0,1]，失败为 None）

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
图像数据数组	`images`	是	list[Any]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
CLIP 模型名	`clip_model_name`	否	openai/clip-vit-large-patch14	HF 模型 ID
本地模型目录	`model_path`	否	/opt/emr/models	目录路径或 oss:// 路径
CLIP-IQA 锚词	`prompt`	否	quality	字符串
输入源类型	`image_src_type`	否	image_url	image_url / image_base64 / image_binary

执行参数：device（cuda；cpu/cuda/cuda:N）、dtype（float16；float16/float32）、batch_size（16）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("quality_score", emr_udf(
    ImageQualityScore,
    construct_args={"device": "cuda", "model_path": "/opt/emr/models"},
    num_gpus=1, batch_size=16,
)(images=col("image_url")))

ImageEmbedding（本地 CLIP 图像嵌入）

使用本地 CLIP 模型生成图像嵌入向量。

输出（RETURN_DTYPE）：list(float64())

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
图像数据数组	`images`	是	list[Any]（URL/Base64/二进制）

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
模型	`model`	否	None → openai/clip-vit-base-patch32	HF 模型 ID 或本地子目录
本地模型基础目录	`model_path`	否	/opt/emr/models	目录路径
输入源类型	`image_src_type`	否	image_url	image_url / image_base64 / image_binary
图像格式提示	`image_format`	否	png	png / jpg 等
L2 正则化向量	`normalize`	否	True	bool

执行参数：dtype（float32；float16/float32/bfloat16）、batch_size（16）、rank（GPU 索引，0）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("embedding", emr_udf(
    ImageEmbedding,
    construct_args={"model": "openai/clip-vit-base-patch32", "normalize": True},
    num_gpus=1, batch_size=16,
)(images=col("image_url")))

视频算子

本节列出视频算子的输入数据列、业务参数与使用示例。执行参数与基础设施参数统一见附录。

VideoConvert（视频格式/编码转换）

基于 ffmpeg 的通用视频格式转换。

输出（RETURN_DTYPE）：string()

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
输入视频路径列表	`input_paths`	是	list[str]
输出视频路径列表	`output_paths`	是	list[str]，与 input_paths 等长

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出格式	`output_format`	是	—	mp4 / avi / mov / mkv / flv / webm 等
额外 ffmpeg 参数	`extra_params`	否	None	list[str]，透传 ffmpeg；编码器/比特率在此指定

执行参数：timeout（None）；基础设施参数：io_config / oss_*（见附录 A）

说明

VideoConvert 没有独立的视频编码器、音频编码器、视频比特率和音频比特率参数。需要指定这些设置时，通过 extra_params 以 FFmpeg 参数的形式传入，参见下方使用示例。

使用示例

df = df.with_column("converted_path", emr_udf(
    VideoConvert,
    construct_args={"output_format": "mp4",
                    "extra_params": ["-c:v", "libx265", "-b:v", "2M", "-c:a", "aac", "-b:a", "128k"]},
)(col("input_path"), col("output_path")))

VideoCompress（自适应视频压缩）

多阶段自适应压缩到目标文件大小。

输出（RETURN_DTYPE）：string()

输入数据列（__call__）（video_paths 在当前签名中没有默认值，调用时必须传入；仅使用二进制输入时也应传一个可为 null 的 video_paths 列。video_binaries 为可选。）

中文名	英文参数名	必选	类型/取值
输入视频路径列表	`video_paths`	是（可传 null 列）	list[str]
输入视频二进制列表	`video_binaries`	否	list[bytes]
视频格式	`video_formats`	否	list[str]（仅二进制时需要）
输出文件名基础	`output_basenames`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出 OSS 目录	`output_oss_dir`	否	""	字符串（空=临时目录）
目标最大文件大小(MB)	`max_output_size_mb`	否	50.0	浮点数
目标帧率(fps)	`target_fps`	否	5.0	浮点数
最小分辨率高度(px)	`min_resolution_height`	否	360	整数
允许直接输出的格式	`allowed_formats`	否	["mp4","avi","mov"]	list[str]

执行参数：rank（None=CPU libx264 / 整数=GPU h264_nvenc）、timeout（None）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("compressed_path", emr_udf(
    VideoCompress,
    construct_args={"max_output_size_mb": 50.0, "target_fps": 5.0,
                    "output_oss_dir": "oss://bucket/compressed/"},
)(video_paths=col("video_path")))

VideoClip（视频片段剪裁）

按时间戳范围精确剪裁片段。

输出（RETURN_DTYPE）：struct({segments: list[string], segments_binary: list[binary]})

输入数据列（__call__）：video_paths / video_binaries 至少一个。

中文名	英文参数名	必选	类型/取值
输入视频路径列表	`video_paths`	二选一	list[str]
输入视频二进制列表	`video_binaries`	二选一	list[bytes]
视频格式	`video_formats`	否	list[str]
时间戳范围列表	`timestamp_ranges`	否	list[list[float]]，每项 [start, end] 秒
输出文件名基础	`output_basenames`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出 OSS 目录	`output_oss_dir`	否	""	字符串（空=临时目录）
返回片段二进制	`output_segments_binary`	否	False	bool
输出视频格式	`output_video_format`	否	None	如 mp4（None=保持原格式）

执行参数：timeout（None）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

# timestamp_ranges 列每行形如 [[0.0, 5.0], [10.0, 15.0]]
df = df.with_column("clips", emr_udf(
    VideoClip,
    construct_args={"output_oss_dir": "oss://bucket/clips/", "output_video_format": "mp4"},
)(video_paths=col("video_path"), timestamp_ranges=col("ranges")))

VideoConcat（视频拼接）

顺序拼接多个视频成一个输出。

输出（RETURN_DTYPE）：string()

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频路径列表的列表	`video_paths_list`	是	list[list[str]]，每个内层=一个拼接任务
输出视频路径列表	`output_paths`	是	list[str]，与 video_paths_list 等长

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出格式	`output_format`	否	mp4	mp4 / avi 等
视频编码器	`video_codec`	否	libx264	libx264 / h264_nvenc 等
音频编码器	`audio_codec`	否	aac	aac / libmp3lame 等
额外 ffmpeg 参数	`extra_params`	否	None	list[str]

执行参数：timeout（None）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

# video_paths_list 列每行形如 ["oss://bucket/a.mp4", "oss://bucket/b.mp4"]
df = df.with_column("concat_path", emr_udf(
    VideoConcat,
    construct_args={"output_format": "mp4", "video_codec": "libx264", "audio_codec": "aac"},
)(col("video_paths_list"), col("output_path")))

VideoCrop（视频区域裁剪）

支持坐标 / 分辨率 / 宽高比三种裁剪模式。

输出（RETURN_DTYPE）：string()

说明

必须至少指定 bbox、target_width/target_height、aspect_ratio 之一；bbox 优先级最高。

输入数据列（__call__）：video_paths / video_binaries 至少一个。

中文名	英文参数名	必选	类型/取值
输入视频路径列表	`video_paths`	二选一	list[str]
输入视频二进制列表	`video_binaries`	二选一	list[bytes]
视频格式	`video_formats`	否	list[str]
输出文件名基础	`output_basenames`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出目录	`output_oss_dir`	否	""	字符串
目标宽度(px)	`target_width`	否	None	整数
目标高度(px)	`target_height`	否	None	整数
宽高比	`aspect_ratio`	否	None	浮点数（如 16/9=1.7778）
裁剪区域(坐标)	`bbox`	否	None	(x, y, w, h)
裁剪位置	`crop_mode`	否	center	center / top / bottom / left / right
像素对齐步长	`force_divisible_by`	否	2	整数
libx264 质量因子	`crf`	否	23.0	浮点数 [0, 51]
编码预设	`preset`	否	medium	libx264 预设值
NVENC 质量参数	`cq`	否	0	浮点数 [0, 51]（0=自动）
NVENC 速率控制	`rc`	否	vbr	vbr / cbr 等

执行参数：rank（None=CPU / 整数=GPU）、timeout（None）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("cropped_path", emr_udf(
    VideoCrop,
    construct_args={"target_width": 1280, "target_height": 720, "crop_mode": "center",
                    "output_oss_dir": "oss://bucket/cropped/"},
)(video_paths=col("video_path")))

VideoFaceBlur（视频人脸模糊）

使用 InsightFace、YuNet 或 YOLO5Face 检测并模糊人脸，支持跨帧跟踪、检测确认、前后补帧、远景小脸补充与音轨保留；最终输出浏览器兼容 H.264 MP4。

输出（RETURN_DTYPE）：string()（模糊后视频路径）

输入数据列（__call__）：video_paths / video_binaries 至少一个。

中文名	英文参数名	必选	类型/取值
视频路径列表	`video_paths`	二选一	list[str]
视频二进制列表	`video_binaries`	二选一	list[bytes]
视频格式	`video_formats`	否	list[str]
输出文件名基础	`output_basenames`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/说明
输出目录	`output_dir`	否	""	本地或远端目录；空=临时目录
模型基础目录	`model_path`	否	/opt/emr/models	InsightFace/Haar 模型基础目录
InsightFace 子目录	`model_name`	否	insightface	字符串
模糊类型	`blur_type`	否	gaussian	gaussian / mean / box
模糊半径	`radius`	否	10.0	推荐 5–20
检测置信度	`det_thresh`	否	0.3	0–1
检测输入尺寸	`det_size`	否	(640,640)	(宽,高)
保留音轨	`keep_audio`	否	True	bool
ffmpeg 超时	`timeout`	否	None	秒
YuNet 模型	`yunet_model_path`	否	None	与 yolov5face_model_path 互斥
YuNet 阈值	`yunet_score_threshold`	否	0.5	0–1
YuNet 跟踪窗口	`yunet_tracking_window_seconds`	否	0.5	秒；None=沿用通用窗口
启用跟踪	`enable_tracking`	否	True	bool
最大丢失帧	`max_missed_frames`	否	None	未验证轨迹的帧数上限
跟踪后补帧	`tracking_post_roll_seconds`	否	2.0	秒
跟踪前补帧	`tracking_pre_roll_seconds`	否	2.0	秒
检测确认帧数	`detection_confirmation_frames`	否	2	整数
确认最大间隔	`detection_confirmation_max_gap`	否	1	帧
Haar 最小置信度	`haar_min_confidence`	否	3.0	浮点数
人脸框扩展	`bbox_expand_ratio`	否	0.2	单边比例
跟踪 IoU 阈值	`tracking_iou_threshold`	否	0.3	0–1
YOLO5Face 模型	`yolov5face_model_path`	否	None	基础检测模型
远景 YOLO5Face 模型	`yolov5face_far_model_path`	否	None	依赖基础 YOLO5Face 模型
YOLO 基础阈值	`yolov5face_confidence_threshold`	否	0.3	0–1
YOLO 远景阈值	`yolov5face_far_confidence_threshold`	否	0.5	0–1
YOLO 输入尺寸	`yolov5face_image_size`	否	640	正整数
YOLO 远景尺寸	`yolov5face_far_image_size`	否	1280	正整数
YOLO NMS 阈值	`yolov5face_nms_threshold`	否	0.5	0–1
远景补充框上限	`yolov5face_max_far_supplement_side`	否	32	输出像素
YOLO 跟踪窗口	`yolov5face_tracking_window_seconds`	否	0.5	秒；None=沿用通用窗口

基础设施参数：io_config / oss_*（见附录 A）

说明

YuNet 与 YOLO5Face 基础模型互斥，只能配置其中一个。远景 YOLO5Face 模型必须与基础 YOLO5Face 模型配套使用。

使用示例

df = df.with_column("blurred_path", emr_udf(
    VideoFaceBlur,
    construct_args={"model_path": "/opt/emr/models", "det_thresh": 0.3,
                    "keep_audio": True, "enable_tracking": True,
                    "bbox_expand_ratio": 0.2,
                    "output_dir": "oss://bucket/blurred/"},
    num_gpus=1,
)(video_paths=col("video_path")))

VideoBlackBorderCrop（视频黑边裁剪）

自动检测并裁剪黑边，保持音视频同步。

输出（RETURN_DTYPE）：string()（裁剪后视频路径，失败为 None）

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
输入视频路径列表	`input_col`	是	list[str]
输出视频路径列表	`output_col`	是	list[str]，与 input_col 等长

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
检测方法	`detection_method`	否	threshold_ratio	threshold_ratio / edge_detection / histogram
黑边亮度阈值	`black_threshold`	否	10	整数 [0, 255]
有效像素比例阈值	`valid_pixel_ratio`	否	0.1	浮点数 [0, 1]
采样帧数	`sample_frames`	否	20	整数
保留音频	`is_keep_audio`	否	True	bool
核心区域比例	`core_region_ratio`	否	0.5	浮点数 [0, 1]
连续黑行阈值	`continuous_black_rows`	否	3	整数
连续黑列阈值	`continuous_black_cols`	否	3	整数
暗区域亮度阈值	`dark_region_brightness`	否	50	整数 [0, 255]
边缘检测灵敏度	`edge_sensitivity`	否	1.0	浮点数

执行参数：timeout（None）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("cropped_path", emr_udf(
    VideoBlackBorderCrop,
    construct_args={"detection_method": "threshold_ratio", "black_threshold": 10,
                    "is_keep_audio": True},
)(col("input_path"), col("output_path")))

VideoMotionScore（视频运动评分）

用光流算法计算视频运动强度及相关指标。

输出（RETURN_DTYPE）：struct({...})，字段：standardized_score(0–1)、motion_pattern(static/low/medium/high_motion)、mean_score、median_score、dynamic_mean_score、high_percentile_score、dynamic_high_percentile_score、density_score、video_resolution[w,h]、total_frames、sample_frames_count、used_algorithm、status、total_process_time

输入数据列（__call__）：video_paths / video_binaries 至少一个。

中文名	英文参数名	必选	类型/取值
视频路径列表	`video_paths`	二选一	list[str]
视频二进制列表	`video_binaries`	二选一	list[bytes]
视频格式	`video_formats`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
光流算法	`optical_flow_algorithm`	否	farneback	farneback / tv-l1 / dis-ultrafast / dis-fast / dis-medium / dis-accurate
帧采样比例	`sample_ratio`	否	0.125	浮点数 [0, 1]
运动幅度阈值(滤噪)	`mag_threshold`	否	0.01	浮点数
光流离群值阈值	`flow_threshold`	否	6.0	浮点数
平滑窗口大小	`smooth_window`	否	5	整数
帧下采样比例	`downsample_ratio`	否	1.0	浮点数 [0, 1]
高运动决策阈值	`high_motion_threshold`	否	0.02	浮点数

执行参数：batch_size（10）、num_workers（4）、use_cuda（False）、rank（GPU 索引 0）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("motion", emr_udf(
    VideoMotionScore,
    construct_args={"optical_flow_algorithm": "farneback", "sample_ratio": 0.125},
    num_cpus=4,
)(video_paths=col("video_path")))

VideoKeyframeExtract（视频关键帧提取）

CLIP 语义或帧差（histogram）两种方式提取关键帧。

输出（RETURN_DTYPE）：struct({oss_paths: list[string], frame_ids: list[int64], scores: list[float64]})

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频输入列表	`videos`	是	list，元素随 video_src_type：URL/base64 字符串或 bytes

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
提取方法	`method`	否	histogram	clip（语义，需 GPU）/ histogram（帧差，轻量）
CLIP 文本描述	`text`	否	""	字符串（空=无文本约束）
模型基础目录	`model_path`	否	/opt/emr/models	路径
CLIP 模型名	`clip_model_name`	否	openai/clip-vit-base-patch32	HF 模型 ID
最大关键帧数	`max_num_frames`	否	32	整数 (>0)
候选帧采样率(fps)	`fps`	否	1.0	浮点数
输入格式类型	`video_src_type`	否	video_url	video_url / video_base64 / video_binary
视频格式	`video_format`	否	mp4	字符串
关键帧 OSS 输出目录	`output_oss_dir`	否	""	字符串（空=不上传）
返回关键帧图像数据	`return_keyframes`	否	True	bool
输出图像格式	`img_type`	否	.jpg	.jpg / .png 等
峰值检测阈值	`t1`	否	0.8	浮点数
标准差阈值	`t2`	否	-100	浮点数
最大递归深度	`all_depth`	否	5	整数
histogram 检测阈值	`threshold`	否	0.5	浮点数 [0, 1]
histogram 最小关键帧间隔(帧)	`min_interval`	否	10	整数（取 max(1, x)）

执行参数：dtype（float16；float16/float32）、batch_size（16）、rank（GPU 索引 0）；基础设施参数：io_config / oss_*（见附录 A）

使用示例

df = df.with_column("keyframes", emr_udf(
    VideoKeyframeExtract,
    construct_args={"method": "histogram", "max_num_frames": 32,
                    "output_oss_dir": "oss://bucket/keyframes/"},
)(videos=col("video_url")))

VisionUnderstanding（通用多模态理解）

支持任意组合的图像、视频、文本输入（多模态）。

输出（RETURN_DTYPE）：struct(content, reasoning_content, finish_reason, prompt_tokens, completion_tokens, total_tokens, model, id, error)

输入数据列（__call__）：images / videos / texts 按需提供。

中文名	英文参数名	必选	类型/取值
图像数据	`images`	否	单个或列表（URL/base64/二进制）
视频数据	`videos`	否	单个或列表
文本提示	`texts`	否	单个或列表（字符串）

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
模型名	`model`	否	None → qwen3.6-plus（Model Manager 默认）	字符串
模型版本	`version`	否	None	如 "250115"（拼成 {model}-{version}）
媒体数据源格式	`source_type`	否	url	url / base64 / binary
系统文本提示	`system_text`	否	None	字符串
系统图像 URL	`system_image_url`	否	None	URL
系统视频 URL	`system_video_url`	否	None	URL
图像质量级别	`image_detail`	否	None	high / low / auto
图像编码格式	`image_format`	否	jpeg	jpeg / png / webp / gif
视频编码格式	`video_format`	否	mp4	mp4 / avi / mov
视频帧率(fps)	`video_fps`	否	1.0	浮点数 [0.2, 5]

通用 LLM 生成参数（max_tokens / temperature / top_p / stop / frequency_penalty / presence_penalty / logit_bias / tools / enable_thinking / llm_config）见附录 D

执行参数：max_concurrency（100）、request_timeout（1200s）

使用示例

df = df.with_column("understanding", emr_udf(
    VisionUnderstanding,
    construct_args={"model": "qwen3.6-plus", "system_text": "你是图像理解助手"},
    concurrency=4,
)(images=col("image_url"), texts=col("question")))

VideoUnderstanding（视频内容理解）

仅视频输入的简化版理解。

输出（RETURN_DTYPE）：struct(content, reasoning_content, finish_reason, prompt_tokens, completion_tokens, total_tokens, cached_tokens, reasoning_tokens, model, id, error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频 URL 列表	`video_urls`	是	list[str]（HTTP/HTTPS）

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
默认文本提示	`text_prompt`	否	请详细描述这个视频的内容	字符串
模型名	`model`	否	None	字符串
系统提示	`system_message`	否	None	字符串或列表
视频帧率(fps)	`fps`	否	None	浮点数

通用 LLM 生成参数（max_tokens / temperature / top_p / stop / frequency_penalty / presence_penalty / logprobs / top_logprobs / logit_bias / response_format / tools / tool_choice / enable_thinking）见附录 D

执行参数：无（运行模式固定为 Model Manager）

使用示例

df = df.with_column("understanding", emr_udf(
    VideoUnderstanding,
    construct_args={"text_prompt": "请详细描述这个视频的内容", "fps": 1.0},
    concurrency=4,
)(video_urls=col("video_url")))

VideoFineUnderstanding（精细化视频理解）

适合长视频的多维度结构化分析。

输出（RETURN_DTYPE）：struct(final_summary, video_duration, resolution, total_clips, clips(JSON), prompt_tokens, completion_tokens, total_tokens, model, error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频 URL 列表	`video_urls`	是	list[str]
查询/问题列表	`queries`	是	list[str]，与 video_urls 等长

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
模型名	`model`	否	None	如 qwen3.6-plus
帧采样率(fps)	`fps`	否	1.0	浮点数
分辨率级别	`media_resolution`	否	medium	low / medium / high
推理强度	`reasoning_effort`	否	minimal	minimal / standard / high
上下文窗口大小	`clip_context`	否	medium	short / medium / long
最大输出 token	`max_tokens`	否	None	整数
采样温度	`temperature`	否	None	浮点数

执行参数：request_timeout（1200s）

使用示例

df = df.with_column("fine_understanding", emr_udf(
    VideoFineUnderstanding,
    construct_args={"fps": 1.0, "media_resolution": "medium", "reasoning_effort": "minimal"},
    concurrency=4,
)(col("video_url"), col("query")))

QwenVLVideoUnderstanding（Qwen-VL 视频理解）

通过 EMR 内置 Model Manager 调用 Qwen-VL 多模态模型。

输出（RETURN_DTYPE）：struct(content, reasoning_content, finish_reason, prompt_tokens, completion_tokens, total_tokens, cached_tokens, reasoning_tokens, model, id, error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频 URL 列表	`videos`	是	list[str]（仅 HTTP/HTTPS）
用户提示列表	`user_prompts`	否	list[str]，与 videos 等长（None 用默认提示）

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
模型名	`model`	否	None → qwen3.6-plus（Model Manager 默认）	字符串
默认用户提示	`prompt`	否	None → 请给出这段视频的详细描述。	字符串
系统提示	`system_message`	否	None	字符串
视频帧采样率(fps)	`fps`	否	None	浮点数

通用 LLM 生成参数（max_tokens / max_completion_tokens / temperature / top_p / stop / frequency_penalty / presence_penalty / enable_thinking）见附录 D

执行参数：无（运行模式固定为 Model Manager）

使用示例

df = df.with_column("understanding", emr_udf(
    QwenVLVideoUnderstanding,
    construct_args={"model": "qwen3.6-plus", "prompt": "请给出这段视频的详细描述。"},
    concurrency=4,
)(videos=col("video_url"), user_prompts=col("prompt")))

VideoSceneSeg（视频场景分割）

基于 EMR 内置 Model Manager VLM 的场景切割与人物识别。

输出（RETURN_DTYPE）：struct(task_status[COMPLETED/FAILED], video_duration, segment_count, character_count, segments(JSON), characters(JSON), error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频 URL 列表	`video_urls`	是	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
模型名	`model`	否	None	字符串
视频帧提取速率(fps)	`fps`	否	2.0	浮点数
最小场景时长(秒)	`min_segment_duration`	否	None	浮点数 [1.0, 100.0]
最大场景时长(秒)	`max_segment_duration`	否	None	浮点数 [1.0, 100.0]
最大输出 token	`max_tokens`	否	None	整数

执行参数：request_timeout（600s）

使用示例

df = df.with_column("scene_seg", emr_udf(
    VideoSceneSeg,
    construct_args={"fps": 2.0, "min_segment_duration": 2.0, "max_segment_duration": 60.0},
    concurrency=4,
)(video_urls=col("video_url")))

VideoInpaint（视频修复，去水印与字幕）

VLM 检测位置 + OpenCV 修复，去除水印和字幕。

输出（RETURN_DTYPE）：struct(inpainted_video_path, subtitle_bbox"x,y,x2,y2", resolution"WxH", error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频 URL 列表	`video_urls`	是	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
去除目标列表	`targets`	否	["watermark","subtitle"]	list[str]
输出目录	`output_oss_dir`	否	""	字符串（OSS 或本地）
VLM 模型名(检测位置)	`model`	否	None	字符串
抽样帧数	`sample_frames`	否	5	整数
OpenCV 修复半径	`inpaint_radius`	否	5	整数

执行参数：request_timeout（300s）

使用示例

df = df.with_column("inpaint", emr_udf(
    VideoInpaint,
    construct_args={"targets": ["watermark", "subtitle"], "sample_frames": 5,
                    "output_oss_dir": "oss://bucket/inpaint/"},
    concurrency=2,
)(video_urls=col("video_url")))

VideoRiskRec（视频内容安全审核）

基于阿里云内容安全 API 的视频/音频审核。

输出（RETURN_DTYPE）：struct(VideoDecision[PASS/BLOCK/REVIEW], VideoDecisionDetail, AudioDecision[PASS/BLOCK/REVIEW], AudioDecisionDetail, Message, risk_result(JSON))

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频唯一标识符列表	`data_ids`	是	list[str]
视频 URL 列表	`urls`	是	list[str]
视频标题列表	`title_col`	否	list[str]
上传者 ID 列表	`account_id_col`	否	list[str]

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
应用 ID	`app_id`	否	0	整数
审核场景类型	`biztype`	否	""	字符串
结果类型	`result_type`	否	0	0=仅违规 / 1=全部结果
帧抽样间隔(秒)	`interval`	否	2	整数
VLM 模型名	`model`	否	None	字符串

执行参数：timeout（120s）、poll_interval（10s）、num_coroutines（5）

使用示例

df = df.with_column("risk", emr_udf(
    VideoRiskRec,
    construct_args={"biztype": "videoDetection", "result_type": 0, "interval": 2},
    concurrency=4,
)(data_ids=col("id"), urls=col("video_url")))

VideoSmartEdit（视频智能编辑）

从长视频中智能提取价值内容片段。

输出（RETURN_DTYPE）：struct(total_segments, video_duration, clips(JSON), error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
视频 URL 列表	`video_urls`	是	list[str]
编辑任务描述列表	`task_descriptions`	是	list[str]（自然语言），与 video_urls 一一对应

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
输出目录	`output_dir`	否	""	字符串（本地或 oss://）
处理模式	`mode`	否	normal	simple / normal / fine
片段时长阈值(秒)	`segment_duration`	否	60	整数
模型名	`model`	否	None	字符串
视频帧提取速率(fps)	`fps`	否	2.0	浮点数
最大输出 token	`max_tokens`	否	None	整数

执行参数：request_timeout（600s）

使用示例

df = df.with_column("smart_edit", emr_udf(
    VideoSmartEdit,
    construct_args={"mode": "normal", "segment_duration": 60, "fps": 2.0,
                    "output_dir": "oss://bucket/edited/"},
    concurrency=4,
)(col("video_url"), col("task_desc")))

文档与文本算子

本节列出文档与文本算子的输入数据列、业务参数与使用示例。执行参数与基础设施参数统一见附录。

TextGeneration（纯文本生成）

通过 EMR 内置 Model Manager 完成文本生成，返回完整 EMR struct。

输出（RETURN_DTYPE）：struct(content, reasoning_content, finish_reason, prompt_tokens, completion_tokens, total_tokens, cached_tokens, reasoning_tokens, model, id, error)

输入数据列（__call__）

中文名	英文参数名	必选	类型/取值
原始文本列表	`raw_texts`	是	list[str]，每行一条输入

业务参数（construct_args）

中文名	英文参数名	必选	默认值	取值范围/可选值
用户提示模板	`prompt`	否	None	字符串，支持 {query} 占位符
模型名	`model`	否	qwen3.6-plus	字符串
系统提示	`system_message`	否	None	字符串

通用 LLM 生成参数（max_tokens / temperature / top_p / stop / frequency_penalty / presence_penalty / logit_bias / tools / enable_thinking / llm_config）见附录 D

执行参数：max_concurrency（100）、request_timeout（1200s）

使用示例

df = df.with_column("generated", emr_udf(
    TextGeneration,
    construct_args={"model": "qwen3.6-plus", "prompt": "总结以下内容：{query}"},
    concurrency=8,
)(raw_texts=col("text")))

通用约定

参数必选性：业务参数看 __init__ 签名，输入数据列看 __call__ 签名。签名中没有默认值的参数为必选，有默认值的为可选。
路径协议：各算子支持的协议范围不同。使用 oss://、s3://、hdfs://、http(s):// 或本地路径前，请查看对应算子的说明，同时确认是否需要本地临时盘。
失败行为：处理失败时，算子可能返回 None、空字符串、空列表或带 error 字段的 struct。服务型算子在 service_error_policy 设置为 raise 时会中断整个作业。
模型依赖：AI 算子的服务能力由内置 Model Manager 统一提供。使用镜像内本地模型的算子，请查看对应算子的 GPU 需求、模型目录和默认模型。

使用限制

ImageQualityScore 需从 daft.emr.functions.media 导入，TextGeneration 需从 daft.emr.functions.bailian 导入。这两个算子暂不支持从 daft.emr.functions 顶层导入。
算子的可用性与引擎镜像版本相关。如果调用时提示算子不存在，请确认当前作业使用的引擎版本。

附录

以下参数为多数算子通用，各算子文档中不再重复列出。

附录 A：通用存储/IO 参数

几乎所有基础算子都支持下列参数，用于访问远端存储（OSS 或 S3）。这些是基础设施参数，产品表单中由「输入类型 / OSS 路径」区域统一处理，不在「算子参数」中重复展示。

中文名	英文参数名	默认值	类型
IO 配置对象	`io_config`	None	IOConfig
OSS 端点	`oss_endpoint`	None	字符串
OSS AK ID	`oss_access_key_id`	None	字符串
OSS AK Secret	`oss_access_key_secret`	None	字符串

凭证优先级：io_config > oss_* 参数 > daft.emr.set_config() 全局配置 > 环境变量（OSS_ENDPOINT / ACCESS_KEY_ID / ACCESS_KEY_SECRET）。

附录 B：emr_udf 包装参数

emr_udf() 本身的参数，控制资源与并发（非算子业务逻辑）。

中文名	英文参数名	默认值	说明
每实例 CPU 数	`num_cpus`	1	浮点数
每实例 GPU 数	`num_gpus`	0	浮点数
最大并发实例数	`concurrency`	1	整数
批大小	`batch_size`	None	整数（None=默认）
服务请求并发	`service_batch_concurrency`	None	仅服务型 operator；单 actor 内并发请求数
单请求行数	`service_batch_rows_per_request`	None	仅服务型 operator；每次服务请求打包的行数
服务错误策略	`service_error_policy`	isolate	isolate=隔离失败行；raise=失败时中断作业

附录 C：执行/性能参数说明

各算子的执行参数（仅影响性能/资源，不改变输出语义），常见含义：

英文参数名	含义	出现的算子
timeout	`ffmpeg/请求超时（秒）`	多数视频基础算子、VideoRiskRec
request_timeout	`LLM HTTP 请求超时（秒）`	多数 AI 算子
batch_size	`模型推理批大小`	ImageEmbedding/ImageQualityScore/VideoKeyframeExtract/VideoMotionScore
max_concurrency	`最大并发请求数`	VisionUnderstanding/TextGeneration
num_workers	`并行工作线程数`	VideoMotionScore
use_cuda / device / rank	`GPU 启用 / 设备 / 设备索引`	GPU 相关算子
dtype	`推理精度（float16/float32/bfloat16）`	CLIP 相关算子
服务接入	`服务型 AI 算子统一使用 EMR 内置 Model Manager；本地模型算子使用镜像内模型`	客户无需配置服务地址、鉴权密钥或路由开关；本地模型依赖按正文与 0.5 表确认

附录 D：通用 LLM 生成参数（AI 算子）

除 VideoFineUnderstanding 外，AI 算子通过内置 Model Manager 透传以下生成参数。各算子支持的子集略有差异（以正文「通用 LLM 生成参数」标注为准）。

中文名	英文参数名	默认值	取值范围/类型
最大生成 token	`max_tokens`	None	整数
生成 token 上限	`max_completion_tokens`	None	整数
采样温度	`temperature`	None	浮点数 [0, 2]
核采样阈值	`top_p`	None	浮点数
停止词列表	`stop`	None	list[str]
频率惩罚	`frequency_penalty`	None	浮点数 [-2.0, 2.0]
存在惩罚	`presence_penalty`	None	浮点数 [-2.0, 2.0]
Token 概率偏置	`logit_bias`	None	dict[str, int]
是否返回 log 概率	`logprobs`	None	bool
返回 top-k 对数概率	`top_logprobs`	None	整数
响应格式	`response_format`	None	dict
工具列表	`tools`	None	list[dict]
工具选择策略	`tool_choice`	None	字符串或 dict
启用深度思考	`enable_thinking`	None	bool
额外生成参数 catch-all	`llm_config`	None	dict（显式参数覆盖同名键）