用华为云 FunctionGraph + OBS 打造 Serverless 图片处理服务

用华为云 FunctionGraph + OBS 打造 Serverless 图片处理服务

一、场景背景

很多业务系统都有图片处理需求：用户上传一张原图，系统需要生成缩略图、加水印、转格式。传统做法是在 ECS 上跑一个常驻的图片处理服务，不管有没有请求都在烧钱。

Serverless 的思路是：把图片处理逻辑写成函数，上传到 OBS 时自动触发函数执行，处理完把结果写回另一个桶。没有请求时零成本，有请求时按调用次数计费，特别适合流量波动大的场景。

华为云 FunctionGraph（函数工作流）+ OBS 事件触发正好能实现这套架构，下面从头搭建一遍。

二、架构设计

用户上传原图 → OBS 源桶（image-input）
                      ↓ 触发事件
              FunctionGraph 函数（图片处理）
                      ↓
              OBS 目标桶（image-output）← 存放缩略图

核心组件：

• OBS 源桶：接收用户上传的原始图片
• OBS 目标桶：存放处理后的缩略图
• FunctionGraph 函数：Python 运行时，使用 Pillow 库处理图片
• OBS 触发器：源桶有新对象上传时自动调用函数

三、前置准备

1. 已开通华为云 OBS 和 FunctionGraph 服务
2. 安装了 OBSutil 工具（用于上传文件测试）
3. 本地有 Python 3.9+ 环境，用于编写和调试函数代码
4. 确认当前区域支持 FunctionGraph（大部分主流区域已支持）

四、创建 OBS 桶

在控制台或用 OBSutil 创建两个桶：

• image-input：源桶，存放用户上传的原图，权限设为私有
• image-output：目标桶，存放缩略图，权限可设为公共读（方便预览）

桶的存储类别选"标准存储"，分区选"多AZ"提高可靠性。

五、编写函数代码

函数的核心逻辑：从事件中获取上传的图片信息，从源桶下载图片，用 Pillow 生成缩略图，上传到目标桶。

import json
import os
import io
from PIL import Image
from obs import ObsClient

# 从环境变量读取配置
AK = os.environ.get('OBS_AK')
SK = os.environ.get('OBS_SK')
SERVER = os.environ.get('OBS_ENDPOINT')
SRC_BUCKET = os.environ.get('SRC_BUCKET', 'image-input')
DST_BUCKET = os.environ.get('DST_BUCKET', 'image-output')
THUMB_WIDTH = int(os.environ.get('THUMB_WIDTH', '300'))

# 初始化 OBS 客户端（复用连接，避免每次调用都新建）
obs_client = ObsClient(access_key_id=AK, secret_access_key=SK, server=SERVER)

def handler(event, context):
    """
    OBS 触发器入口函数
    event 包含上传文件的桶名和对象键
    """
    # 解析事件，获取上传的文件信息
    records = event.get('records', [])
    results = []

    for record in records:
        bucket_name = record.get('bucket', {}).get('name')
        object_key = record.get('object', {}).get('key')

        if not bucket_name or not object_key:
            continue

        # 跳过非图片文件
        if not is_image(object_key):
            results.append({'key': object_key, 'status': 'skipped'})
            continue

        try:
            # 1. 从源桶下载图片
            resp = obs_client.getObject(bucket_name, object_key)
            if resp.status < 300:
                image_data = resp.body.read()
            else:
                results.append({'key': object_key, 'status': 'download_failed'})
                continue

            # 2. 生成缩略图
            thumb_data = generate_thumbnail(image_data, THUMB_WIDTH)

            # 3. 上传到目标桶
            thumb_key = 'thumb_' + object_key
            upload_resp = obs_client.putObject(DST_BUCKET, thumb_key, content=thumb_data)

            if upload_resp.status < 300:
                results.append({
                    'key': object_key,
                    'thumb_key': thumb_key,
                    'status': 'success'
                })
            else:
                results.append({'key': object_key, 'status': 'upload_failed'})

        except Exception as e:
            context.logger.error(f"处理 {object_key} 失败: {str(e)}")
            results.append({'key': object_key, 'status': 'error', 'msg': str(e)})

    return {
        'statusCode': 200,
        'body': json.dumps(results, ensure_ascii=False)
    }

def is_image(filename):
    """判断文件是否为图片"""
    ext = os.path.splitext(filename)[1].lower()
    return ext in ('.jpg', '.jpeg', '.png', '.bmp', '.webp')

