处理MSRVTT数据集时多进程音视频提取报TypeError问题求助

错误信息 #

运行脚本时触发以下错误：

(valor) xxx:/VALOR/utils$ python extract_frame_and_wav_multiprocess.py
0%|                                       | 0/10005 [00:00<?, ?it/s]
Exception in thread Thread-3:
Traceback (most recent call last):
File "/anaconda3/envs/valor/lib/python3.9/threading.py", line 973, in _bootstrap_inner
self.run()
File "/anaconda3/envs/valor/lib/python3.9/threading.py", line 910, in run
self._target(*self._args, **self._kwargs)
File "/anaconda3/envs/valor/lib/python3.9/multiprocessing/pool.py", line 576, in _handle_results
task = get()
File "/anaconda3/envs/valor/lib/python3.9/multiprocessing/connection.py", line 256, in recv
return _ForkingPickler.loads(buf.getbuffer())
TypeError: __init__() missing 2 required positional arguments: 'stdout' and 'stderr'

错误指向multiprocessing/connection.py的这段代码：

def recv(self):
    """Receive a (picklable) object"""
    self._check_closed()
    self._check_readable()
    buf = self._recv_bytes()
    return _ForkingPickler.loads(buf.getbuffer())

执行脚本代码 #

使用的extract_frame_and_wav_multiprocess.py脚本：

import os
import os.path as P
import ffmpeg
import json
import tqdm
import numpy as np
import threading
import time
import multiprocessing
from multiprocessing import Pool
import subprocess


### change diffenrent datasets
input_path = '../datasets/msrvtt/raw_videos'
output_path = '../datasets/msrvtt/testt'
data_list = os.listdir(input_path)

def execCmd(cmd):
    r = os.popen(cmd)
    text = r.read()
    r.close()
    return text

def pipline(video_path, video_probe, output_dir, fps, sr, duration_target):
    video_name = os.path.basename(video_path)
    audio_name = video_name.replace(".mp4", ".wav")
    video_name = video_name.replace(".mp4", "")


    #extract video frames fps
    fps_frame_dir = P.join(output_dir, f"frames_fps{fps}", video_name)
    os.makedirs(fps_frame_dir, exist_ok=True)
    cmd = "ffmpeg -loglevel error -i {} -vsync 0 -f image2 -vf fps=fps={:.02f} -qscale:v 2 {}/frame_%04d.jpg".format(
              video_path, fps, fps_frame_dir)
    subprocess.call(cmd, shell=True)

    # Extract Audio
    sr_audio_dir = P.join(output_dir,f"audio_{sr}hz")
    os.makedirs(sr_audio_dir, exist_ok=True)
    audio_file_path = P.join(sr_audio_dir, audio_name)
    cmd = "ffmpeg -i {} -loglevel error -f wav -vn -ac 1 -ab 16k -ar {} -y {}".format(
            video_path, sr, audio_file_path)
    subprocess.call(cmd, shell=True)


def extract_thread(video_id):
    
    video_name = os.path.join(input_path, video_id)
    if not os.path.exists(video_name):
        return
    probe = ffmpeg.probe(video_name)
    pipline(video_name, probe, output_path, fps=4, sr=22050, duration_target=10)


def extract_all(video_ids, thread_num, start):
    length = len(video_ids)
    print(length)
    with Pool(thread_num) as p:
        list(tqdm.tqdm(p.imap(extract_thread, video_ids), total=length))

if __name__=='__main__':
    thread_num = 50
    start = 0

    print(len(data_list))
    extract_all(data_list, thread_num, start)

问题原因 #

你的怀疑方向正确：DiDeMo无音频，ffmpeg.probe()返回的对象结构简单，可被Pickle序列化；而MSRVTT包含完整音频流，probe返回的带音频信息的复杂对象无法被multiprocessing的_ForkingPickler正确序列化/反序列化，跨进程传递时触发错误。

解决方法 #

1. 避免跨进程传递不可序列化对象 #

把ffmpeg.probe()的调用移到pipline函数内部，让每个子进程自行处理视频探测，不用传递复杂对象：

def pipline(video_path, output_dir, fps, sr, duration_target):
    # 子进程内部调用probe，避免跨进程传递复杂对象
    probe = ffmpeg.probe(video_path)
    video_name = os.path.basename(video_path)
    audio_name = video_name.replace(".mp4", ".wav")
    video_name = video_name.replace(".mp4", "")

    # 后续帧和音频提取代码保持不变...

# 修改extract_thread，不再传递probe参数
def extract_thread(video_id):
    video_name = os.path.join(input_path, video_id)
    if not os.path.exists(video_name):
        return
    pipline(video_name, output_path, fps=4, sr=22050, duration_target=10)

2. 降低进程数 #

当前设置的50个进程可能超出系统承载，加剧资源竞争和序列化问题，建议先调整为10-20个进程测试：

if __name__=='__main__':
    thread_num = 15  # 从10开始逐步调整
    start = 0

    print(len(data_list))
    extract_all(data_list, thread_num, start)

3. 增加异常捕获处理 #

MSRVTT可能存在少量损坏的视频文件，增加异常捕获可以避免单个视频的问题导致整个进程池崩溃：

def extract_thread(video_id):
    video_name = os.path.join(input_path, video_id)
    if not os.path.exists(video_name):
        return
    try:
        pipline(video_name, output_path, fps=4, sr=22050, duration_target=10)
    except Exception as e:
        print(f"处理视频 {video_id} 失败: {str(e)}")
        return

4. 替换subprocess.call为subprocess.run（可选） #

subprocess.call已被官方推荐用subprocess.run替代，能更好地控制命令执行和输出：

# 替换帧提取的subprocess调用
subprocess.run(cmd, shell=True, check=False, capture_output=True)
# 替换音频提取的subprocess调用同理

内容的提问来源于stack exchange，提问作者mike.w