用Spark重构视频数据处理管线：从单机脚本到分布式批处理的完整实践

从流到批我用Spark重构视频数据处理管线的全过程差不多两年前我们团队接到了一个有点棘手的需求——每天要处理几十万条短视频做内容审核、场景识别、质量评分还要把结果实时同步给推荐系统。最开始用的是单机脚本加一堆第三方库扛到数据量涨上来之后任务跑一天都跑不完内存频繁爆掉队列里积压的任务越来越多。后来我花了差不多一个月时间把整套处理流程迁移到Spark上从架构设计到代码落地把视频数据在分布式环境下的一套完整处理方案给趟了出来。这篇文章就把我自己的实操过程、踩过的坑和最终沉淀下来的方案整理出来希望能给正打算用Spark做视频数据处理的同学一些能直接上手的参考。我这里的处理对象并不是原始视频文件的编解码而是围绕视频文件的元数据解析、抽帧、视觉特征提取、质量评分、标签化这一整套流程。换句话说Spark负责的是“视频文件引发的海量计算”而不是用Spark去解码视频流。这个定位很关键想明白了整个架构就不会跑偏。1. 为什么选Spark来做视频数据处理很多人一听到“视频数据处理”第一反应是FFmpeg、OpenCV、GPU集群很少有人会直接想到Spark。但实际上视频领域的数据体量一旦大起来真正的瓶颈不完全是“解一帧画面要多久”而是“几十万个视频的元数据怎么管”“抽出来的特征怎么聚合”“不同来源的视频怎么统一规整”。这些恰恰是Spark擅长的场景。1.1 视频数据处理的两条不同路线我把视频处理分成两条路线先搞清楚自己在哪条路线上再做技术选型才不会乱。第一条路线叫“流内计算”也就是对视频像素本身做计算比如解码、抽帧、目标检测、画质增强。这类任务计算密集、数据局部性强一个视频文件最好能被一个计算节点完整处理避免反复传输帧数据。这条路线通常用FFmpeg配合GPU推理框架更合适。第二条路线叫“元数据与特征聚合计算”也就是把视频文件的路径、时长、分辨率、编码格式、抽帧结果、AI识别标签、用户行为统计这些信息汇总起来做大规模清洗、关联、聚合、分析。这类任务数据量大、格式杂、需要分布式存储和弹性计算Spark是很好的选择。我在实际项目中同时用到了两条路线边缘节点用FFmpeg做抽帧和初步分析产出的结构化结果汇聚到Spark里做全局处理。很多团队遇到的问题是强行把第一条路线塞给Spark结果发现处理速度还不如单机整个集群的资源利用率却高得吓人。想清楚自己的核心瓶颈在哪才知道Spark该放在什么位置。1.2 Spark能否处理“视频文件”本身直说结论Spark本身不能直接解析视频编码格式也没有内置的MP4/FLV解码器。但Spark完全可以管理“视频处理任务”和“处理结果”。我的做法是用Spark的Task调度能力去调度分布在多台机器上的FFmpeg进程让每个Executor启动一个FFmpeg子进程完成对某个视频文件的抽帧再把产物路径和元数据以Record的形式写回DataFrame。这种“Spark 外部工具进程”的模式在行业里很常见类似于Spark调用外部Python脚本做模型推理。关键在于资源的整合和任务的编排。Spark负责并行度管理、失败重试、数据血缘外部进程负责真正的计算。这样一个视频文件的所有中间产物包括抽帧截图、音频波形、质量评分、AI标签都能统一落到分布式文件系统里后续的离线分析、检索、样本生成全部走Spark读取。1.3 相比传统单机方案的优势我最早的单机方案是Python脚本遍历目录对每个视频文件调用FFmpeg再把结果写进SQLite。这个方案在小规模数据下完全没问题代码简单、调试方便。但数据量上了规模之后我遇到了几个难以回避的问题。第一处理速度无法横向扩展。单机跑满就是跑满加机器没有意义除非重写任务调度逻辑。第二任何一次内存溢出或进程卡死都可能让整个任务中止没有任务级重试机制。第三数据没有统一视图要么散落在各个子目录里要么汇总到一张文本表里做质量分析、时长分布分析、标签聚合时非常痛苦。Spark解决的就是这三个问题。处理任务的并行度可以通过调整分区数控制任务失败由Spark统一重试所有输出结果以规范的列式存储落盘用SQL就能做多维分析。说白了Spark给视频数据处理带来的是一种工程化的能力把原来靠脚本维护的临时方案变成可以持续迭代的数据管线。2. 视频数据处理的整体架构与核心设计思路在动手写代码之前我花了不少时间梳理整个架构。视频数据处理跟普通日志处理不一样文件体积大、处理步骤多、中间产物杂如果架构设计阶段不把这些问题想清楚后面写代码会非常痛苦。2.1 主流程的四个阶段整个管线我按照数据流向拆成了四个阶段采集与注册、分析与抽帧、特征提取、结果汇聚。采集与注册阶段负责监听文件目录或者消息队列把新视频文件的路径、大小、MD5、上传时间等元数据登记到Spark的DataFrame中。这个阶段不做重计算只保证“有哪些视频需要处理”这件事是清晰的。分析与抽帧阶段对每个视频文件启动外部FFmpeg进程获取视频流的编码格式、分辨率、帧率、时长、关键帧位置并根据配置抽取出若干帧画面。抽取出来的帧被存为JPEG文件路径与视频文件ID关联。特征提取阶段把抽取的帧读入内存做统一张量化处理交给预训练好的图像分类模型或者目标检测模型得到帧级标签向量和置信度。这个阶段因为集中在Executor中做适合使用批量推理框架比如OpenCV的DNN模块或者ONNX Runtime的批量模式。结果汇聚阶段把一个视频的所有帧级特征做聚合得到视频级标签分布、画质评分、场景切变率、内容风险分等结果。再关联上原始元数据写入统一的分析结果表供在线服务或离线报表使用。这四个阶段听起来很简单但真正落地时每个阶段都有不少细节。比如抽帧阶段的并发度控制如果并行抽帧的任务太多每台机器的CPU会被打满反而拖慢整体速度。后来我在配置里对并发抽帧任务数做了上限限制每个Executor同时最多跑两个FFmpeg进程。2.2 为什么选用外部进程而非纯JVM方案我考虑过在JVM里直接调用JavaCV或者FFmpeg的JNI接口来做抽帧但最终没有选择这条路。原因有三个。第一JVM内存压力。视频抽帧本质上是C/C库的工作通过JNI调用时内存管理跨语言边界稍有不慎就容易出现native内存泄漏排查起来极其困难。而外部进程方式下FFmpeg的内存空间与JVM完全隔离进程结束后操作系统自动回收资源。第二部署灵活性。纯JVM方案意味着每台机器都要配置复杂的native依赖库不同操作系统版本、不同CPU架构都会带来兼容性问题。外部进程方式只需要在镜像里装一个FFmpeg可执行文件所有逻辑通过命令行参数控制部署和维护都简单得多。第三故障隔离。FFmpeg在处理损坏的视频文件时可能直接崩溃外部进程方式下这只是单个进程的异常退出Spark可以捕获非零退出码并重试或者跳过如果是JNI调用一个native层的崩溃可能直接带崩整个Executor进程损失就大了。我也不是完全不用JavaCV它更适合在单机或者少量并发的场景使用。在Spark分布式环境下稳定性和隔离性比调用方便更重要。2.3 数据表模型设计从视频文件到结构化结果把视频处理结果结构化是整个项目的核心之一。我先设计的是事实表然后才是维度表和聚合表。主表叫视频元数据表每个视频文件一行记录包含文件ID、原始路径、上传时间、时长、分辨率、编码格式、文件大小、MD5值这些基础字段。然后是帧级特征表每个视频可能产生几十到几百行记录每行代表一帧画面的特征结果包含帧序号、帧时间戳、JPEG存储路径、模型版本、标签向量JSON、置信度、画质指标等。最后是视频级标签聚合表通过Spark SQL的GROUP BY操作把帧级特征聚合成视频维度包含主要标签TOP5、风险标签、画质评分、场景多样性指数等。这套表结构的好处是既保留了最细粒度的帧级数据又提供了高效的视频级查询路径。做样本标注时可以下钻到帧级特征做线上服务时只读视频级聚合表思路清晰。3. 从零搭建Spark视频处理管线关键步骤与代码实现架构设计完成之后下面进入实操环节。我会把整个管线的关键步骤拆开讲包括环境准备、数据读取、调用FFmpeg抽帧、调用模型推理、结果写入这套流程是完整的、可落地的。3.1 环境准备与依赖安装我的运行环境是3台8核32GB内存的服务器组成的Spark Standalone集群。这里不讨论生产环境用YARN的情况原理是共通的。操作系统是CentOS 7.9安装了Spark 3.2.0对应Scala 2.12Java 1.8Python 3.8FFmpeg 4.2。我选择用PySpark而非Scala编写主要的处理逻辑因为视频处理涉及大量与Python生态交互的环节比如调用图像模型、处理JSON、对接OpenCV等Python开发效率高得多。而计算密集的聚合操作走Spark内置算子性能差异可以接受。FFmpeg安装在每台Worker节点的/opt/ffmpeg/bin/ffmpeg路径下。不要只装在Driver节点上否则任务分发到Executor时会出现“找不到ffmpeg命令”的错误。还需要准备一个预训练图像模型。我用的是MobileNetV2量化版在ONNX Runtime里做推理。模型文件放在共享存储上各Executor首次使用时加载到本地内存。3.2 视频元数据采集与注册第一步是把目录下的视频文件列表“注册”成一个DataFrame。这一步不能直接用Python的os.listdir因为在大规模目录下单机遍历文件列表是瓶颈。我改用Spark的批量文件发现机制通过wholeTextFiles或者自定义InputFormat来获取文件状态。核心代码如下from pyspark.sql import SparkSession from pyspark.sql.functions import col, lit, md5, input_file_name spark SparkSession.builder \ .appName(VideoMetaRegister) \ .config(spark.sql.shuffle.partitions, 16) \ .getOrCreate() video_df spark.read.format(binaryFile) \ .option(pathGlobFilter, *.mp4) \ .option(recursiveFileLookup, true) \ .load(/data/videos/raw) video_meta_df video_df.select( md5(col(content)).alias(file_md5), col(path).alias(file_path), col(length).alias(file_size), input_file_name().alias(src_path) )这里用到的是Spark 3.0提供的binaryFile数据源格式。它把每个文件作为一个二进制列读取自动附带路径和大小信息对大批量小文件非常友好。视频文件虽然不算小但用binaryFile方式可以避免在Driver端集中枚举文件列表把文件发现也变成分布式的。如果你用的是Spark 2.x没有binaryFile格式可以用自定义PartitionedFile的路径方式但更省事的做法是升级到Spark 3.x版本这个特性差距还是很明显的。3.3 在Executor中调用FFmpeg抽帧这是整个管线里最需要细抠的部分。我封装了一个函数把FFmpeg的调用变成一个纯粹的Python函数交给RDD的mapPartitions方法去执行。import subprocess import json import os def extract_video_frames(path, output_dir, video_id): 调用FFmpeg抽取关键帧并获取基础元数据 返回: 帧信息列表 元数据字典 # 第一步获取视频流信息 probe_cmd [ /opt/ffmpeg/bin/ffprobe, -v, quiet, -print_format, json, -show_streams, -show_format, path ] probe_result subprocess.run( probe_cmd, capture_outputTrue, textTrue, timeout30 ) if probe_result.returncode ! 0: return {video_id: video_id, error: probe_failed, frames: []} meta json.loads(probe_result.stdout) video_stream next( (s for s in meta[streams] if s[codec_type] video), None ) if video_stream is None: return {video_id: video_id, error: no_video_stream, frames: []} duration float(video_stream.get(duration, 0)) width int(video_stream.get(width, 0)) height int(video_stream.get(height, 0)) codec video_stream.get(codec_name, unknown) # 第二步按固定间隔抽帧 frame_dir os.path.join(output_dir, video_id) os.makedirs(frame_dir, exist_okTrue) # 最多抽20帧避免太多导致处理过慢 n_frames min(20, max(5, int(duration / 10))) frame_files [] for i in range(n_frames): timestamp max(0.1, (i * duration / n_frames)) out_file os.path.join(frame_dir, fframe_{i:03d}.jpg) extract_cmd [ /opt/ffmpeg/bin/ffmpeg, -ss, str(timestamp), -i, path, -frames:v, 1, -q:v, 3, -y, out_file ] ret subprocess.run(extract_cmd, capture_outputTrue, timeout15) if ret.returncode 0 and os.path.exists(out_file): frame_files.append({ frame_index: i, timestamp: round(timestamp, 3), frame_path: out_file, width: width, height: height }) return { video_id: video_id, duration: duration, width: width, height: height, codec: codec, frames: frame_files, error: None }这个函数的关键点有三个。一是用ffprobe先探测视频流的真实信息而不是依赖文件扩展名因为实际场景里经常出现文件名是mp4但内容编码格式是H.265的情况。二是用固定时间间隔抽取而不是按关键帧抽取固定间隔更能反映视频内容的整体分布关键帧抽取很多时候抽出来的都是黑场或者片头。三是超时参数timeout一定要加有些损坏的视频FFmpeg会卡住如果没有超时控制整个Spark任务都会卡住。3.4 分布式处理主流程有了抽帧函数之后把它封装成Spark能够调度的任务。from pyspark.sql import Row import uuid frame_schema [video_id, frame_index, timestamp, frame_path] video_info_schema [video_id, duration, width, height, codec, error] def process_video_batch(iterator): results [] for row in iterator: video_path row.file_path video_id str(uuid.uuid4()) output_dir /data/videos/frames info extract_video_frames(video_path, output_dir, video_id) video_record Row( video_idinfo[video_id], durationinfo.get(duration, 0), widthinfo.get(width, 0), heightinfo.get(height, 0), codecinfo.get(codec, unknown), errorinfo.get(error) ) results.append(video_record) for frame in info.get(frames, []): results.append(Row( video_idvideo_id, frame_indexframe[frame_index], timestampframe[timestamp], frame_pathframe[frame_path] )) return iter(results) video_meta_rdd video_meta_df.rdd.mapPartitions(process_video_batch) frame_info_df video_meta_rdd.toDF(video_info_schema [f for f in frame_schema])这里我用mapPartitions而不是map原因是mapPartitions允许在一个批次内复用FFmpeg进程池、批量准备目录结构减少重复创建的开销。每个分区处理一个连续的视频文件集合也是更合理的调度粒度。不过这里有一个分支处理点需要说明信息记录和帧记录在同一个迭代器里返回但它们的schema不同直接合并成同一个DataFrame会非常混乱。实际项目中我选择把信息记录和帧记录分别收集成两个RDD再单独落表。上面代码只是一种思路为了演示方便实际使用时建议拆开。3.5 帧级特征提取与模型推理抽出来的帧是JPEG文件接下来要把它们变成机器学习模型能处理的向量。这一步我用ONNX Runtime批量处理。import onnxruntime as ort import numpy as np from PIL import Image class FrameFeatureExtractor: def __init__(self, model_path, input_size(224, 224)): self.session ort.InferenceSession(model_path) self.input_size input_size self.input_name self.session.get_inputs()[0].name self.output_name self.session.get_outputs()[0].name def preprocess(self, image_path): img Image.open(image_path).convert(RGB) img img.resize(self.input_size) arr np.asarray(img, dtypenp.float32) arr arr / 255.0 mean np.array([0.485, 0.456, 0.406], dtypenp.float32) std np.array([0.229, 0.224, 0.225], dtypenp.float32) arr (arr - mean) / std # PyTorch风格通道在前 arr np.transpose(arr, (2, 0, 1)) return np.expand_dims(arr, axis0) def predict_batch(self, image_paths): batch np.concatenate( [self.preprocess(p) for p in image_paths], axis0 ) outputs self.session.run([self.output_name], {self.input_name: batch}) return outputs[0]在Executor端使用时每个分区初始化一个特征提取器实例然后对该分区内的所有帧路径进行批量推理。def extract_features_batch(iterator): extractor FrameFeatureExtractor(/models/mobilenetv2.onnx) for batch_paths in chunked(iterator, 32): # batch_paths 是 (video_id, frame_path) 元组列表 paths [p for _, p in batch_paths] features extractor.predict_batch(paths) for (video_id, _), feature_vec in zip(batch_paths, features): label_id int(np.argmax(feature_vec)) confidence float(np.max(feature_vec)) yield (video_id, label_id, confidence, feature_vec.tolist())之所以批量推理是为了充分利用GPU/CPU的向量化计算能力。单张图片推理的启动开销占比太高32张一批的效率相比单张能提升约4倍。3.6 结果汇聚与写入特征提取完成之后所有视频的帧级标签还需要聚合成视频级标签。这里用SQL操作实现非常方便。frame_features_df.createOrReplaceTempView(frame_features) video_meta_df.createOrReplaceTempView(video_meta) video_label_agg_df spark.sql( SELECT video_id, COUNT(*) AS frame_cnt, PERCENTILE_APPROX(confidence, 0.5, 100) AS median_confidence, COLLECT_SET( CONCAT_WS(:, CAST(label_id AS STRING), CAST(ROUND(confidence, 4) AS STRING)) ) AS label_conf_pairs, MAX(confidence) AS max_confidence FROM frame_features GROUP BY video_id ).join(video_meta_df, video_id) video_label_agg_df.write.mode(overwrite) \ .format(parquet) \ .save(/data/videos/aggregated_results)写入格式我强烈推荐Parquet不要用CSV或者JSON。Parquet是列式存储后续按标签ID过滤、按置信度范围统计时只要扫相关列I/O开销小很多。而且Parquet原生支持压缩3GB的JSON数据转成Parquet基本能降到500MB左右。4. 性能调优并发度、资源分配与任务倾斜的处理跑起来不难但要跑得快、跑得稳就需要对Spark的执行机制有更深的理解。这里分享几个实际调优的经验都是我在真实集群上调过的。4.1 Executor资源配置的搭配逻辑视频处理分为两个主要阶段抽帧阶段是CPU密集型的特征提取阶段是CPU内存密集型的。这意味着Executor不能配太少的内存否则特征提取阶段会频繁GC甚至OOM。我最终的资源配比是配置项参数值说明spark.executor.memory8g每个Executor的堆内存spark.executor.cores2每个Executor占用2核spark.executor.instances6总共启动6个Executorspark.driver.memory4gDriver端主要负责调度不需要太大spark.sql.shuffle.partitions24shuffle分区数按数据总量调整每个Executor配2核而不是4核是因为抽帧阶段要起FFmpeg子进程子进程的内存不占用JVM堆但占用机器总内存。如果Executor配的核数太多同时启动的FFmpeg进程数量会翻倍机器内存可能不够用。4核机器配2核Executor留一半资源给外部进程是我实际测试下来最稳的配置。4.2 数据倾斜的发现与规避我遇到过一次明显的倾斜问题。有一个视频源目录下放着几个超长视频文件每个3小时以上其他的视频都是几分钟。用默认分区方式跑的时候处理超长视频的Executor任务要跑20分钟而其他任务一两分钟就结束了整体任务被这几个大文件拉长了近10倍。解决方式是抽帧阶段的输入分区不再按文件数量划分而是按文件大小估算的处理代价划分。Spark的repartition可以传一个分区键但更通用的做法是先把文件列表按大小粗略分桶让每个分区里的总文件大小相近。from pyspark.sql.functions import ntile, col # 按大小排个序再分桶让每个桶的总大小接近 ranked_df video_meta_df.orderBy(col(file_size).desc()) bucket_df ranked_df.withColumn(bucket, ntile(24).over(Window.orderBy(file_size))) # 每个bucket作为一个处理分区用ntile窗口函数把文件按大小排序后均匀分成24桶每桶数据量接近处理时间也接近。这样虽然牺牲了一点文件大小与处理时长的绝对精确映射但至少解决了“一个大文件拖垮整个任务”的问题。4.3 小文件问题的处理Spark处理视频还有个天然的矛盾视频文件BLOB格式很大动辄几百MB但处理后产出的帧文件是大量小JPEG几KB到几十KB不等。如果把这些小文件全部直接写到共享存储上会产生数万个零碎文件后续读取时NameNode压力巨大。我的解决办法是两段式存储第一段抽帧结果写到本地磁盘的临时目录通过OSS或HDFS的批量上传接口一次性同步到共享存储第二段在Spark里根本不直接管理这些JPEG文件的数据内容只管理它们的路径索引真正的帧文件作为“外部数据资产”由独立的文件服务管理。这样Spark维护的元数据量非常小几十万视频的帧索引表也就几百万行处理起来很轻松。5. 实操中常见的故障与排查经验跑视频处理管线不像跑WordCount那么简单直观视频文件格式杂、损坏率高、外部依赖多我实际跑下来遇到过很多稀奇古怪的问题。这里整理几类最常见的方便大家排查时少走弯路。5.1 FFmpeg进程卡死导致Executor僵住这是我在项目初期遇到的最头疼的问题。FFmpeg在解析一些格式不规范的文件时会进入阻塞状态既不退出也不报错导致executor的线程一直被占住。我排查了很久才发现单靠subprocess.run的timeout参数不够因为timeout只作用于整个调用过程但FFmpeg子进程自身可能在内部循环里不响应信号。后来我的解决方法是两层超时保护。第一层是subprocess的超时第二层是在调用前设置一个线程计时器如果超过60秒还没有返回直接terminate掉进程组。import signal def run_ffmpeg_with_guard(cmd, timeout30): def timeout_handler(signum, frame): raise TimeoutError(ffmpeg timeout) signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: ret subprocess.run(cmd, capture_outputTrue, timeouttimeout) signal.alarm(0) return ret except TimeoutError: # 杀死整个进程组 os.killpg(os.getpgid(subprocess.Popen(cmd).pid), signal.SIGKILL) raisesignal.alarm是所在进程维度的定时器保证即使subprocess阻塞在底层C库调用里也能被强行打断。实践中这个方案非常有效再没有出现过Executor无限挂死的情况。5.2 模型文件在每台节点上没有同步特征提取阶段出现过一种诡异的表现前几个Executor处理正常跑着跑着突然报“Cannot open file”。排查后发现是因为模型文件只放在Driver本机的目录里Spark在把任务分发到不同节点时部分节点上并没有这个模型文件。这个问题让我意识到Spark集群里任何被Executor使用的资源文件必须在所有Worker节点上都存在。要么把模型拷贝到共享存储如HDFS、OSS再由各节点加载要么在Spark提交脚本里加上--files参数带入每个Executor。我用的是后一种方式简单有效spark-submit \ --files /models/mobilenetv2.onnx \ --archives /opt/ffmpeg/bin/ffmpeg#ffmpeg \ video_processing_job.py--files会把模型文件分发到每个Executor的工作目录--archives则能把整个FFmpeg目录压缩包分发过去并自动解压。这样就不用手动到每台机器上装依赖了集群扩容时也不用额外操作。5.3 视频编码格式兼容性差异视频处理最麻烦的就是格式问题。H.264、H.265、VP9、AV1每种编码的解码开销都不一样有些老视频还是MPEG-2或者MJPEG格式。一开始全部用统一的FFmpeg命令处理结果发现有不少视频报错或者抽帧质量很差。解决方法是走双通道检测流程。先用ffprobe解析编码格式根据编码类型差异设置不同的处理参数。比如H.265的视频抽帧时加上-hwaccel cuda参数启用GPU硬解处理速度快了8倍MJPEG的视频因为编码简单可以不用开硬解反而软解质量更好。有些异常编码的视频则需要在抽帧时多转一道先转成中间格式再做特征提取。if codec hevc and use_gpu: extract_cmd.insert(1, -hwaccel) extract_cmd.insert(2, cuda)这套逻辑写进处理函数之后抽帧失败率从3.2%降到了0.4%处理速度也明显提升。错误本身就是数据的一部分不要指望所有视频文件都是标准的、可解码的。5.4 数据倾斜之外的“任务饿死”现象最后补充一个冷门但很实际的问题。在Standalone集群下Spark的公平调度默认不开启如果同时跑多个任务先提交的任务可能占满所有Executor后续任务一直等不到资源看起来像“卡死了”。排查方法也很简单查看Spark UI的Active Stages页面如果看到了大量的Pending任务而Executor全部Busy基本上就是任务饿死。我后来统一给所有视频处理任务设置了FAIR调度池再把不同优先级的任务放到不同的池子中。核心抽取任务放在高优先级池实验性质的离线分析任务放到低优先级池。这样即便集群满载重要任务也能分到足够的资源跑完。6. 把Spark视频处理能力延伸到更多场景整个管线跑通之后我一直在想它还能扩展到哪里。目前这套架构的底层是“统一的视频数据接入 分布式计算 结构化结果存储”这个底座可以支撑不少上层应用。6.1 从离线分析走向准实时处理如果用Spark Streaming或者Structured Streaming替换批处理的读取部分这套管线可以直接变成准实时系统。视频文件上传到对象存储后触发一个消息Streaming任务感知到新文件就立刻拉取处理延迟可以控制在分钟级别。处理结果写入在线存储推荐系统和审核系统就能拿到最新标签。我这里用的是批量模式为主因为视频处理本身耗时较长秒级实时意义不大分钟级准实时已经能覆盖大多数业务需求。6.2 用回归任务做视频画质评估帧级特征提取不只是能提取分类标签配合回归模型还能评估视频画质。比如我额外加了一个图像质量回归模型输入是抽帧图像输出是MOS分数。所有帧的MOS分数聚合起来就能得到整个视频的画质评分。画质评分可以用于推荐排序、转码优先级调度、异常检测等场景。6.3 基于帧级特征的视频去重与检索视频去重是很多内容平台的核心需求。有了帧级特征向量之后可以通过局部敏感哈希对向量做近似近邻搜索找到内容相似的视频集合。不需要逐帧比对视频像素级特征而是比对向量之间的距离计算效率高得多。我把帧级特征向量存到了向量检索服务中平时做离线去重分析时直接批量读取Spark分析结果两个系统的配合很顺畅。6.4 视频内容标签体系的持续迭代最后一次提醒视频数据的价值是动态的标签体系一定需要持续迭代。我刚开始接入的是预训练的图像分类模型标签覆盖比较粗。后续逐步加入自训练的场景分类模型、文字检测模型、声音事件检测模型标签的粒度和准确度明显改善。Spark的好处是数据已经结构化了模型迭代后只需要回填特征向量和标签不需要重新做整条管道。这算是Spark带给我最大的收获——它不只是一次性的批处理工具而是一个可以持续积累、迭代、扩展的数据平台基石。我个人在实际操作中的体会是做视频数据处理最难的地方不是写Spark代码而是理解视频文件本身的特点和分布式计算之间的匹配问题。视频文件不像日志那样可以随意切割它的解码强依赖于完整文件但它的元数据和特征结果又非常适合分布式聚合。想清楚这个边界才有可能把Spark用对地方。另外监控和可观测性一定要从第一天就做好我是吃了好几次任务静默失败的亏之后才补上的日志和指标系统。如果你正准备做类似的事情建议先用小规模数据把这个流程验证一遍再上全量省下的调试时间远比预想的多。