做视频数据处理的人，大概都经历过那个"看起来简单、跑起来想死"的阶段：需求就是一句"把视频逐帧处理一遍"，结果代码写出来，一跑就是几小时，中途内存飙红，进程被系统杀掉，再重启又得从头再来。我自己被这种问题折磨过很多次之后，写了 hyperframes 这个小项目。它不发明任何新算法，只做一件事：把"逐帧读取—批量化—并行处理—断点续传"这整条链路整理成一个可靠、高效的框架。这篇博文就是 hyperframes 的完整复盘——为什么这样设计、核心代码怎么写的、实际跑起来性能如何、中间踩了哪些坑。如果你正在做视频抽帧、数据集清洗、训练前预处理这类工作，这篇文章应该能帮你省下不少时间。
1. hyperframes 从哪来：项目缘起与边界
1.1 逐帧处理的三个致命痛点
先说最直接的痛点。绝大多数人第一次写视频处理代码，都是这样的：打开 OpenCV 的VideoCapture，然后一个while True循环，cap.read()一帧处理一帧。这套写法在小视频上完全没问题，可一旦数据量上来，它有三个致命的毛病。
第一个是完全没有断点能力。一个 1080p 30fps 的视频，一小时就有 108000 帧。你跑了 40 分钟，进度条走到 60%，突然因为内存不足被 OOM Killer 干掉，那么前面 40 分钟的所有计算结果全部作废，只能从头再跑。但凡你处理的是几千个视频，这种"从头再来"的代价就是灾难级的。
第二个是串行执行浪费硬件。逐帧循环天生是单线程的，你的 CPU 可能有很多核，但在这种写法下，一个核在拼命干活，其余核全部围观。即使你想加多线程，也会发现 Python 的 GIL 像一道锁一样卡在那里，实际加速比远低于预期。
第三个是帧级调用的固定开销太高。Python 里每调一次函数都有几十到几百纳秒的固定开销，听起来不多，但乘以 10 万帧、100 万帧，就变成了几秒到几十秒的纯调用开销。而这还只是函数调用本身，还没算上每一帧都进行数组创建、类型转换、上下文切换的成本。把 100 万个单帧操作合并成 6 万个批次操作，省下的就是这中间的重复消耗。
hyperframes 的核心动机，就是系统性地解决这三个问题：问题一靠断点续传解决，问题二靠多进程并行解决，问题三靠批处理解决。
1.2 hyperframes 做什么，不做什么
在设计这个项目之前，我先明确了两条边界：它要解决什么问题，以及它坚决不碰什么问题。
hyperframes 要做的是这四个能力：
- 一个统一的帧来源抽象，支持指定起始帧和结束帧，支持中途中断后从断点恢复；
- 一个批处理迭代器，把逐帧读取变成批量产出，从根上减少 Python 层的调用次数；
- 一个带背压控制的并行执行管道，用多进程消费批次任务，同时严格限制积压在内存里的任务数量；
- 一套轻量的进度追踪机制，把处理进度、速度、已用时间记录成结构化数据，方便随时恢复和监控。
它不做的事也很明确：不做视频剪辑，不做滤镜库，不做模型推理框架，更不做分布式计算。处理函数完全由使用者提供，hyperframes 只负责把帧安全高效地送到你的函数手里。这样设计的好处是边界清晰，核心代码量可以控制在很小的范围，出问题时也容易排查——这个问题在后面的排错部分会体现得很明显。
2. 整体架构与关键选型思路
2.1 解码层：OpenCV、PyAV 还是裸调 ffmpeg
第一层选型就卡住了我一阵子：视频解码到底用什么？
| 方案 | 上手难度 | 随机跳帧 | 硬件解码 | 控制粒度 | 适用场景 |
|---|---|---|---|---|---|
| OpenCV VideoCapture | 最低 | 粗糙，seek 慢 | 依赖编译选项 | 弱 | 快速demo、通用处理 |
| PyAV | 中等 | 精确 | 支持 NVDEC/QSV | 强 | 帧精确定位、硬件加速 |
| ffmpeg 子进程 | 中等 | 不适用，只适合流式 | 完全可控 | 强 | 大规模转码、GPU批处理 |
最终 hyperframes 的默认解码层选的是 OpenCV，但留出了替换接口。原因很简单：OpenCV 的read()在 4.x 版本里解码阶段会释放 GIL，这意味着在某些场景下配合多线程还能获得一部分加速；而且它对新手最友好，一个VideoCapture对象就能搞定绝大多数视频格式。
但 OpenCV 有明显的短板：CAP_PROP_POS_FRAMES跳帧在很多编码格式下效率极低。H.264 的帧间压缩决定了你不可能凭空跳到第 5000 帧——解码器必须从最近的关键帧逐帧解码到目标位置。所以 opencv 的 seek 在遇到长 GOP 视频时会表现为"跳一下卡半天"。如果你的场景是需要精确跳到任意帧做分析——比如抽帧审核——那应该换成 PyAV，用seek()配合时间戳来做精确跳转。这部分我在第 5 节会专门展开。
这里也要给一个非常实用的建议：如果能接受流式处理，直接在命令行调ffmpeg二进制通过管道输出原始像素数据，是吞吐量最高的一条路。代价是你失去了按帧号随机访问的能力，只能顺序流式消费。hyperframes 没有默认走这条路，但在第 6 节我会提一下怎么把它加进来。
2.2 为什么 hyperframes 坚持批处理
很多人在做帧处理时有一个习惯性误区：觉得"反正处理函数是一帧一帧写的，那循环也应该一帧一帧来"。这在 Python 里恰好是性能杀手。
批处理的好处其实是三层叠加的。第一层是函数调用开销的摊薄。假设你要处理 100 万帧，单帧调用意味着你要做 100 万次函数调用、100 万次数组边界检查、100 万次 Python 层面的循环控制。如果每批处理 16 帧，调用次数直接降到 6 万多次，单纯这部分的耗时就能减少一个数量级。
第二层是数据类型转换的开销。视频帧从解码器出来通常是uint8的 H×W×C 像素数组，而绝大多数机器学习预处理需要浮点、归一化、指定尺寸。如果你逐帧做astype(np.float32)再resize，这些操作每次都伴随单帧的内存分配和释放。合并成批次后，你可以一次性为整个批次分配连续内存，CPU 的缓存命中率会明显改善。
第三层是为后续的 GPU 计算做准备。深度学习推理几乎都是张量批量操作，模型天然喜欢吃(B, H, W, C)这样的输入。你逐帧处理完再拼成 batch，跟直接在管道里产出 batch，性能差别非常大。hyperframes 的BatchIterator就是负责在源头就把帧聚合成批次，从数据结构上保证下游计算友好。
但是批处理也有代价，最大的代价就是内存。一帧 1080p RGB 图像是 1920×1080×3 个字节，约 6.2MB，16 帧一批就是约 99MB。如果你不控制批大小和积压任务数，内存会迅速失控。这个问题在第 3 节和第 5 节都有详细的处理和说明。
2.3 并行模型：线程池、进程池与有界任务流
Python 的并发模型选择，很多人第一反应是ThreadPoolExecutor。这在线程能释放 GIL 的场景下确实有效——比如 OpenCV 的read()在解码阶段会释放 GIL，所以单纯做"多线程读取+主线程处理"能获得一定的加速。但问题在于，如果你的处理函数是纯 Python 逻辑或者重度依赖 NumPy 的计算，GIL 会立刻成为瓶颈，多线程甚至可能比单线程更慢（因为线程切换开销）。
所以 hyperframes 选的是ProcessPoolExecutor——多进程才是真正绕开 GIL 的方式。但多进程有一个绕不开的代价：数据的序列化和反序列化。你要把每一批帧（一堆 NumPy 数组）从主进程 pickle 之后发送给子进程，计算完结果再 pickle 回来。这个过程如果处理不好，性能会直接崩溃。
对这个问题的解法是"以批为单位传输"。单帧 6MB，序列化 1 帧发给子进程，开销占比极高；而一次发送一个 100MB 的批次，每批 16 帧平摊下来的单帧序列化成本就低得多。这也是为什么 hyperframes 坚持先批量、再并行的架构顺序——批处理是并行的前提，反过来并行又要求批量传输。
另一个非常容易被忽视的设计点是有界任务流。很多人用ProcessPoolExecutor时会写成这样：
executor.map(process_fn, all_batches)这行代码会把所有批次任务一次性全塞进任务队列。如果视频有 6000 个批次，主进程就会先把这批任务的参数全部序列化完、塞进管道，内存瞬间被撑爆。正确做法是维护一个滑动窗口，只让有限数量的任务在飞行中，完成一个再提交一个。这段代码我放在了第 3.3 节，它是全项目里对内存影响最大的一段。
2.4 帧数据的统一布局
最后一个选型问题是帧数据在内存里怎么摆放。最自然的方式是把一个批次堆叠成形状为(B, H, W, C)的连续 NumPy 数组。这个布局对后续处理极度友好：可以直接整体做astype、整体resize、直接交给 PyTorch 或 TensorFlow 的张量构造。
但这里有一个很多人忽略的陷阱：np.stack(batch)会执行一次内存拷贝。16 帧堆叠成(16, 1920, 1080, 3)，意味着一次约 100MB 的 memcpy。这不是问题，问题在于很多人会在每个批次都做 stack，又把 stack 之后的数组保存到列表里，最后列表越来越大——内存就是这么被拖死的。
hyperframes 的约定很明确：批处理管道的输出是"一次性"的，消费完毕就要释放引用，绝不把所有批次累积在内存里。如果你确实需要把所有结果整理到文件里，也应该流式写入，而不是攒成一个巨型数组再写。
3. 核心模块实现拆解
3.1 FrameReader：可跳转、可续传的帧来源
FrameReader是整个框架的入口，它的核心职责是：统一"从哪里读取、读到哪一帧、现在读到哪一帧"这三个状态。我先把最基础的 OpenCV 版本写出来：
import cv2 import numpy as np from pathlib import Path from typing import Optional, Tuple class FrameReader: """基于 OpenCV 的帧读取器，支持起始帧、结束帧和位置查询。""" def __init__( self, video_path: str, start_frame: int = 0, end_frame: Optional[int] = None, ): self.video_path = str(video_path) self.start_frame = start_frame self.end_frame = end_frame self._cap = None self._pos = 0 def open(self) -> None: if self._cap is not None: return self._cap = cv2.VideoCapture(self.video_path) if not self._cap.isOpened(): raise IOError(f"无法打开视频: {self.video_path}") if self.start_frame > 0: self._cap.set(cv2.CAP_PROP_POS_FRAMES, self.start_frame) self._pos = self.start_frame def read(self) -> Tuple[bool, Optional[np.ndarray]]: if self.end_frame is not None and self._pos >= self.end_frame: return False, None ok, frame = self._cap.read() if ok: self._pos += 1 return ok, frame @property def position(self) -> int: return self._pos def close(self) -> None: if self._cap is not None: self._cap.release() self._cap = None别看这段代码短，几个细节是反复测试后确定的。一是end_frame的判断必须在read()之前做，否则你会多读一帧且无法回退。二是position是自增计数，而不是用cap.get(CAP_PROP_POS_FRAMES)——因为某些容器的帧号元数据并不可靠，自己数才是最诚实的。三是close()里把_cap置为 None，这是为了支持 reopen 后的断点续传场景，后面会讲到。
如果你发现 OpenCV 跳帧太慢，可以替换成 PyAV 版本。核心思路一样，只是seek的逻辑要换成基于时间戳。我不会在这里贴完整代码，但给你一个方向：PyAV 的container.seek(offset=timestamp, backward=True)配合stream.frames的 dts/pts 换算，能做到亚帧级精确定位。
3.2 BatchIterator：按块吐数据的内存闸门
有了FrameReader，下一步是把它变成"按批吐帧"的迭代器。这里的关键设计是：迭代器内部维护一个小缓冲区，读满batch_size帧才吐出来，绝不一次读太多。
class FrameBatchIterator: """将 FrameReader 的逐帧读取聚合成批次产出。""" def __init__(self, reader: FrameReader, batch_size: int = 16): self.reader = reader self.batch_size = batch_size def __iter__(self): self.reader.open() batch = [] while True: ok, frame = self.reader.read() if not ok: if batch: yield np.stack(batch, axis=0) break batch.append(frame) if len(batch) == self.batch_size: yield np.stack(batch, axis=0) batch = [] self.reader.close()这里有一个性能细节：np.stack(batch, axis=0)得到的是(B, H, W, C)的连续数组，但batch这个列表里的单帧数组依然被引用着，直到列表被清空。如果处理函数持有返回的批次对象不放，而batch列表又继续累积下一批的帧，内存就会悄悄上涨。所以我在 yield 之后马上batch = []重置列表，并且要求调用方不要长期持有批次对象。这个约定写进了项目文档的第一条。
批大小怎么选？我建议从 8 到 32 之间试。太小的批（比如 4）没有明显摊薄效果；太大的批（比如 128）会让单批内存轻松超过 800MB，批处理过程中如果下游函数又复制一次数据，就有 OOM 风险了。1080p 视频我个人常用 16，4K 视频建议降到 4 到 8。
3.3 并行管道：控制并发与内存的关键代码
批处理迭代器解决了调用次数问题，但还没有解决"多核吃满"的问题。真正让 hyperframes 跑得快的，是这段有界并行管道的代码：
import concurrent.futures as cf import multiprocessing as mp import os from typing import Callable, Iterator class HyperPipeline: """有界并行的批处理管道。 max_inflight 控制积压任务的上限，防止大量批次同时 排队在主进程的内存里。 """ def __init__( self, batch_iter: Iterator, workers: int = None, max_inflight: int = 4, ): self.batch_iter = batch_iter self.workers = workers or max(1, mp.cpu_count() - 1) self.max_inflight = max_inflight def run(self, fn: Callable): with cf.ProcessPoolExecutor(max_workers=self.workers) as pool: futures = [] for batch_id, batch in enumerate(self.batch_iter): # 控制飞行中的任务数量 if len(futures) >= self.max_inflight: done, _ = cf.wait(futures, return_when=cf.FIRST_COMPLETED) for f in done: yield f.result() futures = [f for f in futures if f not in done] futures.append(pool.submit(fn, batch_id, batch)) for f in cf.as_completed(futures): yield f.result()max_inflight是这里最核心的参数。它限制的是"主进程里已经提交给子进程、但还没返回结果"的任务数量。为什么必须限制？因为每提交一个任务，就需要把批次数据序列化并写入进程间管道，如果不受控地全部提交，内存会被未消费的任务队列占满。
max_inflight=4是我测试后的经验值：在workers=8的情况下，4 个在飞行中的任务已经足够让所有子进程保持忙碌，同时把内存峰值控制在可接受范围内。如果你把max_inflight调到和workers一样甚至更大，理论上吞吐量不会有明显提升，但内存峰值会显著增长。这个参数值得你为所用机器单独调一调。
还有一个细节：cf.wait(futures, return_when=FIRST_COMPLETED)会等到至少一个任务完成，返回的done集合里可能不止一个任务。我取走所有已完成的 future 后，再把剩余任务重新赋值给futures。这样做的好处是列表不会无限增长，每个批次的结果都能尽早被 yield 出去，比如写文件、打印进度。
3.4 断点续传与进度追踪
断点续传的实现思路不复杂，但做得好不好对体验影响很大。我的方案是：处理函数在处理完一批后，返回一个"这一批的最后帧号"，主进程把这个帧号写进一个 JSON 状态文件。下次启动时，如果检测到状态文件存在，就让FrameReader从这个帧号重新开始。
import json import os from pathlib import Path class Checkpoint: def __init__(self, path: str): self.path = Path(path) self.data = {} def load(self): if self.path.exists(): with open(self.path, "r", encoding="utf-8") as f: self.data = json.load(f) def save(self, video_name: str, frame_pos: int): self.data[video_name] = frame_pos with open(self.path, "w", encoding="utf-8") as f: json.dump(self.data, f, ensure_ascii=False, indent=2)不要小看这个Checkpoint。在真实的数据处理项目里，一个视频跑两小时，跑到 90% 崩掉，如果没有断点能力，你损失的不只是一晚上的时间，还有可能是整个任务排期的崩溃。有了断点，重启脚本后自动跳过已完成的部分，这种体验上的提升是质变的。
4. 完整可运行的示例与实测数据
4.1 项目结构和依赖清单
hyperframes 的项目结构非常克制，核心代码就三个文件：
hyperframes/ ├── hyperframes/ │ ├── __init__.py │ ├── reader.py # FrameReader 与 FrameBatchIterator │ ├── pipeline.py # HyperPipeline 与 Checkpoint │ └── utils.py # 进度统计、时间格式化 ├── examples/ │ └── frame_stats.py # 示例：统计视频帧的亮度均值 └── requirements.txt依赖只有四个：opencv-python、numpy、tqdm（可选，进度条）、av（可选，PyAV 解码）。requirements.txt 写成：
opencv-python>=4.5 numpy>=1.21 tqdm>=4.64 av>=10.0安装就是一行命令。这个项目的定位决定了它不应该引入重量级依赖，任何用不到的功能都应该通过插件机制让使用者自己装。
4.2 核心代码逐段讲解
示例任务是计算视频每一帧的亮度均值，然后把结果逐批保存成 CSV。这个任务虽然简单，但完全覆盖了"读取—批量—并行—产出"的完整链路。
import csv import sys from hyperframes.reader import FrameReader, FrameBatchIterator from hyperframes.pipeline import HyperPipeline, Checkpoint def process_batch(batch_id: int, batch): """子进程里执行的处理函数：计算每帧亮度均值。""" # batch 形状: (B, H, W, C)，uint8 gray = batch.mean(axis=3, dtype=np.float64) # 转灰度 brightness = gray.mean(axis=(1, 2)) # 每帧均值 return batch_id, brightness def main(video_path: str): reader = FrameReader(video_path) batch_iter = FrameBatchIterator(reader, batch_size=16) pipeline = HyperPipeline(batch_iter, workers=8, max_inflight=4) # 检查断点 ckpt = Checkpoint("checkpoint.json") ckpt.load() last_pos = ckpt.data.get(video_path, 0) if last_pos: reader = FrameReader(video_path, start_frame=last_pos) batch_iter = FrameBatchIterator(reader, batch_size=16) pipeline = HyperPipeline(batch_iter, workers=8, max_inflight=4) with open("output.csv", "a", newline="") as f: writer = csv.writer(f) for batch_id, brightness in pipeline.run(process_batch): for i, v in enumerate(brightness): frame_no = batch_id * 16 + i writer.writerow([frame_no, round(v, 4)]) # 每处理 100 批落一次检查点 if batch_id % 100 == 0: ckpt.save(video_path, (batch_id + 1) * 16) if __name__ == "__main__": main(sys.argv[1])注意几个关键点。处理函数process_batch必须设计成无副作用的纯函数——它只接收批次数据，只返回结果。所有文件写入、状态保存都必须放在主进程里做。这样设计是为了避免多个子进程同时写同一个文件造成数据损坏，也方便断点续传时整个状态保持一致。
csv.writer的逐行写入在这里并没有成为瓶颈，因为 Python 的文件写是带缓冲的。如果你要写的数据量特别大，建议换成savetxt按批写入，或者直接写二进制文件。
4.3 实测：三种模式的性能对比
我在一台 8 核 CPU 的机器上，用一段 5 分钟、1080p、30fps 的 H.264 视频做了三组测试。视频总共 9000 帧，任务是"计算亮度均值 + 写 CSV"。结果如下：
| 处理模式 | 总耗时 | 吞吐量 | 峰值内存 |
|---|---|---|---|
| 逐帧串行（while read） | 约 220 秒 | 约 41 帧/秒 | 约 120MB |
| 批量串行（无并行） | 约 130 秒 | 约 69 帧/秒 | 约 200MB |
| 批量并行（8进程，inflight=4） | 约 45 秒 | 约 200 帧/秒 | 约 680MB |
第一组到第二组的差距，完全是批处理摊薄调用开销的效果；第二组到第三组的差距，则是多进程真正吃满多核的结果。峰值内存 680MB 是 8 个 worker 每个持有若干批次数据的总和，可接受但也到了需要注意的程度——如果视频是 4K，批大小就得往下调。
这个实测结果也印证了一个规律：瓶颈在哪一步，优化就要针对哪一步。如果处理函数很轻（比如只是亮度均值），那解码和传输就是瓶颈；如果处理函数很重（比如跑一个神经网络），那计算就是瓶颈，并行度设计又要重新考虑。所以 hyperframes 的性能参数（batch_size、workers、max_inflight）没有写死，而是全部暴露成可配置项。
5. 常见问题与排查实录
5.1 解码速度远低于预期
这是被问得最多的一个问题。现象是：明明处理函数很简单，但整个管道跑得很慢，CPU 占用也不高。排查方向通常是三个。
第一个是软件解码拖后腿。OpenCV 默认走 CPU 软解，H.264 1080p 的软解吞吐量大约在 200-400 帧/秒，遇到高码率或者 4K 视频会急剧下降。如果你手上是支持 NVDEC 的 NVIDIA 显卡，或者 Intel 核显支持 QSV，应该换 PyAV 并启用硬件解码。PyAV 的硬件解码可以把解码吞吐量提升 2 到 5 倍，代价是编码实现复杂度上升。我强烈建议把"解码器"作为可插拔组件，而不是把 OpenCV 写死在所有代码路径里。
第二个是不必要的数据转换。很多人循环里会做cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)，如果你后续用的是深度学习模型，这个转换是需要的；但如果只是算统计量，OpenCV 默认的 BGR 布局完全可以直接计算。每次多余的 cvtColor 都是在为一整帧的像素做遍历，成本极高，能省则省。
第三个是频繁的内存分配。在批量管道里，一个常见的隐性开销是每一批都np.stack新建数组、astype新建数组。如果这些操作无法避免，至少可以复用输出缓冲区，而不是每次都从操作系统重新申请。用np.empty预分配一个全局批次缓冲区，可以显著降低分配次数。
5.2 帧数对不上与丢帧
视频的帧数统计是个坑。cv2.CAP_PROP_FRAME_COUNT返回的"总帧数"在很多容器格式下是估算值，尤其对于可变帧率（VFR）视频，元数据里的帧数和实际解码出来的帧数经常不一致。
如果你发现视频明明标注 9000 帧，但read()循环里却读出了 9023 帧或者 8978 帧，不要惊讶，这很正常。处理方法也很简单：永远以read()的成功次数为准，不要拿元数据帧数去做循环终止条件。hyperframes 的FrameReader设计了end_frame这个可选参数，就是允许你在确实需要精确控制读取范围时手动指定，而默认情况下它读到 EOF 为止。
还有一个隐蔽的问题是"丢帧"——解码器在某些损坏帧处读取失败，read()返回(False, None)。如果你在这个位置直接 break，就会丢掉后面所有帧。正确处理是：连续读取失败达到一定阈值（比如 3 次）才认为视频异常终止，否则尝试继续读。这个逻辑在完整版的FrameReader里有实现。
5.3 内存爆炸的三种死法
内存问题在并行管道里特别常见，我总结了三种典型死法。
死法一：无界提交。就是前面说的，用pool.map一次性提交所有批次，海量数据积压在任务队列里。解法是换成有界任务流，max_inflight设成 worker 数的 0.5 到 1 倍。
死法二：结果累积。使用pipeline.run()时，如果你把 yield 出来的结果都append到一个大列表里，内存同样会涨。尤其是处理结果本身比较大时（比如特征向量），1 万批的结果轻松超过几个 GB。正确的做法是流式消费：写完文件立刻丢弃，不要回头保留。
死法三：引用泄漏。某些 OpenCV 版本在VideoCapture释放后并不立即归还所有内存，或者内存碎片化导致峰值偏高。解决方法是批量重用容器，并且对大数组显式del后调用gc.collect()。注意，gc.collect()不要放在循环里频繁调用，那会拖慢速度，只在批与批之间偶尔调用是合理的。
5.4 几个容易被人忽略的暗坑
经验告诉我们，真正的坑往往不在主路径上，而在角落。
第一个是Windows 下多进程的__main__问题。ProcessPoolExecutor在 Windows 上默认使用spawn方式创建子进程，你的脚本入口必须是if __name__ == "__main__":保护的，否则每个子进程都会重新执行整个模块，轻则报错，重则无限递归创建进程。在 Linux 和 macOS 上默认是fork，没这个问题，但为了跨平台，我一直坚持写入口保护。
第二个是在父进程打开了VideoCapture后再 fork 子进程。Linux 的 fork 会继承父进程的文件描述符，如果父进程先打开了视频文件再创建进程池，子进程可能会持有同一个文件描述符，导致读取错位或者文件锁冲突。正确做法是在创建ProcessPoolExecutor之前不要打开捕获器，FrameReader.open()尽量推迟到迭代开始时才执行。
第三个是进度条的位置。tqdm这种进度条应该只在主进程里输出，绝不能写进子进程的处理函数里，否则你会看到终端被乱七八糟的进度条刷屏。正确的做法是让子进程返回batch_id，主进程根据返回的批次号更新进度条。
第四个坑是np.stack与批次传输的序列化开销。我之前提到过以批为单位传输，但即使做了批处理，如果你是 4K 视频，16 帧一批就是近 400MB 的序列化数据，这个开销依然巨大。这时应该把批大小降到 4，并且评估传输成本是否已经超过并行收益。如果传输成为新瓶颈，就得考虑共享内存方案——用multiprocessing.shared_memory把帧缓冲区做成可读共享，子进程只接收元数据，这是下一步的优化方向。
6. 扩展方向与个人体会
6.1 还能往上叠加的能力
hyperframes 目前是个基础框架，但它预留了几个明显的扩展方向。
第一个是接入硬件解码链路。把默认的 OpenCVFrameReader替换成 PyAV 加 NVDEC/QSV 的实现，可以让解码吞吐量翻倍。这个替换不涉及上层 API 的任何改动，因为FrameReader已经把接口固定住了。
第二个是共享内存传输。当前ProcessPoolExecutor的任务参数是 pickle 传输的，序列化大数组有成本和内存峰值。如果用multiprocessing.shared_memory或者mp.Array做零拷贝传输，子进程直接读取共享缓冲区，吞吐量会有质的提升。代价是实现复杂度上了一个台阶，如果不是超大视频集，建议先不做。
第三个是场景感知的批处理策略。目前是固定批大小，其实可以做成自适应：处理函数是轻量任务时自动调大批大小，重度任务时自动调小。这个可以通过让处理函数返回一个"预估耗时"的元数据来实现，但优先级不高。
第四个是结果持久化的插件化。目前示例里用 CSV，实际项目中可能是 Parquet、HDF5、SQLite 或者直接写 TFRecord。把这些抽象成Sink接口，就能让框架在不同项目中复用。
6.2 做这个项目我学到的东西
最后分享几点个人体会。
第一，批处理不是银弹，但几乎总是值得先试。很多"性能问题"其实是调用模式问题，不是算法问题。你先把逐帧改成逐批跑一遍，大量看起来复杂的优化手段可能根本不需要。
第二，并行代码的难点是内存，不是速度。写ProcessPoolExecutor十分钟就够了，真正的功夫在"如何控制积压任务、如何管理结果生命周期、如何在崩溃后恢复"。这些不起眼的部分，决定了系统能不能在大数据集上稳定跑完。
第三，工具的边界很重要。我见过太多项目试图把视频处理框架做成万金油，结果核心逻辑被各种业务耦合得面目全非。hyperframes 从一开始就坚持"只送帧，不做业务"，这让它的代码量维持在几百行，每个文件都能被新人一眼看懂。
如果你也在做类似的事，我的建议很简单：先把 frame source 抽象出来，再上批处理，最后再加并行和断点。这个顺序如果颠倒，后期返工的代价会很大。hyperframes 本身不算复杂，但它把前述几个关键设计串成了一个可以复用的小框架，希望这篇复盘对你也有参考价值。
