使用ffmpeg pyspark和hadoop逐帧处理视频

Question

我想使用Spark和hadoop并行处理mp4视频帧。我不想在处理之前提取所有帧。我正在寻找一种按视频时间顺序读取帧数据的方法，然后在帧使用 yarn 在 Hadoop 集群上触发执行器时馈送帧。mp4 视频文件可以位于本地文件系统或 HDFS 上。

我可以使用 ffmpeg 创建一个管道，然后读取原始帧字节(例如，image = np.fromstring(pipe.stdout.read(1920*1080*3(， dtype='uint8'((。有没有办法将数据(即流，因为帧作为可变解码时间的函数进入(到火花RDD并具有一个地图函数来做某事，例如，计算平均强度？

我已经阅读 Spark 文档很长时间了，但找不到在这种情况下有效的任何内容。我可能只为树木而错过森林。如果可以的话，请提供帮助，即使它涉及不使用ffmpeg和管道。

Answer 1

经过多次试验和错误，我有一个可行的解决方案。虽然这可能并不适合所有人，但它可能会帮助一些人，所以这里是：

我首先创建了一个从视频中提取帧的脚本，此脚本必须存在于所有工作节点上：

#!/home/hadoop/anaconda2/bin/python
import os
import sys
import subprocess as sp
import numpy as np
import cv2
import copy
# RDD.pipe sends via stdin
i = 0
try:
        i = input()
except:
    sys.exit()

file_name = 'v.mp4'
FFMPEG_BIN = "ffmpeg" # on Linux ans Mac OS
command = [ FFMPEG_BIN,
               '-i', '/home/hadoop/' + file_name,
               '-f', 'image2pipe',
               '-vf', 'select=gte(n, %d)' % i,
               '-vframes', '1',
               '-pix_fmt', 'rgb24',
               '-loglevel', 'panic',
               '-vcodec', 'png', '-']
pipe = sp.Popen(command, stdout=sp.PIPE, bufsize=10**8)
data = pipe.stdout.read()
pipe.stdout.close()
import base64
print(base64.b64encode(data))

然后，在 pyspark 脚本中，我使用脚本参数创建一个 RDD：

params = [str(i)  for i in range(1, 1001)]
rdd1 = sc.parallelize(params, numSlices=1000)
pipeRDD = rdd1.pipe('/home/hadoop/src/extract_frame.sh')
resizedRDD = pipeRDD.map(resizeMapper)
test = resizedRDD.collect()

测试现在有前 1000 帧。调整大小映射器调整每个帧的大小，这里是：

def resizeMapper(x):
    import base64
    import cv2
    a = base64.b64decode(x)
    im = cv2.imdecode(np.fromstring(a, dtype=np.uint8), 1)
    im = cv2.resize(im, (200, 200))
    return im

我希望这对那里的人有所帮助。