音频编解码与变换-实时计算 Flink版(Flink)-阿里云帮助中心

使用限制

仅实时计算引擎 VVR 11.8 及以上版本支持。
本文使用音频片段引用 (AUDIO_CLIP_REF_TYPE) 和解码后的音频波形 (AUDIO_WAVEFORM_TYPE) 为 pyflink.multimodal.types 中封装的 DataFrame API 自定义类型，具体定义见文档音频算子使用概述

算子清单

分类	算子	说明
解码	audio_decode	将 `DataType.binary()`、URI 或 `AUDIO_CLIP_REF_TYPE` 解码为 float32 PCM `AUDIO_WAVEFORM_TYPE` 。适合在重采样、声道转换、语音检测、切分或识别前统一输入，例如把 MP3、WAV 或 FLAC 文件转换为下游波形算子可直接处理的样本数据。
解码	audio_encode	将 float32 PCM `AUDIO_WAVEFORM_TYPE` 编码为 WAV、FLAC、AIFF、MP3 或 OGG 字节。适合在波形完成变换、拼接或切分后落盘、传输，或交给只接受编码文件的系统，例如把标准化后的 16 kHz 单声道波形保存为 WAV。
格式转换	audio_convert_format	先解码输入，再按目标格式重新编码为新字节。适合在下游要求特定容器或编码格式时使用，例如将来源不统一的音频归一成 WAV、FLAC 等格式。
	audio_standardize	先转换声道，再重采样，将音频波形统一为指定采样率和声道数。适合在 ASR、VAD 或模型推理前规范不同来源的输入。
	audio_resample	使用 soxr HQ 改变音频波形的采样率，同时保留原声道数。适合下游只要求特定采样率、但需要保持原声道布局时使用，例如把 48 kHz 立体声转换为 16 kHz 立体声。
	audio_concat	按数组顺序拼接多个采样率、声道数和采样格式一致的音频波形。适合重组切分片段或合并分段处理结果，例如把并行降噪后的连续片段恢复为一段音频。

通用 Runtime 参数

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

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

音频解码与编码

audio_decode

把编码音频或音频片段引用解码为 float32 PCM 波形。

输入类型： DataType.binary()（编码音频字节列）、DataType.string()（音频 URI 列）或 AUDIO_CLIP_REF_TYPE（音频片段引用列）。

函数签名：

audio_decode(
    *columns,
    on_error="raise",
    max_decoded_bytes=1024 * 1024 * 1024,
    concurrency=None
)

参数	类型	默认值	说明
`on_error`	`str`	`"raise"`	`"raise"` 表示抛出解码异常；`"null"` 表示对无法解码的受支持媒体输入返回空值。Python 默认值与 SQL 不同。
`max_decoded_bytes`	`int`	`1073741824`	单行解码后的字节数上限，必须大于 0 且不超过 2147483647。

空值输入返回 None。URI 无法访问、输入类型不受支持和无效片段引用仍会抛出异常。

返回类型： AUDIO_WAVEFORM_TYPE。

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

result = df.with_column(
    "waveform",
    audio_decode(
        col("audio_bytes"),
        on_error="null",
        max_decoded_bytes=512 * 1024 * 1024
    )
)

audio_encode

把音频波形编码为指定格式的字节。

输入类型：AUDIO_WAVEFORM_TYPE，解码后的音频波形列。

函数签名：

audio_encode(
    *columns,
    format="wav",
    concurrency=None
)

参数	类型	默认值	说明
`format`	`str`	`"wav"`	目标音频格式。

支持的格式、别名和默认编码子类型如下：

格式	可用名称	默认编码子类型
WAV	`"wav"`、`"wave"`	`PCM_16`
FLAC	`"flac"`	`PCM_16`
OGG	`"ogg"`、`"oga"`	`VORBIS`
AIFF	`"aiff"`、`"aif"`	`PCM_16`
MP3	`"mp3"`、`"mpeg"`	`MPEG_LAYER_III`

若当前 libsndfile 不支持目标格式或默认编码子类型，算子会抛出异常。

空值输入返回 None；输入不是合法音频，或波形含 NaN、Inf 时会抛出异常。

