Processing Audio Data in LakeInsight

本演示主要内容：

1. 从 LakeSoul 多模态湖仓中读取音频数据

2. 使用 Daft，调用 whisper 模型将音频转为文字

可以进入到 LakeInsight 演示环境，打开 演示/display_audio.ipynb 查看演示代码。

1. 环境初始化与参数配置

本步骤导入演示所需的组件：

Ray Data：从 LakeSoul 表读取音频数据。
LakeSoul Ray Connector：注册 ray.data.read_lakesoul() 接口。
Daft：将音频转录过程组织为 DataFrame 计算任务。
faster-whisper：实际执行语音识别。
Pandas / HTML：生成包含音频播放器的结果表格。

主要参数：

TABLE：要读取的 LakeSoul 表。
NAMESPACE：LakeSoul 命名空间。
LIMIT：本次演示处理的音频数量。
MODEL：本地 faster-whisper 模型目录。

import os
import warnings
import logging
logging.disable(logging.WARNING)

import daft
import lakesoul.ray  # Registers ray.data.read_lakesoul.
import ray
import base64
import html
import re
import pandas as pd
from daft import DataType, col
from IPython.display import HTML, display


TABLE = "librispeech_minimal_parquet_s3"

NAMESPACE = "default"
LIMIT = 4
MODEL = "/opt/models/faster-whisper-base.en"

2. 文本标准化与 WER 计算

数据集中的 answer 是人工提供的标准答案，Whisper 输出的是预测文本。

两者可能只有大小写或标点不同，例如：

标准答案：A ROUTE SLIGHTLY LESS DIRECT THAT'S ALL
预测结果：a route slightly less direct. That's all.

比较前需要统一大小写并去除普通标点，避免把格式差异当作识别错误。

def normalize(text):
    return re.sub(r"[^A-Z0-9']+", " ", text.upper()).strip()

3. 从 LakeSoul 读取音频数据

使用 ray.data.read_lakesoul() 读取 LakeSoul 表。

本示例需要以下字段：

sample_id：样本唯一标识，用于定位和关联记录。
context.bytes：存储在 LakeSoul 中的 FLAC 音频二进制。
answer：原始数据集提供的标准转录文本。

读取完成后，将嵌套字段转换为更适合 Daft 处理的结构：

context.bytes → audio_bytes
answer → reference_text batch_size 和 thread_count 被设置得较小，用于控制演示环境中的并发和内存占用。

if not ray.is_initialized():
    ray.init(
        num_cpus=1,
        include_dashboard=False, # 剔除 ray 日志
        log_to_driver=False,
        logging_level=logging.ERROR,
    )

# 读取LakeSoul表
source = ray.data.read_lakesoul(
    TABLE,
    namespace=NAMESPACE,
    batch_size=LIMIT,
    thread_count=1,
    retain_partition_columns=True,
)

4. 使用 Daft 和 faster-whisper 转录

Daft 本身不负责语音识别，它负责组织 DataFrame 计算并通过 Ray 调度 UDF。

WhisperTranscriber 是一个 Daft 有状态 UDF：

Actor 初始化时加载一次 faster-whisper 模型。
每行输入一段 FLAC 音频二进制。
将二进制临时写入音频文件。
调用 faster-whisper 进行英文语音识别。
合并所有识别片段，生成 prediction 字段。

max_concurrency=1 限制同一个模型实例一次只处理一个任务，降低 CPU 和内存压力。

# 设置 Daft 执行器
daft.set_runner_ray(noop_if_initialized=True)

@daft.cls(cpus=1, max_concurrency=1)
class WhisperTranscriber:
    def __init__(self): # 加载 Whisper 模型
        from faster_whisper import WhisperModel

        self.model = WhisperModel(
            MODEL,
            device="cpu", # cpu 推理
            compute_type="int8",
        )
    # 音频转录文本方法 
    @daft.method(return_dtype=DataType.string())
    def transcribe(self, audio_bytes: bytes) -> str:
        import tempfile

        with tempfile.NamedTemporaryFile(suffix=".flac") as audio_file:
            audio_file.write(audio_bytes)
            audio_file.flush()
            segments, _ = self.model.transcribe( # 执行 Whisper
                audio_file.name,
                language="en",
                beam_size=1,
                temperature=0,
            )
            return " ".join(segment.text.strip() for segment in segments)


transcriber = WhisperTranscriber()

limited_rows = source.limit(LIMIT)
print("本次处理行数:", limited_rows.count())


audio_df = (
    daft.from_ray_dataset(limited_rows) # ray dataset 转为 daft dataframe
    .select(
        col("context")["bytes"].alias("audio_bytes"),
        col("answer").alias("reference_text"),
    ).into_partitions(1) # 使用一个分区节省内存
)

results = (
    audio_df
    .with_column(
        "prediction",
        transcriber.transcribe(col("audio_bytes")), # 转录音频
    )
    .select(
        "audio_bytes",
        "reference_text",
        "prediction",
      )
    .to_arrow()
    .to_pylist()
)

本次处理行数: 4

5. 展示并验证转录结果

最终结果表格包含：

音频：根据音频二进制生成的浏览器播放器。
标准答案：LakeSoul 表中保存的标准答案。
Whisper 转录结果：Whisper 重新转录得到的预测文本。

comparison_rows = []

for row in results:
    reference = row["reference_text"] # 数据集标准答案
    prediction = row["prediction"]    # 转录之后重新识别的文本
    audio_base64 = base64.b64encode(row["audio_bytes"]).decode("ascii") # 二进制编码成base64，使其可以嵌入notebook html

    audio_player = (
        '<audio controls preload="none">'
        f'<source src="data:audio/flac;base64,{audio_base64}" '
        'type="audio/flac">'
        "</audio>"
    )

    comparison_rows.append({
      "音频": audio_player,
      "标准答案": html.escape(reference),
      "Whisper 转录结果": html.escape(prediction),
  })

comparison_df = pd.DataFrame(comparison_rows) # 列表转换为Pandas DataFrame

display(
    HTML(
        comparison_df.to_html(
            escape=False,
            index=False,
        )
    )
)

音频	标准答案	Whisper 转录结果
	A ROUTE SLIGHTLY LESS DIRECT THAT'S ALL	a route slightly less direct, that's all.
	NANCY'S CURLY CHESTNUT CROP SHONE IN THE SUN AND OLIVE'S THICK BLACK PLAITS LOOKED BLACKER BY CONTRAST	Nancy's curly chestnut crop, shown in the sun and olive's thick black plates looked blacker by contrast.
	THERE BEFELL AN ANXIOUS INTERVIEW MISTRESS FITZOOTH ARGUING FOR AND AGAINST THE SQUIRE'S PROJECT IN A BREATH	There befell an anxious interview, Mistress Fitzhuth arguing for and against the Squire's project in a breath.
	ROBIN FITZOOTH SAW THAT HIS DOUBTS OF WARRENTON HAD BEEN UNFAIR AND HE BECAME ASHAMED OF HIMSELF FOR HARBORING THEM	Robin Fitzhuth saw that his doubts of warrant and had been unfair, and he became ashamed of himself for harboring them.

Processing Audio Data in LakeInsight

本演示主要内容：​

1. 从 LakeSoul 多模态湖仓中读取音频数据​

2. 使用 Daft，调用 whisper 模型将音频转为文字​

1. 环境初始化与参数配置​

2. 文本标准化与 WER 计算​

3. 从 LakeSoul 读取音频数据​

4. 使用 Daft 和 faster-whisper 转录​

5. 展示并验证转录结果​