intermediateView Python source

Transcribe audio files with Whisper

This example transcribes 73 short clips of read English speech (about 8 minutes in all) with Whisper tiny.en, running on the CPU inside the engine. It then compares each transcript with the clip's reference transcript in SQL and reports the character error rate.

Run it

Run all commands below from the repository root (sail-examples/):

uv run multimodal/transcribe-audio/main.py

The first run downloads the sample clips (a 9 MB Parquet file) and the Whisper tiny.en model converted for CTranslate2 (about 80 MB) from Hugging Face into the local Hugging Face cache. No account or key is needed. The dataset and model revisions are pinned.

On these 73 sample clips, the transcripts have a character error rate of about 5% against the references. Some differences are spelling choices rather than mistakes: Whisper writes "Mr." where the reference has "MISTER".

Output

The newest capture appears first. Older output stays here when a result changes.

Recorded output · Sail 0.7.3

Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.

The example has changed since this capture. Rerun it to refresh the output.

+----------------+-------+--------------------------------------------------------------------------------+
|              id|seconds|                                                                      transcript|
+----------------+-------+--------------------------------------------------------------------------------+
|1272-128104-0000|  5.855|Mr. Quilter is the apostle of the middle classes and we are glad to welcome h...|
|1272-128104-0001|  4.815|                   Nor is Mr. Quilter's manner less interesting than his matter.|
|1272-128104-0002| 12.485|He tells us that at this festive season of the year, with Christmas and roast...|
|1272-128104-0003|    9.9|He has grave doubts whether Sir Frederick Layton's work is really Greek after...|
|1272-128104-0004|   29.4|Linnell's pictures are a sort of up-guards and atom paintings, and Mason's ex...|
+----------------+-------+--------------------------------------------------------------------------------+
only showing top 5 rows

+-----+-------+--------------------+
|clips|minutes|character_error_rate|
+-----+-------+--------------------+
|   73|   8.02|0.054469721243191285|
+-----+-------+--------------------+

Python code

main.py148 lines

View main.py on GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "certifi==2026.10.19",
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
#   "numpy==2.5.4",
#   "faster-whisper==1.2.1",
#   "ctranslate2==4.8.2",
#   "av==18.0.0",
#   "onnxruntime==1.31.1",
#   "tokenizers==0.22.1",
#   "huggingface-hub==0.36.0",
#   "tqdm==4.70.1",
# ]
# ///
"""Transcribe a folder of audio files with Whisper inside the engine, and score the transcripts.

The pattern: a table of audio paths goes through `mapInArrow`, which loads the speech model once
per partition and returns one transcript per file. SQL then compares each transcript with a
reference transcript and computes the character error rate.
"""

import atexit
import shutil
import tempfile
from pathlib import Path

import pyarrow as pa
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F

MODEL_REPO = "Systran/faster-whisper-tiny.en"  # OpenAI Whisper tiny.en, converted for CTranslate2
MODEL_REVISION = "0d3d19a32d3338f10357c0889762bd8d64bbdeba"  # pinned: every run uses the same weights
MODEL_FILES = ["model.bin", "config.json", "tokenizer.json", "vocabulary.txt"]
# 73 short read-speech clips from the LibriSpeech corpus (CC BY 4.0), with reference transcripts.
CLIPS_REPO = "hf-internal-testing/librispeech_asr_dummy"
CLIPS_REVISION = "5be91486e11a2d616f4ec5db8d3fd248585ac07a"
CLIPS_FILE = "clean/validation-00000-of-00001.parquet"

ARROW_SCHEMA = pa.schema([("id", pa.string()), ("seconds", pa.float64()), ("transcript", pa.string())])
SCHEMA = "id string, seconds double, transcript string"


def temp_dir(prefix: str) -> str:
    """A temporary directory that is removed when the process exits."""
    path = tempfile.mkdtemp(prefix=prefix)
    atexit.register(shutil.rmtree, path, ignore_errors=True)
    return path


