Yandex Cloud大文件上传触发Serverless函数的实现方法

整套方案完全基于Yandex托管服务实现，无需搭建任何私有服务器，能力和AWS侧S3预签名上传+Lambda事件触发的方案完全对齐。

服务映射关系 #

和参考的AWS方案对应关系如下，所有能力原生支持，不需要二次开发适配：

• AWS S3 → Yandex Object Storage：完全兼容S3 API，支持分片上传、生命周期管理、事件通知
• AWS Lambda → Yandex Cloud Functions：支持对象存储事件直接触发，覆盖所有主流语言运行时
• AWS API Gateway → Yandex API Gateway：原生对接云函数、对象存储，负责对外提供HTTP API路由

落地步骤 #

1. 基础资源配置 #

先把核心托管资源创建配置好：

• 创建Object Storage存储桶
  • 权限设置为公有读、公有写全禁止，所有上传操作走预签名授权，避免恶意上传产生额外费用
  • 配置生命周期规则：针对上传路径（比如/uploads/前缀下所有对象）设置上传满3天自动删除，自动满足留存周期要求，不需要手动写清理逻辑
  • 配置事件通知：新增「对象创建完成」触发规则，事件类型只勾选分片上传完成合并、普通PUT上传完成两类，触发目标绑定后续创建的后处理云函数，避免分片上传过程中误触发函数
• 创建上传后处理的Cloud Functions
  • 按业务需求选运行时写后处理逻辑即可，函数触发时会自动传入新上传文件的桶名、对象键、文件大小、MD5等元信息，不需要额外传参
  • 给函数绑定的服务账号配置对应存储桶的只读权限，方便函数读取文件内容处理

2. 大文件上传逻辑（支持无上限大文件，适配2GB+场景） #

不要走API网关透传文件的模式，Yandex API Gateway默认请求体上限128MB，就算手动调参也不适合传输GB级大文件，直接用预签名分片直传方案，和AWS S3的大文件上传逻辑完全一致：

• 先写一个轻量的凭证签发云函数，绑定到API Gateway的HTTP路由（比如POST /api/get-upload-token），这个函数只做签名计算，不处理文件流，单实例QPS可以到上千，性能足够：
  • 接口接收前端传入的文件名、文件总大小、文件类型参数
  • 调用Object Storage的S3兼容SDK，初始化分片上传任务，生成每个分片对应的预签名PUT地址、最终分片合并的预签名POST地址，签名有效期设置为1小时即可
  • 把上传ID、分片预签名地址列表、合并地址返回给前端
• 前端侧上传逻辑完全不需要依赖服务端中转：
  1. 本地将大文件按100MB_{200MB的固定大小切分，2GB文件仅需拆成10}20个分片
  2. 并发将每个分片直传到对应的预签名Object Storage地址，单分片上传失败仅需重传对应分片，天然支持断点续传
  3. 所有分片上传完成后，调用合并接口通知Object Storage将所有分片合并为完整文件
• 合并完成后，Object Storage会自动触发之前绑定的后处理云函数执行，不需要额外配置回调、轮询逻辑。

凭证签发函数核心代码示例（Python运行时，基于S3兼容SDK实现）：

import boto3
from botocore.config import Config

# 初始化S3兼容客户端，走内网端点不产生外网流量费用
s3_client = boto3.client(
    's3',
    endpoint_url='https://storage.yandexcloud.net',
    config=Config(signature_version='s3v4'),
    region_name='ru-central1'
)
BUCKET_NAME = "替换成你自己的存储桶名"

def handler(event, context):
    req_params = event['queryStringParameters']
    file_key = f"uploads/{req_params['file_name']}"
    part_count = int(req_params['part_count'])
    
    # 初始化分片上传任务
    mpu_task = s3_client.create_multipart_upload(Bucket=BUCKET_NAME, Key=file_key)
    upload_id = mpu_task['UploadId']
    
    # 生成分片预签名上传地址
    part_presign_urls = []
    for part_num in range(1, part_count + 1):
        url = s3_client.generate_presigned_url(
            'upload_part',
            Params={
                'Bucket': BUCKET_NAME,
                'Key': file_key,
                'UploadId': upload_id,
                'PartNumber': part_num
            },
            ExpiresIn=3600
        )
        part_presign_urls.append(url)
    
    # 生成分片合并预签名地址
    complete_url = s3_client.generate_presigned_url(
        'complete_multipart_upload',
        Params={
            'Bucket': BUCKET_NAME,
            'Key': file_key,
            'UploadId': upload_id
        },
        ExpiresIn=3600
    )

    return {
        'statusCode': 200,
        'headers': {'Access-Control-Allow-Origin': '*'},
        'body': {
            'upload_id': upload_id,
            'part_urls': part_presign_urls,
            'complete_url': complete_url
        }
    }

3. 避坑提示 #

• 存储桶生命周期规则记得勾选「应用到所有对象版本」，如果开了版本控制的话，避免旧版本文件没被自动删除产生额外存储费用
• 预签名凭证的权限要做最小化限制，只允许生成指定上传前缀的签名，不要给签发函数配置整桶的写权限
• 如果需要做上传用户鉴权，直接在凭证签发接口里加用户校验逻辑即可，校验通过再发签名，不需要改动上传链路

内容的提问来源于stack exchange，提问作者DuckQueen