华为云 FunctionGraph 实战：构建 Serverless 定时数据处理管道

像流水线一样处理数据，却不用管任何一台服务器

前言

在业务系统中，定时数据处理是一个极其常见的场景：每天凌晨清理过期文件、每小时汇总业务指标、定期同步数据库到数据仓库……传统做法是找一台 ECS 跑 Crontab，但这带来了一系列问题——服务器闲时浪费资源、忙时可能扛不住、还要操心补丁、监控、重启。

华为云 FunctionGraph（函数工作流）提供了一种截然不同的思路：写一个函数，配一个定时触发器，剩下的交给平台。没有服务器要管理，没有 Crontab 要维护，函数只在被触发时运行，按执行次数和时长计费。对于间歇性数据处理任务，成本远低于常驻服务器。

本文将完整分享一次实战经验：用 FunctionGraph 构建一个定时数据处理管道，自动扫描 OBS 桶中的日志文件，提取关键指标并生成汇总报告。整个过程不涉及任何服务器运维。

一、场景描述

假设我们有如下业务需求：

• 业务系统每天往 OBS 桶里写入大量访问日志文件（JSON 格式）
• 需要每天凌晨 2 点自动处理前一天的日志
• 提取每个文件的请求总数、错误率、平均响应时间三项指标
• 将汇总结果写入另一个 OBS 桶，供下游报表系统消费
• 如果处理失败，需要记录错误日志并发送通知

整体数据流如下：

OBS 日志桶 → FunctionGraph 函数 → OBS 汇总桶
                  ↓
              错误日志桶

二、前置准备

2.1 创建 OBS 桶

需要三个桶：

• log-bucket：存放原始日志文件（业务系统写入）
• report-bucket：存放每日汇总报告
• error-bucket：存放处理失败的错误日志

在华为云控制台创建这三个桶，注意选择同一个区域（本文以 cn-north-4 为例），桶策略保持默认的私有读写。

2.2 准备日志文件格式

原始日志文件为 JSON 格式，每行一条记录：

{"timestamp": "2026-10-27T10:15:30Z", "method": "GET", "path": "/api/users", "status": 200, "response_time": 45}
{"timestamp": "2026-10-27T10:15:31Z", "method": "POST", "path": "/api/orders", "status": 500, "response_time": 1200}
{"timestamp": "2026-10-27T10:15:32Z", "method": "GET", "path": "/api/products", "status": 200, "response_time": 78}

三、编写处理函数

3.1 函数逻辑设计

函数的核心处理流程：

1. 根据触发时间计算昨天的日期
2. 列出 log-bucket 中以该日期为前缀的所有文件
3. 逐个下载文件，解析 JSON，提取指标
4. 汇总所有文件的指标
5. 生成报告文件，写入 report-bucket
6. 如果中途异常，将错误信息写入 error-bucket

3.2 函数代码

使用 Python 运行时，代码如下：

import json
import os
import datetime
from obs import ObsClient

# 初始化 OBS 客户端（通过环境变量获取 AK/SK）
obs_client = ObsClient(
    access_key_id=os.environ.get('OBS_AK'),
    secret_access_key=os.environ.get('OBS_SK'),
    server='obs.cn-north-4.myhuaweicloud.com'
)

LOG_BUCKET = 'log-bucket'
REPORT_BUCKET = 'report-bucket'
ERROR_BUCKET = 'error-bucket'


def handler(event, context):
    """FunctionGraph 入口函数"""
    try:
        # 计算目标日期：默认处理昨天的数据
        target_date = calculate_target_date(event)
        context.logger.info(f"开始处理 {target_date} 的日志数据")

        # 列出该日期的所有日志文件
        files = list_log_files(target_date)
        context.logger.info(f"找到 {len(files)} 个日志文件")

        if not files:
            context.logger.info("没有需要处理的文件，正常退出")
            return {"status": "success", "message": "no files to process"}

        # 逐文件处理并汇总
        summary = {
            "date": target_date,
            "total_files": len(files),
            "total_requests": 0,
            "total_errors": 0,
            "total_response_time": 0,
            "request_count": 0,
            "file_details": []
        }

        for file_key in files:
            file_result = process_single_file(file_key)
            summary["total_requests"] += file_result["request_count"]
            summary["total_errors"] += file_result["error_count"]
            summary["total_response_time"] += file_result["total_response_time"]
            summary["request_count"] += file_result["request_count"]
            summary["file_details"].append({
                "file": file_key,
                "requests": file_result["request_count"],
                "errors": file_result["error_count"]
            })

        # 计算汇总指标
        summary["error_rate"] = round(
            summary["total_errors"] / summary["total_requests"] * 100, 2
        ) if summary["total_requests"] > 0 else 0
        summary["avg_response_time"] = round(
            summary["total_response_time"] / summary["request_count"], 2
        ) if summary["request_count"] > 0 else 0

        # 写入报告
        report_key = f"reports/{target_date}/daily_summary.json"
        write_report(report_key, summary)
        context.logger.info(f"报告已写入: {report_key}")

        return {
            "status": "success",
            "date": target_date,
            "files_processed": len(files),
            "total_requests": summary["total_requests"],
            "error_rate": summary["error_rate"]
        }

    except Exception as e:
        # 异常处理：记录错误日志
        error_info = {
            "timestamp": datetime.datetime.now().isoformat(),
            "error": str(e),
            "event": event
        }
        error_key = f"errors/{target_date}/error_{int(time.time())}.json"
        obs_client.putObject(ERROR_BUCKET, error_key, json.dumps(error_info))
        context.logger.error(f"处理失败: {str(e)}")
        raise e