def download_clips(directory: Path) -> list[tuple[str, str, str]]:
    """Save the sample clips as FLAC files and return (id, path, reference transcript) per clip."""
    import pyarrow.parquet as pq
    from huggingface_hub import hf_hub_download

    table = pq.read_table(
        hf_hub_download(CLIPS_REPO, CLIPS_FILE, repo_type="dataset", revision=CLIPS_REVISION)
    )
    clips = []
    for row in table.select(["id", "audio", "text"]).to_pylist():
        path = directory / f"{row['id']}.flac"
        path.write_bytes(row["audio"]["bytes"])
        clips.append((row["id"], str(path), row["text"]))
    return clips


def transcribe(batches):
    """mapInArrow function: everything before the loop runs once per partition."""
    import threading

    from faster_whisper import WhisperModel
    from huggingface_hub import hf_hub_download
    from tqdm import tqdm

    # faster-whisper creates a tqdm progress bar even when progress is off, and tqdm's default lock
    # starts a multiprocessing helper process. In local mode this UDF runs inside the Sail server
    # process, so swap in a thread lock (a setting for the whole process) and start no process there.
    tqdm.set_lock(threading.RLock())

    # Download where the task runs (cached per machine), so this also works on cluster workers.
    files = [hf_hub_download(MODEL_REPO, name, revision=MODEL_REVISION) for name in MODEL_FILES]
    model_dir = str(Path(files[0]).parent)
    # One thread per model: the engine runs partitions in parallel already.
    model = WhisperModel(model_dir, device="cpu", compute_type="float32", cpu_threads=1)

    for batch in batches:
        rows = {name: [] for name in ARROW_SCHEMA.names}
        for clip_id, path in zip(batch.column("id").to_pylist(), batch.column("path").to_pylist()):
            # One temperature of 0 uses greedy decoding without sampled retries.
            segments, info = model.transcribe(
                path, language="en", beam_size=1, temperature=0.0, condition_on_previous_text=False
            )
            rows["id"].append(clip_id)
            rows["seconds"].append(round(info.duration, 3))
            rows["transcript"].append(" ".join(segment.text.strip() for segment in segments))
        yield pa.RecordBatch.from_pydict(rows, schema=ARROW_SCHEMA)


def normalized(column: str):
    """Lowercase, letters, digits and apostrophes only, single spaces: compare words, not punctuation."""
    text = F.regexp_replace(F.lower(F.col(column)), r"[^a-z0-9' ]", " ")
    return F.trim(F.regexp_replace(text, r" +", " "))


def run(spark: SparkSession) -> dict[str, DataFrame]:
    # A table of paths, one row per audio file, with the reference transcript to score against.
    clips = spark.createDataFrame(
        download_clips(Path(temp_dir("clips-"))), "id string, path string, reference string"
    )

    # Transcribe, then save once: the queries below must not run the model again.
    transcripts_path = temp_dir("transcripts-")
    clips.select("id", "path").repartition(4).mapInArrow(transcribe, SCHEMA).write.mode("overwrite").parquet(
        transcripts_path
    )
    transcripts = spark.read.parquet(transcripts_path).join(clips.select("id", "reference"), "id")

    # Character error rate: edits needed to turn the transcript into the reference, per reference character.
    scored = transcripts.select(
        "id",
        "seconds",
        "transcript",
        F.levenshtein(normalized("transcript"), normalized("reference")).alias("edits"),
        F.length(normalized("reference")).alias("reference_chars"),
    )
    summary = scored.agg(
        F.count("*").alias("clips"),
        F.round(F.sum("seconds") / 60, 2).alias("minutes"),
        (F.sum("edits") / F.sum("reference_chars")).alias("character_error_rate"),
    )
    return {"transcripts": scored.orderBy("id"), "summary": summary}


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    results = run(spark)
    results["transcripts"].select("id", "seconds", "transcript").show(5, truncate=80)
    results["summary"].show()
    spark.stop()
    server.stop()

full_test_clean.py149 lines

