文档首页/ 数据工坊 DWR/ 最佳实践/ 抽帧截图（自定义算子）
更新时间：2026-04-03 GMT+08:00
查看PDF
分享

抽帧截图（自定义算子）

方案概述

需要对视频抽帧截图时，在不编写额外代码的情况下使用抽帧截图算子对视频进行指定时间点截帧。

方案架构

方案优势

用户自定义算子，无需依赖额外服务，功能更灵活。

约束与限制

该抽帧截图算子目前暂不支持中文对象。

资源和成本规划

表1 资源和成本规划

资源

资源说明

数量

每月费用

OBS

算子请求OBS API。

1

通过算子对数据进行处理，都会涉及到对OBS API的调用，每调用一次API都计算一次请求次数。对象存储服务OBS会根据调用API的请求次数进行费用收取，收取详情参见OBS请求费用说明。

FunctionGraph函数

算子使用FunctionGraph函数工作流。

1

通过算子对数据进行处理，可能会使用到函数工作流的资源，比如算子执行时长，函数工作流会根据资源使用情况进行收费，收费详情参见函数工作流计费说明。

操作流程

您首先需要在FunctionGraph导入自定义算子需要的函数依赖包。然后需要在FunctionGraph创建自定义抽帧截图函数。之后需要在DWR工作流编排界面完成配置动态参数，创建同步工作流。最后触发工作流进行验证。

图1 操作流程
  1. 在FunctionGraph控制台，导入函数依赖包。
  2. 在FunctionGraph控制台，创建抽帧截图函数。
  3. 配置相关参数。
  4. 在DWR控制台，创建同步工作流。
  5. 触发工作流执行。

实施步骤

  1. 创建依赖包。

    1. 登录FunctionGraph控制台。
    2. 在左侧列表选择“函数 > 依赖包管理”。
    3. 在依赖包管理页面，单击“创建依赖包”。
    4. 填写相关参数，完成依赖包创建。
      图2 创建依赖包

  2. 在“函数列表”页面单击“创建函数”完成函数创建。

    其中委托需要创建能够授权工作流FunctionGraph访问OBS的权限，创建方式参考创建委托。

  3. 函数创建成功后进入函数详情页添加依赖包。

    1. 在“代码依赖包”模块单击“添加依赖包”。
    2. 选择“私有依赖包”，添加1中创建好的依赖包。

  4. 依赖添加成功后，在FunctionGraph控制台添加算子代码并保存部署。参考示例如下： 

    # -*- coding:utf-8 -*-
import json
# 引入模块
from obs import ObsClient
import os
import time
import urllib.parse
 
 
def handler(event, context):
    print(context)
    print(event)
    obsClient = ObsClient(
        access_key_id=context.getAccessKey(),
        secret_access_key=context.getSecretKey(),
        server='https://obs.cn-north-4.myhuaweicloud.com' #OBS的endpoint，不同region的OBS Endpoint不同
    )
    bucketName = event['Records'][0]['obs']['bucket']['name']
    objectKey = urllib.parse.unquote(event['Records'][0]['obs']['object']['key'])  # a/b/c
    frame = event["dynamic_source"]["f"] #截取的秒数
    height = event["dynamic_source"]["h"]
    width = event["dynamic_source"]["w"]
    # 使用访问OBS
    start = time.time()
    resp = obsClient.getObject(bucketName, objectKey, "/tmp/" + objectKey)
    end = time.time() - start
    # print(resp)
    print("download test.jpg run :" + str(end - start))
    start = time.time()
    cmd = "/opt/function/code/ffmpeg -i /tmp/" + objectKey + " -ss {f} -vframes 1 -vf scale={w}:{h} -y /tmp/{f}.jpg".format(
        f=frame, w=width, h=height)
    print(cmd)
    os.system(cmd)
    dsObjectKey = "{}.jpg".format(frame)
    start = time.time()
    out_obj = "output/{obj}/{file}".format(obj=objectKey, file=dsObjectKey)
    resp = obsClient.putFile(bucketName, out_obj,
                             "/tmp/" + dsObjectKey)
    end = time.time() - start
    print(" upload run :" + str(end - start))
    # 关闭obsClient
    obsClient.close()
    tasks = [
        {
            "output": {
                "bucket": bucketName,
                "object": out_obj,
                "location": "cn-north-4"
            }
        }
    ]
    event["dynamic_source"] = {
        "tasks": tasks
    }
    return event

  5. 函数保存成功后，进入数据工坊DWR控制台，进入工作流界面，单击“创建工作流”。
  6. 在工作流编排界面选择自定义算子并完成编排。其中基本属性部分的函数唯一标识选择4创建好的自定义函数。

    加入动态参数f（截图秒数frame），h（输出图片高度height），w（输出图片宽度width）。

    图3 自定义算子设置

  7. 单击“保存”，保存工作流。可选择同步并支持匿名访问。
  8. 保存成功后，通过url方式触发工作流。其中可以指定参数f（截图秒数frame）,h（输出图片高度）获得截帧结果。

    https://test-workflow-bj4.obs.cn-north-4.myhuaweicloud.com/VCG42N1291956568.mp4?x-workflow-graph-name=vframe/f_0,h_100

相关文档