先说个我自己的经历早几年接到过一批“把平台存量视频全部跑一遍内容标签”的需求当时我的第一反应是“把视频路径发给 Spark像处理普通日志一样搞个 map 就行”。结果任务一跑就连续踩坑从输入格式到执行器内存全都不对劲最后把链路整体重做才稳定下来。如果你也在规划类似的项目——不管是监控视频、短视频还是长视频内容有一个结论可以先放在前面用 Spark 处理视频数据核心难点从来不在“视频”本身而在“怎么把视频拆成适合分布式计算的样子”。视频文件是出了名的二进制大块头直接扔给 Spark 去读并非不能跑但代价极高。真正的做法是让 Spark 负责调度、切帧、元数据提取、推理任务的派发与结果聚合让每个执行器只处理“一个视频的某一段”或者“一批帧的路径”。这篇文章我会按这个思路把整个链路的架构设计、切帧策略、元数据建模、推理协作、流式落地的实操细节都拆开讲希望能给你省掉我当年踩过的那些坑。1. 视频和大数据的真实关系先想清楚再动手1.1 Spark 在视频链路里到底扮演什么角色视频数据有一个和大数据常见场景很不一样的特点它的“数据量”不只是文件体积大还在于单条有效信息的提取成本高。一份业务日志读进来就能查一段视频你得先解封装、解码、抽帧可能还要调用模型推理才能得到“这帧画面里有行人”这种业务字段。所以在设计系统前必须把角色分清Spark 的角色执行分布式任务调度、元数据管理、帧路径分发、推理结果聚合、统计分析其它组件的角色FFmpeg 负责解码切帧推理服务负责模型计算对象存储或 HDFS 负责帧文件持久化不该做的事让 Spark 执行器在一个进程里直接解码视频流或者把超大二进制文件作为 DataFrame 的一行传来传去。我后来在项目里把这条链路叫“切、存、算分离”。切是切帧存是帧文件和元数据算是标签和统计。三者各司其职Spark 才能发挥它真正擅长的并行计算能力。1.2 不建议用 Spark 直接啃视频文件的三个原因网上能看到不少“用 Spark 读视频”的 Demo大多是先读二进制再调用解码库。真放到生产环境至少有三个问题一是输入分片和视频解码不对齐。Spark 读取文件时会按照块大小切分输入而视频文件的分片点是随机的。一个分片可能落在关键帧之前或之后FFmpeg 在解码时需要依赖前面的帧信息导致分片边界上大量重试和重复解码严重拖慢处理速度。二是数据本地性失效。视频文件通常在对象存储或 HDFS 上连续存放分片后各执行器拿到的片段是分散的。Spark 会把任务调度到离数据近的节点但解码器需要在本地临时生成完整的解码上下文本地性优势被削弱还会占用大量临时磁盘。三是容错粒度太粗。如果某个执行器在处理一个大视频文件时崩溃整个任务从该文件的起始位置重跑中间已经解码完的帧全部作废。视频文件越大这种“功亏一篑”的成本越高。所以标准做法是先切帧再让 Spark 分析帧或帧元数据。切帧这一步虽然也消耗算力但它可以做到“按文件粒度并行、每个文件独立失败重试”天然适合分散到批量任务里去做。2. 海量视频的分布式切帧两种可行路线2.1 先切帧再读最稳的批处理组合切帧不是只有一种姿势我实际用过两种先说最稳的一种。把视频文件路径做成分区输入每个 Spark 任务处理一个或多个完整视频文件任务内部调用 FFmpeg 子进程完成解码抽帧抽出来的帧直接写到对象存储或 HDFS同时在 DataFrame 里记录路径、时间戳、视频 ID 等元数据。简单说Spark 不直接解视频而是管理“解视频的进程”。这里的 Spark 代码并不复杂核心是让每个执行器通过 Java/Scala 的 ProcessBuilder 启动 FFmpeg并传递参数。需要注意的几点每个执行器分配的 CPU 要和 FFmpeg 子进程的线程数匹配避免超售FFmpeg 输出要重定向到日志系统不然子进程输出积压会阻塞执行抽帧周期要根据业务需求设置比如每 5 秒一帧、每 10 秒一帧或者按场景切换抽帧不要无脑抽全部帧。我当时在一批监控视频上用过一个折中策略关键场景画面变化较大密集抽帧静止画面间隔抽帧。这样既控制帧总量又保证下游推理覆盖到有效内容。2.2 帧在存储上的布局直接决定后续效率切完帧只是第一步帧文件怎么落盘决定了之后 Spark 读起来顺不顺畅。经验是不要让每个视频把所有帧摊开成一堆小 JPEG 文件。比如一个 30 分钟的 1080p 视频按每秒一帧抽就是 1800 张图几万个视频就是几千万张小文件。小文件既占用文件系统的元数据内存又拖慢后续推理任务的读取速度。我自己常用的布局有两种场景推荐存储布局原因帧要长期保存、供多轮模型使用按“视频 ID 时间段”分桶打包成 Parquet帧数据以压缩字节数组存储文件数可控Spark 读取时可批量解码帧只用于一次推理过期可删抽出帧后直接转成 TFRecord 或自定义二进制格式推理完即可清理减少对象存储压力避免小文件堆积如果你仍然想保留 JPEG 原图也至少按视频或按天分目录避免把几百万张图堆在同一个前缀下。很多对象存储对单前缀下的文件数量有限制这一点等文件多起来再改会非常痛苦。3. 把视频元数据变成可查询的数据资产Spark SQL 建模3.1 元数据表怎么设计才不会后悔切帧和推理之后真正落到数据仓库里的核心资产是元数据。一个好的元数据模型能让后续所有分析工作变得简单反之则会让每条业务查询都变成一场灾难。我通常会建两张表一张是视频维度表一张是帧维度表。视频维度表存的是视频本身的属性CREATE TABLE dwd_video_meta ( video_id STRING COMMENT 视频唯一ID, platform STRING COMMENT 来源端, bucket STRING COMMENT 对象存储桶, path STRING COMMENT 视频文件路径, duration_sec DOUBLE COMMENT 视频时长, width INT, height INT, fps DOUBLE, codec STRING, upload_time TIMESTAMP COMMENT 上传时间, etl_time TIMESTAMP COMMENT 入库时间 ) PARTITIONED BY (dt STRING) STORED AS ORC;帧维度表存的是每一帧或者每一段时间窗口的分析结果CREATE TABLE dwd_frame_meta ( video_id STRING, frame_idx BIGINT, pts_time DOUBLE COMMENT 帧时间戳, frame_path STRING COMMENT 帧文件存储路径, scene_label STRING, detect_result MAPSTRING, DOUBLE COMMENT 模型输出比如目标框和置信度 ) PARTITIONED BY (dt STRING) STORED AS ORC;注意几点video_id 字段加上平台来源因为不同平台的视频 ID 可能重复帧路径不要省略即使推理结果已经写进表里保留路径方便后续重新抽帧或补充分析分区字段建议用 dt和上游任务对齐避免跨分区扫描时效率低下。3.2 常见分析任务的 SQL 与优化有了这两张表日常分析就是标准的 Spark SQL 活了。比如统计各平台每天上传视频的总时长和分辨率分布SELECT platform, dt, COUNT(*) AS video_cnt, SUM(duration_sec) AS total_duration, AVG(width * height) AS avg_pixels FROM dwd_video_meta WHERE dt 2024-01-01 GROUP BY platform, dt;再比如查某个时间段内模型识别出特定场景的帧数占比SELECT video_id, COUNT(*) AS frame_cnt, SUM(IF(detect_result[person] 0.5, 1, 0)) AS person_frame_cnt FROM dwd_frame_meta WHERE dt 2024-01-01 AND scene_label outdoor GROUP BY video_id;这类 SQL 跑起来本身不难真正的性能瓶颈往往出在数据倾斜上。比如头部视频产生的帧数量比普通视频多几个数量级按 video_id 聚合时容易把某个执行器打满。处理思路是增加一个随机分桶键比如concat(video_id, _, rand()*10)先做局部聚合再合并对超大视频单独处理把它们的帧路径单独拆出来跑一个专用任务适度开启 Spark 的 Adaptive Query ExecutionAQE让优化器自动处理倾斜。这一项在 Spark 3.x 里已经成熟我实测能解决大部分倾斜问题。4. 推理环节的分布式协作外部推理集群和 Executor 怎么配合4.1 为什么不要在 Executor 里直接跑深度学习模型很多人一开始会把模型加载到 Spark 执行器里在 map 函数里直接调用 PyTorch 或 TensorFlow。小规模实验没问题上了生产就会被两个问题卡住。第一个问题是执行器里的模型副本数不可控。每个执行器都要加载一份模型参数集群有 200 个执行器就加载 200 份光是显存和内存开销就让人头大。而模型推理又是典型的 GPU 密集型任务Spark 的 CPU 执行器资源模型和 GPU 推理根本不匹配。第二个问题是失败重跑的成本被放大。如果一个执行器处理到一半崩溃Spark 会重新调度任务模型重新加载、数据重新读取推理结果全部从头来。而视频任务往往伴随较大的帧数据重跑代价很高。所以生产上更合理的做法是Spark 负责组织推理任务真正跑模型的是独立推理服务。4.2 将推理服务封装成 Spark 可调用的算子我常用的方式是让执行器把一批帧路径或帧内容打包成请求发给推理服务取回 JSON 结果再写回 DataFrame。具体流程Spark 读取帧路径列表按批次分组执行器从对象存储读取帧数据或直接传路径调用推理服务的 HTTP/gRPC 接口传入图片数据或路径获取推理结果标签、置信度、框坐标将结果解析后写入新列或目标表。代码层面就是把 UDF 包一层def infer_batch(path_iter): images load_frames_from_storage(list(path_iter)) resp requests.post( http://inference-cluster:8080/v1/predict, json{images: images}, timeout30 ) return [json.loads(line) for line in resp.text.split(\n)] result_df frame_df.repartition(200) \ .mapPartitions(infer_batch) \ .toDF(...)重点不是代码本身而是背压控制。推理服务有吞吐上限如果 Spark 侧不分批、疯狂往里灌请求服务就会超时甚至被打挂。我当时的做法是根据推理服务的压测结果给每个分区设置一个“最大待处理量”用流式拉取来代替一次性大批量推送控制背压。另外推理请求最好走 HTTP 长连接或 gRPC避免频繁建立连接的开销。超时要设置重试并且重试间隔要大于服务的恢复时间否则重试高峰会把服务再一次压垮。4.3 写回结果的幂等性和数据一致性推理任务写回结果的时候要特别小心“幂等”。因为 Spark 任务失败会重跑重跑时模型推理结果一定是一样的但写入过程如果不幂等就会产生重复记录。我推荐先把结果写到一张临时表或者临时目录最后统一覆盖到目标分区。换句话说不要让每个执行器直接去写目标表而是让整个任务完成后再执行一次原子性的写入操作。这样即使中途某个分区重跑结果也只会覆盖同一个临时区域不会污染全量数据。如果你的上游数据本身可能变化比如视频文件被重新上传、内容被重新审核还需要在表里加入process_time字段每次任务执行时更新允许下游查询拿到最新版本。这个细节不做线上数据对账时会非常痛苦。5. 流式场景的落地增量消费视频文件与状态清理5.1 用结构化流监控视频落地目录视频处理不全是离线批处理。很多场景下视频是持续产生的比如监控摄像头不断上传录像、用户不断上传短视频。如果每次都全量扫描历史视频时间和资源都扛不住所以要引入流式处理。Spark Structured Streaming 并不能直接吃掉“一段视频二进制流”但它完全可以监控“视频文件落地的目录”或“视频上传消息队列”。一个简单但实用的设计视频上传 - 写入对象存储 发送消息到 Kafka Spark Structured Streaming 订阅 Kafka - 接收文件路径和视频元数据 - 触发切帧任务落地到临时目录 - 再消费切帧结果进入推理管道这种“消息触发 文件落地 流式确认”的组合比试图在流里直接解码视频靠谱得多。Kafka 里的每条消息只代表一个视频的引用真正的流式处理发生在视频文件已经完整落盘之后。5.2 状态管理与检查点的现实问题Structured Streaming 做窗口聚合时会用到状态存储比如“统计过去 5 分钟新上传视频的时长总和”或者“按摄像头 ID 聚合每天的流量”。这些状态默认放在执行节点的内存里任务重启后如果没配置检查点状态就全丢了。检查点的配置也很简单但有一个容易踩的坑目录隔离。同一个检查点目录不能让多个流处理作业共用否则状态互相覆盖。我在一个项目里见过两个逻辑完全不同的作业共用同一个 checkpoint 路径结果一边的重启把另一边的偏移量全重置了数据重复消费了一大片。另一个现实问题是状态膨胀。视频流任务往往按摄像头或用户维度做状态聚合如果维度基数很大状态数据会不断增长。我在生产环境一般会设置状态 TTL。比如.agg(...) .option(stateStoreMinDeltas, 5)同时用.dropDuplicates(video_id)之类的操作对重复消息做去重避免上游重发导致的状态反复。检查点目录也要定期清理历史版本不然小文件越积越多最终拖垮恢复速度。6. 生产环境里容易翻车的五个细节6.1 小文件与文件数的平衡视频链路里最典型的问题就是小文件失控。切帧产生大量小文件、推理结果按天分区写入又产生大量小分区文件两者都会让 Spark 的元数据开销暴涨。我的处理习惯是每个分区落盘前先 coalesce 或 repartition 控制输出文件数目标大小在 128MB 到 256MB 之间。比如结果表每天的数据量是 2GB把分区数控制在 8 到 16 个即可。同时在切帧环节尽量用二进制大文件或 Parquet 打包避免一个视频一套文件目录的“散弹式”布局。6.2 输入路径里的“脏字符”和跨平台文件名视频文件路径里经常出现空格、中文、括号、#、%等特殊字符。FFmpeg 命令行处理这些字符时如果不加引号或转义命令会解析出错Spark 在读取路径时也可能出现意料之外的列分割。我在代码里统一做了一层“路径清洗”解析到元数据后把路径中容易引起歧义的字符替换为下划线同时保存一个original_path字段保留原值。另外对外部系统传来的文件路径一律显式指定编码格式防止中文文件名在不同操作系统间转换时出现乱码。6.3 数据倾斜与 Executor 内存前面提到了数据倾斜这里再强调一个内存上的坑处理带图片字节的 DataFrame 时堆外内存经常被打爆。图片以字节数组形式存在 DataFrame 里会比普通字段占用更多的堆内存和堆外内存。如果你的执行器配置沿用默认的spark.memory.offHeap.enabledfalse很容易出现 OOM。建议调大堆外内存或者在数据进入 DataFrame 之前先做压缩用完立即释放引用。再者Spark 的默认并行度往往不够尤其是mapPartitions里做推理请求时分区数少了会浪费集群资源多了又容易压垮下游服务。我一般从实际推理服务的吞吐倒推开 Spark 并行度。比如推理服务能支撑 200 QPS单个分区每次推理耗时 2 秒那么分区并发控制在 400 左右相对合理。6.4 上下游水位对齐视频数据处理链路通常跨越多个任务切帧、元数据入库、推理、结果聚合、报表产出。任务和任务之间的依赖如果不做水位对齐下游很容易读到不完整的数据。我用的是“数据就绪标记”方案。每一层任务完成并校验数据量后往一个控制表写入分区就绪状态。下游任务启动前先读控制表确认上游分区全部就绪再执行。这样比单纯的定时调度要可靠得多也避免了“执行完了但数据没齐”的尴尬。6.5 测试时一定要用生产级视频最后一条也是我栽过跟头的开发联调时用了一段 10 秒的短视频各种逻辑都正常到了生产却频频失败。原因也很朴素——生产视频有各种编码格式、异常时长、损坏帧、旋转元数据测试短片完全没覆盖到。建议准备一个“视频多样性测试集”包含不同编码H.264、H.265、VP9不同分辨率720p、1080p、4K带旋转元数据的视频时长只有 1 秒的短视频和十几小时的长视频中间有损坏帧的视频。拿这套测试集跑一遍 Spark 任务能提前暴露大多数异常分支。我在项目上线前专门留了两天做这种“脏数据压测”虽然枯燥但在生产里省下的排查时间远超这两天。回到开头说的那段经历当我终于把“切、存、算分离”的思路理顺、把推理环节独立出去、把元数据模型按业务需求设计好之后那批视频处理任务才算真正跑稳了。视频数据处理的难点不在于 Spark 的 API 有多复杂而在于你愿不愿意先花时间把视频这种“非结构化大文件”变成“结构化小单元”再交给 Spark 这个擅长结构化并行的计算引擎。每一步想清楚“谁该干什么”链路自然就顺了。最后再分享一个小习惯我会在每个处理环节都保留“中间结果抽样”——切帧完成后抽几张图看看推理完成后抽几条 JSON 看看结果表写完再 count 一下。视频链路中间环节多出错往往不会报错而是“静默错误”。多看一眼抽样结果比事后对账高效得多。 先说个我自己的经历早几年接到过一批“把平台存量视频全部跑一遍内容标签”的需求当时我的第一反应是“把视频路径发给 Spark像处理普通日志一样搞个 map 就行”。结果任务一跑就连续踩坑从输入格式到执行器内存全都不对劲最后把链路整体重做才稳定下来。如果你也在规划类似的项目——不管是监控视频、短视频还是长视频内容有一个结论可以先放在前面用 Spark 处理视频数据核心难点从来不在“视频”本身而在“怎么把视频拆成适合分布式计算的样子”。视频文件是出了名的二进制大块头直接扔给 Spark 去读并非不能跑但代价极高。真正的做法是让 Spark 负责调度、切帧、元数据提取、推理任务的派发与结果聚合让每个执行器只处理“一个视频的某一段”或者“一批帧的路径”。这篇文章我会按这个思路把整个链路的架构设计、切帧策略、元数据建模、推理协作、流式落地的实操细节都拆开讲希望能给你省掉我当年踩过的那些坑。1. 视频和大数据的真实关系先想清楚再动手1.1 Spark 在视频链路里到底扮演什么角色视频数据有一个和大数据常见场景很不一样的特点它的“数据量”不只是文件体积大还在于单条有效信息的提取成本高。一份业务日志读进来就能查一段视频你得先解封装、解码、抽帧可能还要调用模型推理才能得到“这帧画面里有行人”这种业务字段。所以在设计系统前必须把角色分清Spark 的角色执行分布式任务调度、元数据管理、帧路径分发、推理结果聚合、统计分析其它组件的角色FFmpeg 负责解码切帧推理服务负责模型计算对象存储或 HDFS 负责帧文件持久化不该做的事让 Spark 执行器在一个进程里直接解码视频流或者把超大二进制文件作为 DataFrame 的一行传来传去。我后来在项目里把这条链路叫“切、存、算分离”。切是切帧存是帧文件和元数据算是标签和统计。三者各司其职Spark 才能发挥它真正擅长的并行计算能力。1.2 不建议用 Spark 直接啃视频文件的三个原因网上能看到不少“用 Spark 读视频”的 Demo大多是先读二进制再调用解码库。真放到生产环境至少有三个问题一是输入分片和视频解码不对齐。Spark 读取文件时会按照块大小切分输入而视频文件的分片点是随机的。一个分片可能落在关键帧之前或之后FFmpeg 在解码时需要依赖前面的帧信息导致分片边界上大量重试和重复解码严重拖慢处理速度。二是数据本地性失效。视频文件通常在对象存储或 HDFS 上连续存放分片后各执行器拿到的片段是分散的。Spark 会把任务调度到离数据近的节点但解码器需要在本地临时生成完整的解码上下文本地性优势被削弱还会占用大量临时磁盘。三是容错粒度太粗。如果某个执行器在处理一个大视频文件时崩溃整个任务从该文件的起始位置重跑中间已经解码完的帧全部作废。视频文件越大这种“功亏一篑”的成本越高。所以标准做法是先切帧再让 Spark 分析帧或帧元数据。切帧这一步虽然也消耗算力但它可以做到“按文件粒度并行、每个文件独立失败重试”天然适合分散到批量任务里去做。2. 海量视频的分布式切帧两种可行路线2.1 先切帧再读最稳的批处理组合切帧不是只有一种姿势我实际用过两种先说最稳的一种。把视频文件路径做成分区输入每个 Spark 任务处理一个或多个完整视频文件任务内部调用 FFmpeg 子进程完成解码抽帧抽出来的帧直接写到对象存储或 HDFS同时在 DataFrame 里记录路径、时间戳、视频 ID 等元数据。简单说Spark 不直接解视频而是管理“解视频的进程”。这里的 Spark 代码并不复杂核心是让每个执行器通过 Java/Scala 的 ProcessBuilder 启动 FFmpeg并传递参数。需要注意的几点每个执行器分配的 CPU 要和 FFmpeg 子进程的线程数匹配避免超售FFmpeg 输出要重定向到日志系统不然子进程输出积压会阻塞执行抽帧周期要根据业务需求设置比如每 5 秒一帧、每 10 秒一帧或者按场景切换抽帧不要无脑抽全部帧。我当时在一批监控视频上用过一个折中策略关键场景画面变化较大密集抽帧静止画面间隔抽帧。这样既控制帧总量又保证下游推理覆盖到有效内容。2.2 帧在存储上的布局直接决定后续效率切完帧只是第一步帧文件怎么落盘决定了之后 Spark 读起来顺不顺畅。经验是不要让每个视频把所有帧摊开成一堆小 JPEG 文件。比如一个 30 分钟的 1080p 视频按每秒一帧抽就是 1800 张图几万个视频就是几千万张小文件。小文件既占用文件系统的元数据内存又拖慢后续推理任务的读取速度。我自己常用的布局有两种场景推荐存储布局原因帧要长期保存、供多轮模型使用按“视频 ID 时间段”分桶打包成 Parquet帧数据以压缩字节数组存储文件数可控Spark 读取时可批量解码帧只用于一次推理过期可删抽出帧后直接转成 TFRecord 或自定义二进制格式推理完即可清理减少对象存储压力避免小文件堆积如果你仍然想保留 JPEG 原图也至少按视频或按天分目录避免把几百万张图堆在同一个前缀下。很多对象存储对单前缀下的文件数量有限制这一点等文件多起来再改会非常痛苦。3. 把视频元数据变成可查询的数据资产Spark SQL 建模3.1 元数据表怎么设计才不会后悔切帧和推理之后真正落到数据仓库里的核心资产是元数据。一个好的元数据模型能让后续所有分析工作变得简单反之则会让每条业务查询都变成一场灾难。我通常会建两张表一张是视频维度表一张是帧维度表。视频维度表存的是视频本身的属性CREATE TABLE dwd_video_meta ( video_id STRING COMMENT 视频唯一ID, platform STRING COMMENT 来源端, bucket STRING COMMENT 对象存储桶, path STRING COMMENT 视频文件路径, duration_sec DOUBLE COMMENT 视频时长, width INT, height INT, fps DOUBLE, codec STRING, upload_time TIMESTAMP COMMENT 上传时间, etl_time TIMESTAMP COMMENT 入库时间 ) PARTITIONED BY (dt STRING) STORED AS ORC;帧维度表存的是每一帧或者每一段时间窗口的分析结果CREATE TABLE dwd_frame_meta ( video_id STRING, frame_idx BIGINT, pts_time DOUBLE COMMENT 帧时间戳, frame_path STRING COMMENT 帧文件存储路径, scene_label STRING, detect_result MAPSTRING, DOUBLE COMMENT 模型输出比如目标框和置信度 ) PARTITIONED BY (dt STRING) STORED AS ORC;注意几点video_id 字段加上平台来源因为不同平台的视频 ID 可能重复帧路径不要省略即使推理结果已经写进表里保留路径方便后续重新抽帧或补充分析分区字段建议用 dt和上游任务对齐避免跨分区扫描时效率低下。3.2 常见分析任务的 SQL 与优化有了这两张表日常分析就是标准的 Spark SQL 活了。比如统计各平台每天上传视频的总时长和分辨率分布SELECT platform, dt, COUNT(*) AS video_cnt, SUM(duration_sec) AS total_duration, AVG(width * height) AS avg_pixels FROM dwd_video_meta WHERE dt 2024-01-01 GROUP BY platform, dt;再比如查某个时间段内模型识别出特定场景的帧数占比SELECT video_id, COUNT(*) AS frame_cnt, SUM(IF(detect_result[person] 0.5, 1, 0)) AS person_frame_cnt FROM dwd_frame_meta WHERE dt 2024-01-01 AND scene_label outdoor GROUP BY video_id;这类 SQL 跑起来本身不难真正的性能瓶颈往往出在数据倾斜上。比如头部视频产生的帧数量比普通视频多几个数量级按 video_id 聚合时容易把某个执行器打满。处理思路是增加一个随机分桶键比如concat(video_id, _, rand()*10)先做局部聚合再合并对超大视频单独处理把它们的帧路径单独拆出来跑一个专用任务适度开启 Spark 的 Adaptive Query ExecutionAQE让优化器自动处理倾斜。这一项在 Spark 3.x 里已经成熟我实测能解决大部分倾斜问题。4. 推理环节的分布式协作外部推理集群和 Executor 怎么配合4.1 为什么不要在 Executor 里直接跑深度学习模型很多人一开始会把模型加载到 Spark 执行器里在 map 函数里直接调用 PyTorch 或 TensorFlow。小规模实验没问题上了生产就会被两个问题卡住。第一个问题是执行器里的模型副本数不可控。每个执行器都要加载一份模型参数集群有 200 个执行器就加载 200 份光是显存和内存开销就让人头大。而模型推理又是典型的 GPU 密集型任务Spark 的 CPU 执行器资源模型和 GPU 推理根本不匹配。第二个问题是失败重跑的成本被放大。如果一个执行器处理到一半崩溃Spark 会重新调度任务模型重新加载、数据重新读取推理结果全部从头来。而视频任务往往伴随较大的帧数据重跑代价很高。所以生产上更合理的做法是Spark 负责组织推理任务真正跑模型的是独立推理服务。4.2 将推理服务封装成 Spark 可调用的算子我常用的方式是让执行器把一批帧路径或帧内容打包成请求发给推理服务取回 JSON 结果再写回 DataFrame。具体流程Spark 读取帧路径列表按批次分组执行器从对象存储读取帧数据或直接传路径调用推理服务的 HTTP/gRPC 接口传入图片数据或路径获取推理结果标签、置信度、框坐标将结果解析后写入新列或目标表。代码层面就是把 UDF 包一层def infer_batch(path_iter): images load_frames_from_storage(list(path_iter)) resp requests.post( http://inference-cluster:8080/v1/predict, json{images: images}, timeout30 ) return [json.loads(line) for line in resp.text.split(\n)] result_df frame_df.repartition(200) \ .mapPartitions(infer_batch) \ .toDF(...)重点不是代码本身而是背压控制。推理服务有吞吐上限如果 Spark 侧不分批、疯狂往里灌请求服务就会超时甚至被打挂。我当时的做法是根据推理服务的压测结果给每个分区设置一个“最大待处理量”用流式拉取来代替一次性大批量推送控制背压。另外推理请求最好走 HTTP 长连接或 gRPC避免频繁建立连接的开销。超时要设置重试并且重试间隔要大于服务的恢复时间否则重试高峰会把服务再一次压垮。4.3 写回结果的幂等性和数据一致性推理任务写回结果的时候要特别小心“幂等”。因为 Spark 任务失败会重跑重跑时模型推理结果一定是一样的但写入过程如果不幂等就会产生重复记录。我推荐先把结果写到一张临时表或者临时目录最后统一覆盖到目标分区。换句话说不要让每个执行器直接去写目标表而是让整个任务完成后再执行一次原子性的写入操作。这样即使中途某个分区重跑结果也只会覆盖同一个临时区域不会污染全量数据。如果你的上游数据本身可能变化比如视频文件被重新上传、内容被重新审核还需要在表里加入process_time字段每次任务执行时更新允许下游查询拿到最新版本。这个细节不做线上数据对账时会非常痛苦。5. 流式场景的落地增量消费视频文件与状态清理5.1 用结构化流监控视频落地目录视频处理不全是离线批处理。很多场景下视频是持续产生的比如监控摄像头不断上传录像、用户不断上传短视频。如果每次都全量扫描历史视频时间和资源都扛不住所以要引入流式处理。Spark Structured Streaming 并不能直接吃掉“一段视频二进制流”但它完全可以监控“视频文件落地的目录”或“视频上传消息队列”。一个简单但实用的设计视频上传 - 写入对象存储 发送消息到 Kafka Spark Structured Streaming 订阅 Kafka - 接收文件路径和视频元数据 - 触发切帧任务落地到临时目录 - 再消费切帧结果进入推理管道这种“消息触发 文件落地 流式确认”的组合比试图在流里直接解码视频靠谱得多。Kafka 里的每条消息只代表一个视频的引用真正的流式处理发生在视频文件已经完整落盘之后。5.2 状态管理与检查点的现实问题Structured Streaming 做窗口聚合时会用到状态存储比如“统计过去 5 分钟新上传视频的时长总和”或者“按摄像头 ID 聚合每天的流量”。这些状态默认放在执行节点的内存里任务重启后如果没配置检查点状态就全丢了。检查点的配置也很简单但有一个容易踩的坑目录隔离。同一个检查点目录不能让多个流处理作业共用否则状态互相覆盖。我在一个项目里见过两个逻辑完全不同的作业共用同一个 checkpoint 路径结果一边的重启把另一边的偏移量全重置了数据重复消费了一大片。另一个现实问题是状态膨胀。视频流任务往往按摄像头或用户维度做状态聚合如果维度基数很大状态数据会不断增长。我在生产环境一般会设置状态 TTL。比如.agg(...) .option(stateStoreMinDeltas, 5)同时用.dropDuplicates(video_id)之类的操作对重复消息做去重避免上游重发导致的状态反复。检查点目录也要定期清理历史版本不然小文件越积越多最终拖垮恢复速度。6. 生产环境里容易翻车的五个细节6.1 小文件与文件数的平衡视频链路里最典型的问题就是小文件失控。切帧产生大量小文件、推理结果按天分区写入又产生大量小分区文件两者都会让 Spark 的元数据开销暴涨。我的处理习惯是每个分区落盘前先 coalesce 或 repartition 控制输出文件数目标大小在 128MB 到 256MB 之间。比如结果表每天的数据量是 2GB把分区数控制在 8 到 16 个即可。同时在切帧环节尽量用二进制大文件或 Parquet 打包避免一个视频一套文件目录的“散弹式”布局。6.2 输入路径里的“脏字符”和跨平台文件名视频文件路径里经常出现空格、中文、括号、#、%等特殊字符。FFmpeg 命令行处理这些字符时如果不加引号或转义命令会解析出错Spark 在读取路径时也可能出现意料之外的列分割。我在代码里统一做了一层“路径清洗”解析到元数据后把路径中容易引起歧义的字符替换为下划线同时保存一个original_path字段保留原值。另外对外部系统传来的文件路径一律显式指定编码格式防止中文文件名在不同操作系统间转换时出现乱码。6.3 数据倾斜与 Executor 内存前面提到了数据倾斜这里再强调一个内存上的坑处理带图片字节的 DataFrame 时堆外内存经常被打爆。图片以字节数组形式存在 DataFrame 里会比普通字段占用更多的堆内存和堆外内存。如果你的执行器配置沿用默认的spark.memory.offHeap.enabledfalse很容易出现 OOM。建议调大堆外内存或者在数据进入 DataFrame 之前先做压缩用完立即释放引用。再者Spark 的默认并行度往往不够尤其是mapPartitions里做推理请求时分区数少了会浪费集群资源多了又容易压垮下游服务。我一般从实际推理服务的吞吐倒推开 Spark 并行度。比如推理服务能支撑 200 QPS单个分区每次推理耗时 2 秒那么分区并发控制在 400 左右相对合理。6.4 上下游水位对齐视频数据处理链路通常跨越多个任务切帧、元数据入库、推理、结果聚合、报表产出。任务和任务之间的依赖如果不做水位对齐下游很容易读到不完整的数据。我用的是“数据就绪标记”方案。每一层任务完成并校验数据量后往一个控制表写入分区就绪状态。下游任务启动前先读控制表确认上游分区全部就绪再执行。这样比单纯的定时调度要可靠得多也避免了“执行完了但数据没齐”的尴尬。6.5 测试时一定要用生产级视频最后一条也是我栽过跟头的开发联调时用了一段 10 秒的短视频各种逻辑都正常到了生产却频频失败。原因也很朴素——生产视频有各种编码格式、异常时长、损坏帧、旋转元数据测试短片完全没覆盖到。建议准备一个“视频多样性测试集”包含不同编码H.264、H.265、VP9不同分辨率720p、1080p、4K带旋转元数据的视频时长只有 1 秒的短视频和十几小时的长视频中间有损坏帧的视频。拿这套测试集跑一遍 Spark 任务能提前暴露大多数异常分支。我在项目上线前专门留了两天做这种“脏数据压测”虽然枯燥但在生产里省下的排查时间远超这两天。回到开头说的那段经历当我终于把“切、存、算分离”的思路理顺、把推理环节独立出去、把元数据模型按业务需求设计好之后那批视频处理任务才算真正跑稳了。视频数据处理的难点不在于 Spark 的 API 有多复杂而在于你愿不愿意先花时间把视频这种“非结构化大文件”变成“结构化小单元”再交给 Spark 这个擅长结构化并行的计算引擎。每一步想清楚“谁该干什么”链路自然就顺了。最后再分享一个小习惯我会在每个处理环节都保留“中间结果抽样”——切帧完成后抽几张图看看推理完成后抽几条 JSON 看看结果表写完再 count 一下。视频链路中间环节多出错往往不会报错而是“静默错误”。多看一眼抽样结果比事后对账高效得多。