View full_test_clean.py on GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "certifi==2026.10.19",
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
#   "numpy==2.5.4",
#   "faster-whisper==1.2.1",
#   "ctranslate2==4.8.2",
#   "av==18.0.0",
#   "onnxruntime==1.31.1",
#   "tokenizers==0.22.1",
#   "huggingface-hub==1.5.0",
#   "tqdm==4.70.1",
#   "transformers==5.10.1",
#   "regex==2026.9.29",
# ]
# ///
"""The full test-set evaluation: transcribe all of LibriSpeech test-clean and compute the word error rate.

Uses the same transcription function as main.py on the 2,620 utterances of the standard
test-clean split, normalizes text the way the Whisper paper did, and computes the corpus word
error rate in SQL, for comparison with the paper's published figure for the same model.
"""

import json
import sys
from pathlib import Path

import main
from main import temp_dir, transcribe
from pyspark import cloudpickle
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf

# Send main.py's code with the UDF, not a reference to the module, so workers need no copy of main.py.
cloudpickle.register_pickle_by_value(main)

TEST_REPO = "openslr/librispeech_asr"  # LibriSpeech (CC BY 4.0)
TEST_REVISION = "71cacbfb7e2354c4226d01e70d77d5fca3d04ba1"
TEST_FILE = "clean/test/0000.parquet"  # 350 MB
NORMALIZER = ("openai/whisper-tiny.en", "normalizer.json", "87c7102498dcde7456f24cfd30239ca606ed9063")
# Whisper paper (Radford et al., 2022), Appendix D.1.1, Table 8: tiny.en on LibriSpeech test-clean,
# greedy decoding, word error rate in percent.
PUBLISHED_WER = 5.6


def write_clips(directory: Path) -> list[tuple[str, str, str]]:
    """Save every test-clean utterance as a FLAC file; return (id, path, reference transcript)."""
    import pyarrow.parquet as pq
    from huggingface_hub import hf_hub_download

    source = pq.ParquetFile(
        hf_hub_download(TEST_REPO, TEST_FILE, repo_type="dataset", revision=TEST_REVISION)
    )
    clips = []
    for batch in source.iter_batches(batch_size=200, columns=["id", "audio", "text"]):  # bounded memory
        for row in batch.to_pylist():
            path = directory / f"{row['id']}.flac"
            path.write_bytes(row["audio"]["bytes"])
            clips.append((row["id"], str(path), row["text"]))
    return clips


def run(spark: SparkSession) -> DataFrame:
    clips = spark.createDataFrame(
        write_clips(Path(temp_dir("test-clean-"))), "id string, path string, reference string"
    )
    transcripts_path = temp_dir("test-clean-transcripts-")
    clips.select("id", "path").repartition(8).mapInArrow(
        transcribe, "id string, seconds double, transcript string"
    ).write.mode("overwrite").parquet(transcripts_path)

    @pandas_udf("string")
    def normalized(texts):
        """Whisper's English text normalizer, as used for the paper's word error rates."""
        from huggingface_hub import hf_hub_download
        from transformers.models.whisper.english_normalizer import EnglishTextNormalizer

        repo, name, revision = NORMALIZER
        normalize = EnglishTextNormalizer(
            json.loads(Path(hf_hub_download(repo, name, revision=revision)).read_text())
        )
        return texts.map(normalize)

    @pandas_udf("long")
    def word_errors(hypotheses, references):
        """Word-level edit distance: substitutions, deletions and insertions."""
        import pandas as pd

        def distance(hypothesis: str, reference: str) -> int:
            hyp, ref = hypothesis.split(), reference.split()
            previous = list(range(len(hyp) + 1))
            for i, word in enumerate(ref, start=1):
                current = [i]
                for j, other in enumerate(hyp, start=1):
                    current.append(
                        min(previous[j] + 1, current[j - 1] + 1, previous[j - 1] + (word != other))
                    )
                previous = current
            return previous[-1]

        return pd.Series([distance(h, r) for h, r in zip(hypotheses, references)])

    scored = (
        spark.read.parquet(transcripts_path)
        .join(clips.select("id", "reference"), "id")
        .select(
            "id",
            "seconds",
            normalized("transcript").alias("hypothesis"),
            normalized("reference").alias("truth"),
        )
        .select(
            "id",
            "seconds",
            word_errors("hypothesis", "truth").alias("errors"),
            F.size(F.split(F.trim("truth"), " ")).alias("words"),
        )
    )
    return scored.agg(
        F.count("*").alias("utterances"),
        F.round(F.sum("seconds") / 3600, 2).alias("hours"),
        F.sum("words").alias("reference_words"),
        F.sum("errors").alias("word_errors"),
        F.round(F.sum("errors") / F.sum("words") * 100, 2).alias("wer_percent"),
        F.lit(PUBLISHED_WER).alias("published_wer_percent"),
    )


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    result = run(spark)
    if "--json" in sys.argv:  # one machine-readable line, for automated runs
        import pysail

        print("LARGE " + json.dumps({**result.first().asDict(), "pysail": pysail.__version__}))
    else:
        result.show()
    spark.stop()
    server.stop()