返回类型： DataType.binary()。

from pyflink.multimodal.operators import audio_encode

result = df.with_column(
    "wav_bytes",
    audio_encode(
        col("waveform"),
        format="wav"
    )
)

audio_convert_format

把编码音频转换为另一种编码格式。该算子先解码，再按目标格式重新编码。

输入类型： DataType.binary() 或 DataType.string()，编码音频字节列或音频 URI 列。

函数签名：

audio_convert_format(
    *columns,
    format="wav",
    on_error="raise",
    max_decoded_bytes=1024 * 1024 * 1024,
    concurrency=None
)

参数	类型	默认值	说明
`format`	`str`	`"wav"`	目标音频格式。
`on_error`	`str`	`"raise"`	`"raise"` 表示抛出解码异常；`"null"` 表示把受支持输入的解码失败转为空值。Python 默认值与 SQL 不同。
`max_decoded_bytes`	`int`	`1073741824`	转换过程中允许的最大中间 PCM 字节数。

URI 访问错误、编码错误、格式不受支持和目标编码子类型不可用仍会使作业失败。空值输入返回 None。

返回类型： DataType.binary()。

from pyflink.multimodal.operators import audio_convert_format

result = df.with_column(
    "wav_bytes",
    audio_convert_format(
        col("audio_uri"),
        format="wav",
        on_error="null"
    )
)

音频波形变换

audio_standardize

把音频波形统一为指定采样率和声道数。

输入类型： AUDIO_WAVEFORM_TYPE，解码后的音频波形列。

函数签名：

audio_standardize(
    *columns,
    sample_rate=16000,
    channels=1,
    concurrency=None
)

参数	类型	默认值	说明
`sample_rate`	`int`	`16000`	目标采样率，单位为 Hz，必须大于 0。
`channels`	`int`	`1`	目标声道数，必须大于 0。

算子先转换声道数，再使用 soxr HQ 模式重采样。多声道转单声道时按声道平均；单声道转多声道时复制声道；不同多声道数之间转换时先混为单声道，再复制为目标声道数。输出保留来源 URI 和片段时间字段。

返回类型： AUDIO_WAVEFORM_TYPE。

from pyflink.multimodal.operators import audio_standardize

result = df.with_column(
    "speech_waveform",
    audio_standardize(
        col("waveform"),
        sample_rate=16000,
        channels=1
    )
)

audio_resample

只改变音频波形的采样率，保留声道数。

输入类型： AUDIO_WAVEFORM_TYPE，解码后的音频波形列。

函数签名：

audio_resample(
    *columns,
    sample_rate,
    concurrency=None
)

参数	类型	默认值	说明
`sample_rate`	`int`	必填	目标采样率，单位为 Hz，必须大于 0。

算子使用 soxr HQ 模式重采样，并保留声道数、来源 URI 和片段时间字段。输出帧数为：

ceil(input_frames * target_sample_rate / input_sample_rate)

算子会在必要时裁剪或补零，使结果符合该帧数。目标采样率与输入相同时，算子完成输入校验后返回原波形。

返回类型： AUDIO_WAVEFORM_TYPE。

from pyflink.multimodal.operators import audio_resample

result = df.with_column(
    "resampled",
    audio_resample(
        col("waveform"),
        sample_rate=8000,
    ),
)

audio_concat

按数组顺序拼接多个音频波形。

输入类型： DataType.list(AUDIO_WAVEFORM_TYPE)，非空的音频波形数组列。

函数签名：

audio_concat(
    *columns,
    concurrency=None,
)

输入数组为 None 时返回 None，空数组抛出异常。
所有波形必须具有相同的采样率、声道数和采样格式，并满足 float32、交错排列的波形约束。
只有一个输入时，原样保留其来源 URI 和起止时间，包括只设置部分来源字段的情况。
多个输入只有在 source_uri 全部相同、每段都有完整起止时间，并且后一段起点等于前一段终点时，才保留合并后的来源和起止时间；否则三个来源字段都置为 None。

返回类型： AUDIO_WAVEFORM_TYPE。

from pyflink.multimodal.operators import audio_concat

result = df.with_column(
    "joined",
    audio_concat(col("clips")),
)