Spark读取二进制文件调用Azure转写服务报错及等价代码问询

核心等价实现 #

PySpark中读取二进制文件后，提取的content列本身就是bytes类型，完全等价于本地file.read()返回的字节流。关键是要避免分布式场景下序列化/反序列化导致的字节流损坏问题。

场景1：单文件处理（Driver端直接操作） #

如果仅需处理单个音频文件，可直接提取content列内容：

# 读取单个MP3文件为Spark DataFrame
audio_df = spark.read.format("binaryFile").load("/mnt/datalake/path/to/target.mp3")

# 提取二进制内容（与本地file.read()结果一致）
audio_bytes = audio_df.select("content").first()[0]

# 直接传入Azure语音转写服务使用即可

场景2：批量文件处理（分布式UDF处理） #

针对批量音频文件，需通过Spark UDF在Executor端分布式处理，既避免Driver端内存瓶颈，也能保证字节流完整性：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import azure.cognitiveservices.speech as speechsdk

# 广播Azure语音服务配置（避免每个UDF实例重复初始化客户端）
speech_config_broadcast = spark.sparkContext.broadcast({
    "subscription_key": "你的Azure订阅密钥",
    "region": "你的服务区域"
})

def transcribe_audio(content):
    # Spark返回的content已经是bytes类型，直接使用
    audio_stream = speechsdk.audio.PushAudioInputStream()
    audio_config = speechsdk.audio.AudioConfig(stream=audio_stream)
    
    # 初始化语音识别器
    speech_config = speechsdk.SpeechConfig(
        subscription=speech_config_broadcast.value["subscription_key"],
        region=speech_config_broadcast.value["region"]
    )
    recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)
    
    # 写入音频字节流并执行识别
    audio_stream.write(content)
    audio_stream.close()
    result = recognizer.recognize_once()
    
    # 返回识别结果或错误信息
    return result.text if result.reason == speechsdk.ResultReason.RecognizedSpeech else result.error_details

# 注册UDF
transcribe_udf = udf(transcribe_audio, StringType())

# 读取批量音频文件并生成转写结果
batch_audio_df = spark.read.format("binaryFile").load("/mnt/datalake/audio_dir/*")
transcribed_df = batch_audio_df.withColumn("transcription", transcribe_udf("content"))

# 查看结果
transcribed_df.select("path", "transcription").show(truncate=False)

直接转Pandas报错的原因 #

Spark DataFrame的content列转为Pandas Series时，会经过Spark的序列化/反序列化流程，可能导致字节流的编码或结构损坏，不符合Azure语音服务对原始字节流的要求。直接在Spark分布式环境中处理（UDF或单文件提取）能保证字节流的原始完整性。

内容的提问来源于stack exchange，提问作者Enrique Benito Casado