def calculate_target_date(event):
    """从触发事件中提取目标日期，默认为昨天"""
    if event and 'target_date' in event:
        return event['target_date']
    yesterday = datetime.date.today() - datetime.timedelta(days=1)
    return yesterday.strftime("%Y-%m-%d")


def list_log_files(target_date):
    """列出指定日期的所有日志文件"""
    prefix = f"logs/{target_date}/"
    resp = obs_client.listObjects(LOG_BUCKET, prefix=prefix)
    if resp.status < 300:
        return [obj.key for obj in resp.body.contents]
    return []


def process_single_file(file_key):
    """处理单个日志文件，提取指标"""
    resp = obs_client.getObject(LOG_BUCKET, file_key)
    if resp.status >= 300:
        raise Exception(f"下载文件失败: {file_key}")

    content = resp.body.response.read().decode('utf-8')
    lines = content.strip().split('\n')

    request_count = 0
    error_count = 0
    total_response_time = 0

    for line in lines:
        try:
            record = json.loads(line)
            request_count += 1
            if record.get("status", 200) >= 400:
                error_count += 1
            total_response_time += record.get("response_time", 0)
        except json.JSONDecodeError:
            continue  # 跳过无法解析的行

    return {
        "request_count": request_count,
        "error_count": error_count,
        "total_response_time": total_response_time
    }


def write_report(key, data):
    """将报告写入 OBS"""
    content = json.dumps(data, ensure_ascii=False, indent=2)
    obs_client.putObject(REPORT_BUCKET, key, content)

3.3 代码要点解析

第一，OBS 客户端在函数外部初始化。 FunctionGraph 的函数实例会被复用（不是每次调用都冷启动），把客户端初始化放在全局作用域，可以利用实例复用减少重复创建的开销。这是一个容易被忽略的性能优化点。

第二，错误处理不吞异常。 在 process_single_file 中，JSON 解析失败时用 continue 跳过单行，但整体处理异常会向上抛出，由外层捕获并写入错误桶。这样既容忍了数据中的少量脏数据，又不会让严重错误被静默忽略。

第三，通过环境变量传递 AK/SK。 切勿将访问密钥硬编码在函数代码中。FunctionGraph 控制台支持为函数配置环境变量，密钥在运行时注入，代码可以安全地提交到代码仓库。

四、部署函数

4.1 创建函数

在华为云控制台进入 FunctionGraph 服务：

1. 点击「创建函数」，选择「Python 3.9」运行时
2. 函数名称填 daily-log-processor
3. 将上面的代码粘贴到代码编辑区
4. 执行超时时间设置为 300 秒（数据处理可能耗时较长，默认 3 秒不够用）
5. 内存设置为 512MB（根据日志文件大小调整，OBS SDK 需要一定内存）

4.2 配置环境变量

在函数的「环境变量」配置页：

4.3 配置 OBS 依赖

FunctionGraph 的 Python 运行时默认不包含 OBS SDK，需要通过依赖包方式引入。有两种方式：

• 方式一：将 obs-python-sdk 打包成 zip 上传到函数的依赖层
• 方式二：在函数代码中使用 FunctionGraph 内置的 OBS 操作 API（通过 context 对象）

推荐方式一，灵活性更高。打包方法：

# 在本地创建打包目录
mkdir -p deps/python
pip install esdk-obs-python -t deps/python

# 打包
cd deps
zip -r obs_sdk_layer.zip python/

