音频函数使用概述-实时计算 Flink版(Flink)-阿里云帮助中心

本文介绍音频函数的使用限制、使用方法、音频数据形态，以及音频处理 Pipeline 示例。

使用限制

仅实时计算引擎 VVR 11.8 及以上版本支持。

使用方法

使用说明

在平台 SQL 作业中，音频函数由内置 python 模块提供，无需在作业中引入额外依赖。使用前需要先加载 python 模块，再启用 python 和 core 模块：

LOAD MODULE python;
USE MODULES python, core;

基本调用

LOAD MODULE python;
USE MODULES python, core;

SELECT
  AUDIO_METADATA(audio_bytes).duration_ms AS duration_ms,
  AUDIO_STANDARDIZE(AUDIO_DECODE(audio_bytes), 16000, 1) AS waveform
FROM raw_audio;

参数调用

SQL 函数按签名中的顺序传参，只支持位置参数。函数签名中的 [ ] 表示可选参数。例如，AUDIO_DETECT_SPEECH 的签名如下：

AUDIO_DETECT_SPEECH(
  audio_input
  [, aggressiveness
  [, frame_ms
  [, min_speech_ms
  [, merge_gap_ms
  [, max_segments]]]]]
)

调用时遵循以下规则：

省略的可选参数会使用参数默认值进行填充。
如果需要显式设置靠后的可选参数，前面的可选参数也需要明确写出。
NULL 会作为参数值传入，不等同于省略参数时的默认值。多态数据参数需要使用 NULL 时，可以通过 CAST(NULL AS <精确类型>) 明确类型。
除逐行音频输入（例如 audio_input 和 audio_inputs）、timestamps 和 speech_activity 等数据参数外，其他参数都是配置参数，必须直接使用 SQL 字面量，不能使用表字段或 CAST 表达式。

数据类型说明

音频数据形态

数据形态	SQL 类型	说明
编码音频	`BYTES`	编码后的音频字节。
音频 URI	`STRING`	Flink 文件系统可访问的音频 URI。
音频片段引用	`ROW<uri STRING, start_time_ms BIGINT, end_time_ms BIGINT>`	指向编码音频中的一个时间片段，不复制音频内容。
解码后的音频波形	`AUDIO_WAVEFORM` 对应的 `ROW`	float32、交错排列的 PCM 样本以及采样率、声道数和来源信息。

不同函数接受的数据形态不同，请以各函数的参数表为准。

音频片段引用

ROW<
  uri STRING,
  start_time_ms BIGINT,
  end_time_ms BIGINT
>

音频片段采用半开区间 [start_time_ms, end_time_ms)，结束时间不包含在片段内。相邻片段的后一个起点等于前一个终点。时间必须为非负整数；NULL 起点或终点分别表示文件开头或文件结尾。

解码后的音频波形

ROW<
  data TENSOR('float32'),
  sample_rate INT,
  channels INT,
  frames BIGINT,
  sample_format STRING,
  layout STRING,
  duration_ms BIGINT,
  source_uri STRING,
  start_time_ms BIGINT,
  end_time_ms BIGINT
>

字段	类型	说明
`data`	`TENSOR('float32')`	形状为 `(frames, channels)` 的 float32 PCM 样本。
`sample_rate`	`INT`	采样率，单位为 Hz。
`channels`	`INT`	声道数。
`frames`	`BIGINT`	每个声道的采样帧数。
`sample_format`	`STRING`	当前实现要求为 `'float32'`。
`layout`	`STRING`	当前实现要求为 `'interleaved'`。
`duration_ms`	`BIGINT`	波形时长，单位为毫秒。
`source_uri`	`STRING`	来源 URI；字节输入或无法保留来源时为 `NULL`。
`start_time_ms`	`BIGINT`	波形在来源音频中的开始时间。
`end_time_ms`	`BIGINT`	波形在来源音频中的结束时间，不包含该时刻。

合法波形要求 sample_rate、channels 和 frames 为正数，样本数量与形状一致，且不能包含 NaN 或无穷值。

音频元数据

ROW<
  sample_rate INT,
  channels INT,
  frames BIGINT,
  duration_ms BIGINT,
  format STRING,
  codec STRING,
  bit_rate BIGINT,
  size BIGINT
>

字段	类型	说明
`sample_rate`	`INT`	采样率，单位为 Hz。
`channels`	`INT`	声道数。
`frames`	`BIGINT`	采样帧数。
`duration_ms`	`BIGINT`	时长，单位为毫秒。
`format`	`STRING`	容器格式。
`codec`	`STRING`	尽力获取的编码器短名，例如 `pcm_s16le`。
`bit_rate`	`BIGINT`	比特率，无法确定时为 `NULL`。
`size`	`BIGINT`	编码对象大小，单位为字节。

音频时间范围

静音和语音检测函数返回：

ROW<start_ms BIGINT, end_ms BIGINT, duration_ms BIGINT>

时间戳切分和语音切分也接受不含 duration_ms 的形式：

ROW<start_ms BIGINT, end_ms BIGINT>

时间范围采用半开区间 [start_ms, end_ms)。检测和切分参数中的时间通常使用毫秒；AUDIO_DURATION 和 AUDIO_DURATION_FILTER 使用秒。物化波形片段时，毫秒边界需要换算为采样帧并进行取整，因此非整帧边界可能产生轻微的时间偏差。

完整 Pipeline 示例

以客服录音预处理为例：raw_audio 表存放原始录音字节（audio_bytes BINARY，来自对象存储或消息队列），下游需要按语音区间切出片段用于 ASR。

以下 Pipeline 依次完成解码、标准化到 16 kHz 单声道、检测语音活动区间、切分为独立片段并计算每段时长。

LOAD MODULE python;
USE MODULES python, core;

-- 步骤 1：解码原始音频字节，并将音频标准化一为 16 kHz 单声道波形。
WITH decoded AS (
  SELECT
    id,
    AUDIO_STANDARDIZE(
      AUDIO_DECODE(audio_bytes),
      16000,
      1
    ) AS waveform
  FROM raw_audio
),

-- 步骤 2：过滤解码失败的音频，并检测每段波形中的语音区间。
detected AS (
  SELECT
    id,
    waveform,
    AUDIO_DETECT_SPEECH(
      waveform,
      2,
      30,
      100,
      50,
      1024
    ) AS speech_ranges
  FROM decoded
  WHERE waveform IS NOT NULL
)

-- 步骤 3：根据检测到的语音区间，将每条音频拆分成多个语音片段，并计算每段时长。
SELECT
  t.id,
  s.segment,
  AUDIO_DURATION(s.segment) AS segment_duration_seconds
FROM detected AS t
CROSS JOIN LATERAL TABLE(
  AUDIO_SPLIT_BY_SPEECH(
    t.waveform,
    t.speech_ranges,
    'audio',
    200,
    200,
    50,
    100,
    NULL,
    1024
  )
) AS s(segment);