def generate_thumbnail(image_data, target_width):
    """
    生成缩略图：按宽度等比缩放，输出 JPEG 格式
    """
    img = Image.open(io.BytesIO(image_data))

    # 计算等比缩放后的高度
    w, h = img.size
    ratio = target_width / w
    target_height = int(h * ratio)

    # 缩放并转 RGB（JPEG 不支持 RGBA）
    thumb = img.resize((target_width, target_height), Image.LANCZOS)
    if thumb.mode in ('RGBA', 'P'):
        thumb = thumb.convert('RGB')

    # 输出到内存
    output = io.BytesIO()
    thumb.save(output, format='JPEG', quality=85)
    return output.getvalue()

六、部署函数

在 FunctionGraph 控制台创建函数：

1. 函数名称：image-processor
2. 运行时：Python 3.9
3. 执行超时：30 秒（图片处理一般几秒内完成）
4. 内存：512 MB（Pillow 处理大图需要足够内存）
5. 代码上传：将上述代码打包为 zip（包含 Pillow 依赖），通过控制台上传

配置环境变量：

七、配置 OBS 触发器

在函数详情页添加触发器：

• 触发类型：OBS
• 桶名：image-input
• 事件类型：ObjectCreated（对象创建时触发）
• 前缀过滤：可留空，或设为 uploads/ 只处理特定目录下的图片

配置完成后，每次有图片上传到 image-input 桶，FunctionGraph 就会自动调用 image-processor 函数。

八、验证测试

用 OBSutil 上传一张测试图片：

obsutil cp ./test-photo.jpg obs://image-input/test-photo.jpg

上传完成后，检查 image-output 桶中是否出现了 thumb_test-photo.jpg：

obsutil ls obs://image-output/

如果看到缩略图文件，说明整条链路跑通了。在函数日志页面可以查看每次调用的执行时长和输出。

九、实战踩坑经验

1. Pillow 依赖要一起打包

FunctionGraph 的 Python 运行时预装了标准库，但 Pillow 不在其中。需要在本地 pip install Pillow -t ./package 安装到指定目录，然后和函数代码一起打 zip 上传。漏了这步会报 No module named 'PIL'。

2. 大图处理要关注内存和超时

一张 10MB 的原图，Pillow 解压到内存可能占用 100MB+。如果函数内存配 256MB 直接 OOM。建议根据最大处理图片尺寸合理设置内存，超时时间也别设太短，给网络下载留余量。

3. OBS 客户端要全局初始化

把 ObsClient 的初始化放在函数外部（模块级变量），不要放在 handler 里面。FunctionGraph 的执行容器会复用，全局初始化的客户端在后续调用中直接可用，省去重复建连接的开销。冷启动时初始化一次，热调用直接复用，响应延迟能降一半。

4. 事件格式要确认好

OBS 触发器传入的 event 结构和 API 网关触发器不同。OBS 事件是 records 数组，每个 record 包含 bucket.name 和 object.key。第一次调试时建议先把 json.dumps(event) 打到日志里看实际结构，别凭猜的写解析逻辑。

5. 幂等性要考虑

OBS 触发器理论上至少投递一次，极端情况下可能重复调用。如果处理逻辑有副作用（比如写数据库记录），要加幂等判断：处理前先检查目标桶是否已存在同名缩略图，存在就跳过。

十、成本估算

这套架构的运行成本非常低：

• OBS 存储：按实际存储量计费，1GB 标准存储约 0.099 元/月
• FunctionGraph 调用：每月前 100 万次免费，之后 0.0115 元/千次
• OBS 请求：PUT 请求 0.0033 元/千次，GET 请求 0.001 元/千次

假设每月 1 万张图片，总成本不到 5 元。对比一台 24 小时运行的 ECS（至少几十元/月），Serverless 方案在低流量场景下优势明显。

十一、总结

这篇文章演示了用华为云 FunctionGraph + OBS 搭建一个 Serverless 图片处理服务的完整流程：用户上传原图到 OBS 源桶，自动触发函数生成缩略图并写入目标桶，全程无需管理服务器。

Serverless 的价值在于"按需付费、自动伸缩、零运维"。对于图片处理、日志清洗、定时任务这类事件驱动的场景，FunctionGraph 是非常合适的选择。如果你的业务流量波动大、有明显的空闲时段，不妨算一算从 ECS 迁移到 Serverless 能省多少。

下一步可以扩展这个服务：加上图片水印、支持多种尺寸输出、接入 CDN 加速缩略图分发，逐步把它变成一个完整的图片处理平台。