华为云 FunctionGraph 函数工作流：冷启动实测与适用场景

有个定时任务的需求——每天拉一次第三方 API 的数据，清洗后写数据库。原本跑在 ECS 的 cron 里，但那台机器大部分时间是空闲的，只为每天跑一分钟的任务付费不划算。试了下 FunctionGraph。

先说结论

适合：定时任务、事件驱动、低频调用、突发流量。

不适合：常驻服务、有状态应用、对延迟极敏感的场景（冷启动问题）。

函数怎么写

FunctionGraph 支持多种运行时，Python、Node.js、Java、Go 都有。我用 Python，入口函数约定：

import json

def handler(event, context):
    # event 是触发事件，定时触发器时是时间信息
    # context 里有 request_id、函数名等

    print("start", context.getRequestID())

    result = {
        "statusCode": 200,
        "body": json.dumps({"ok": True})
    }
    return result

依赖放在同目录，打包成 zip 上传。Python 的第三方包要在本地装好再打包，注意平台和 Python 版本要一致。

打包命令：

pip install -r requirements.txt -t ./package
cp handler.py ./package/
cd package && zip -r ../function.zip .

定时触发器

配置界面里可以加定时触发器，用 cron 表达式。

0 0 2 * * ?     # 每天 02:00
0 0/30 * * * ?  # 每 30 分钟

注意 FunctionGraph 用的是六位 cron（秒 分 时 日 月 周），最后一位是周不是年。第一次用七位写法会报错。

冷启动实测

这是最关心的点。测了三组：

Python 运行时冷启动在 1-3 秒区间，取决于包大小和初始化逻辑。

减少冷启动的几个办法：

1. 精简依赖包

我们的包一开始 45MB（因为装了 pandas），冷启动 4 秒。去掉 pandas 换成标准库的 csv 处理，包降到 3MB，冷启动 1.2 秒。

pandas、numpy 这类大包对冷启动影响极大。

2. 把初始化代码放函数外

import pymysql

# 模块级初始化，冷启动时执行一次，热启动复用
conn = None

def get_conn():
    global conn
    if conn is None or not conn.open:
        conn = pymysql.connect(...)
    return conn

def handler(event, context):
    db = get_conn()
    ...

但注意：数据库连接可能因为超时被服务端断开，要处理重连。

3. 预热

配一个定时触发器，每 5 分钟调一次空函数，保持实例热着。但这会产生额外调用次数（虽然费用很低）。

对延迟敏感的场景，预热是必要成本。

一个完整的定时任务示例

需求：每天凌晨 2 点拉数据，清洗后写入 RDS。

import os
import json
import urllib.request
import pymysql

def handler(event, context):
    api_url = os.environ.get("API_URL")
    db_conf = {
        "host": os.environ.get("DB_HOST"),
        "user": os.environ.get("DB_USER"),
        "password": os.environ.get("DB_PASSWORD"),
        "database": os.environ.get("DB_NAME"),
    }

    # 1. 拉数据
    try:
        with urllib.request.urlopen(api_url, timeout=30) as resp:
            raw = json.loads(resp.read())
    except Exception as e:
        print(f"fetch failed: {e}")
        return {"statusCode": 500, "body": str(e)}

    # 2. 清洗
    rows = []
    for item in raw.get("data", []):
        if not item.get("id"):
            continue
        rows.append((
            item["id"],
            item.get("name", "")[:100],
            item.get("value", 0),
        ))

    if not rows:
        print("no valid rows")
        return {"statusCode": 200, "body": "empty"}

    # 3. 写库
    conn = pymysql.connect(**db_conf, connect_timeout=10)
    try:
        with conn.cursor() as cur:
            cur.executemany(
                "INSERT INTO stats (id, name, value) VALUES (%s, %s, %s) "
                "ON DUPLICATE KEY UPDATE name=VALUES(name), value=VALUES(value)",
                rows
            )
        conn.commit()
    finally:
        conn.close()

    print(f"inserted {len(rows)} rows")
    return {"statusCode": 200, "body": f"ok {len(rows)}"}

配置里要设：

• 环境变量（API_URL、DB_*）
• 超时时间（默认 3 秒，一定要改，我们设了 60 秒）
• 内存（默认 128MB，Python 处理 JSON 建议 256MB 起）
• VPC 网络（要访问 RDS 必须配 VPC，否则连不通内网）

VPC 配置是最容易漏的一步。函数默认在公网环境，要访问 RDS 这样的内网服务，必须把函数挂到同一个 VPC 和子网下。

费用怎么算

FunctionGraph 计费三部分：调用次数、执行时长（GB-秒）、公网流量。

免费额度：每月 100 万次调用 + 40 万 GB-秒。

我们的场景：每天 1 次，每次 3 秒，256MB 内存。

时长：3 秒 × 0.25GB × 30 天 = 22.5 GB-秒/月

远低于免费额度，基本白嫖。

对比 ECS：一台最低配 1C2G 每月也要几十块。低频任务用函数明显划算。

踩过的坑

坑一：超时时间太短

默认 3 秒，调用外部 API 经常超时。第一次跑报 Task timed out，改成 60 秒才正常。

坑二：VPC 配了但连不上 RDS

VPC 配了还要确认：

• 子网是否有剩余 IP
• RDS 的安全组是否放行该子网网段
• 函数和 RDS 是否在同一区域

坑三：日志找不到

FunctionGraph 的日志默认写到 LTS（云日志服务）。控制台的「监控」里能看到调用记录，「日志」里能看到 print 输出，但需要先开通 LTS。

坑四：环境变量明文

敏感信息放环境变量里是明文存储的。更安全的做法是用「委托」机制，让函数通过 IAM 临时凭证访问其他服务，不落地长期密钥。

坑五：并发限制

默认单函数并发上限 100。突发流量超过会被限流（返回 429）。可以在控制台申请提升配额，或者配消息队列做缓冲。

什么场景值得用

值得：

• 定时任务（数据同步、报表生成、清理）
• 事件驱动（OBS 上传触发图片处理、消息队列消费）
• Webhook 接收（第三方回调）
• 低频 API

不值得：

• 常驻 Web 服务（冷启动影响体验）
• 长连接（WebSocket）
• 需要本地状态的计算
• 超过 15 分钟的长任务（函数有最大执行时长限制）

小结

FunctionGraph 的核心价值是「不用为闲置付费」。低频任务从 ECS 迁过来，成本能降一个数量级。

代价是要接受冷启动，以及把「常驻进程」的思维改成「无状态函数」。数据库连接、临时文件、内存缓存这些在函数里都要重新设计。

包体积是冷启动的关键，能砍的依赖尽量砍。