然后在控制台创建「依赖包」，上传这个 zip 文件，并在函数配置中关联该依赖包。

五、配置定时触发器

5.1 创建 TIMER 触发器

在函数详情页的「触发器」配置中，添加一个 TIMER 触发器：

• 触发器类型：TIMER
• 定时类型：定时触发（Cron 表达式）
• Cron 表达式：0 0 2 * * ?

这个 Cron 表达式表示每天凌晨 2 点执行一次。FunctionGraph 使用 Quartz Cron 格式，注意秒字段不能省略。

5.2 Cron 表达式速查

5.3 手动测试

配置完成后，可以先手动触发一次测试。在函数详情页点击「测试」，构造一个测试事件：

{
    "target_date": "2026-10-27"
}

指定一个有日志数据的日期，观察执行结果和日志输出。确认报告文件已正确写入 report-bucket。

六、监控与运维

6.1 查看执行日志

FunctionGraph 会自动记录每次函数调用的日志。在控制台的「调用记录」页面可以查看：

• 调用时间、耗时、状态
• 通过 context.logger 输出的自定义日志
• 如果抛出异常，会显示完整的错误堆栈

这些日志保留一段时间后自动清理，不需要自己维护日志存储。

6.2 关键指标关注

日常运维中，重点关注以下指标：

• 调用次数：每天应该固定 1 次（定时触发），如果突然增多说明配置有误
• 执行时长：正常应在数十秒内，如果突然飙升可能是日志文件暴增或 OBS 响应慢
• 错误次数：正常为 0，出现错误时去 error-bucket 查看详细信息

6.3 告警配置

在 FunctionGraph 控制台可以为函数配置告警：

• 错误次数超过阈值时发送通知
• 执行时长超过阈值时发送通知

推荐至少配置「错误次数 > 0」的告警，确保数据处理失败时能第一时间感知。

七、进阶优化

7.1 并行处理

当日志文件数量较多时，单函数串行处理可能超时。优化思路：

将函数拆分为两层——分发函数和处理函数。分发函数列出所有文件后，为每个文件调用一次处理函数（通过 FunctionGraph 的同步调用 API），实现并行处理。最后再调用一个汇总函数合并结果。

这种 Fan-out/Fan-in 模式在 Serverless 架构中非常经典，能充分利用弹性伸缩能力，把串行的 N 个文件处理变成并行，总耗时从 N × 单文件耗时降低到约等于单文件耗时。

7.2 增量处理

如果单次处理的数据量非常大，可以考虑增量处理：每次只处理一小批文件，通过函数间传递游标（类似分页的 offset）来推进进度。FunctionGraph 支持函数自调用，可以在处理完一批后触发下一批。

7.3 成本优化

FunctionGraph 的计费维度是「调用次数 × 执行时长 × 内存」。优化方向：

• 减少调用次数：合并小任务，避免过于频繁的触发
• 降低执行时长：优化代码逻辑，减少不必要的 IO 操作
• 合理设置内存：内存越大执行越快但单价越高，需要找到平衡点。对于 IO 密集型任务，内存设小一些；对于计算密集型任务，内存设大一些能显著缩短执行时间

八、实践总结

这次实战让我对 Serverless 架构有了更深的体会：

优势方面：

1. 零运维：不需要管理服务器、操作系统、运行环境，专注业务逻辑
2. 自动弹性：无论处理 10 个文件还是 1000 个文件，平台自动调度，无需扩容
3. 按需计费：函数不运行时不收费，对于每天只跑一次的任务，成本极低
4. 天然容错：函数执行失败会自动记录，可以配置重试和告警

需要注意的方面：

1. 冷启动：函数首次调用或长时间未调用后会有冷启动延迟（通常数百毫秒到数秒），对定时任务影响不大，但不适合对延迟敏感的在线请求
2. 执行时长限制：FunctionGraph 单次执行最长 15 分钟（可配置），超大数据处理需要拆分
3. 依赖管理：第三方库需要打包上传，不如传统服务器灵活
4. 本地调试：函数运行在云端环境，本地调试需要模拟 context 和 event，有一定开发门槛

选型建议：FunctionGraph 非常适合间歇性、事件驱动、无状态的数据处理场景。如果你的任务是常驻运行、长连接、需要维护复杂状态的，传统 ECS/CCE 可能更合适。技术选型没有银弹，理解每种方案的特点和边界，才能做出正确选择。

希望这篇实战分享对正在探索华为云 Serverless 的同学有所帮助。如果你也在用 FunctionGraph 处理业务数据，欢迎在评论区交流你的实践心得。