How it works

• A table of paths. The clips are saved as FLAC files, and a DataFrame lists one path per file with its reference transcript. Each task opens only the files in its own rows.
• Load once per partition. mapInArrow downloads the model where the task runs (cached per machine), then loads it with faster-whisper before its loop over batches. Each model uses one CPU thread, because the engine already runs partitions in parallel.
• Transcribe. faster-whisper decodes each file to 16 kHz mono audio and transcribes it. Decoding is greedy at a single temperature of 0, with no sampled retries. Runtime and numerical differences can still affect transcripts.
• Save once. The transcripts are written to Parquet and read back, so the scoring query never runs the model again.
• Score in SQL. Both texts are lowercased and reduced to ASCII letters, digits, apostrophes and single spaces with explicit character classes, levenshtein counts the character edits between them, and the error rate is the total edits divided by the total reference characters.

Evaluate the full test set

multimodal/transcribe-audio/full_test_clean.py runs the same transcription function on all of LibriSpeech test-clean: 2,620 utterances, 5.4 hours of read speech. It normalizes both texts with Whisper's English text normalizer (from transformers, with its spelling list from the pinned openai/whisper-tiny.en repository), as the Whisper paper (Radford et al., 2022) did for its word error rates. It then computes the corpus word error rate in SQL: word edits per utterance from a pandas_udf, summed over all utterances and divided by the reference word count.

uv run multimodal/transcribe-audio/full_test_clean.py

The first run downloads the test set (a 350 MB Parquet file) and the normalizer's spelling list. The word error rate comes out at 5.9%. The Whisper paper reports 5.6% for tiny.en on test-clean with greedy decoding (Appendix D.1.1, Table 8). This run uses faster-whisper's CTranslate2 version of the model, not OpenAI's own code.

Use your own audio

Replace download_clips with a list of your own files. faster-whisper decodes audio with PyAV, which bundles the FFmpeg libraries (per the faster-whisper README), so it reads the formats that bundled FFmpeg supports. The paths must be readable from wherever the UDF runs: local paths on one machine, object storage on a cluster (open the file with a client such as fsspec and pass the file object to transcribe). Without reference transcripts, drop the scoring step. A file that cannot be decoded raises an error and fails the job. To record such files and keep going, catch the error per file and return a status column, as the OCR example does.

For other languages, use a multilingual model such as Systran/faster-whisper-small and set language. Whisper's own checkpoints whose names end in .en are English-only; the others are multilingual (per the Whisper model card).

Data

The clips are from the LibriSpeech ASR corpus (Vassil Panayotov, Guoguo Chen, Daniel Povey and Sanjeev Khudanpur, ICASSP 2015), licensed under CC BY 4.0, by way of the hf-internal-testing/librispeech_asr_dummy sample on Hugging Face. The full evaluation reads the test-clean split of the same corpus from the openslr/librispeech_asr dataset on Hugging Face, at a pinned revision.

Related Sail behavior

• Process media through a table of file paths
• Load a model once per partition

Keep exploring

AnalyticsTurn spoken requests into intent analytics →Turn spoken support requests into a report of customer needs. AI data preparationExtract text from scanned PDFs with OCR →Extract text from old scans and see where OCR makes mistakes.