1. 为什么把机器学习模型塞进 Serverless API 是个务实选择

我带过六七个从零起步的数据科学团队，几乎每个团队都经历过这样的尴尬时刻：模型在 Jupyter Notebook 里跑得飞起，AUC 0.92，特征重要性图漂亮得能当屏保；可一到业务方问“这个模型怎么嵌进我们订单系统里”，整个团队就陷入沉默。不是模型不行，是交付形态卡住了——它还躺在本地环境里，像一把没装进枪套的刀，再锋利也派不上用场。Azure Functions 就是那个现成的、轻量级的、按需启停的枪套。它不让你操心服务器配置、负载均衡、自动扩缩容这些事，你只管把训练好的模型逻辑写清楚，HTTP 请求一来，它就拉起一个干净的运行时，跑完立刻释放资源。这不是什么高大上的架构玄学，而是我在给三家零售客户做 RFM 客户分群项目时反复验证过的路径：用 Azure Functions 包一层 K-means 模型，API 响应时间稳定在 800ms 内，月度计算成本比维持一台常驻的 t3.medium EC2 实例低 67%。关键词 Azure 在这里不是云厂商的广告位，而是指代一套被充分验证的、开箱即用的基础设施抽象层——它把数据库连接、日志埋点、密钥管理、HTTPS 终止这些琐碎但致命的环节，全打包成 os.environ.get("DB_CONNECTION_STRING") 这样一行代码就能调用的服务。你不需要成为 Azure 专家，但得明白：Serverless 的核心价值不在“无服务器”这个噱头，而在于把数据科学家最不擅长的工程化负担，替换成他们最熟悉的 Python 函数签名 def main(req: func.HttpRequest) -> func.HttpResponse 。这背后是认知负荷的转移：从“怎么让服务器不死”变成“怎么让函数返回正确的 JSON”。对中小团队而言，这省下的不是几小时运维时间，而是避免了因环境不一致导致的线上模型预测漂移——我亲眼见过一个团队因为测试环境用的是 pandas 1.3.5，生产环境误装了 1.5.0， pd.read_sql 的 parse_dates 行为差异，让整个月度客户分群结果错乱了 12%。

2. 核心设计思路与方案选型背后的硬逻辑

2.1 为什么选 Azure Functions 而非其他 Serverless 平台

很多人看到“Serverless”第一反应是 AWS Lambda，但在我实际落地的 11 个生产级 ML API 项目中，Azure Functions 的胜出不是偶然。关键在三个实操痛点的解决上： 冷启动可控性、依赖包管理成熟度、以及与企业现有数据栈的咬合度 。先说冷启动——Lambda 在首次调用时可能卡顿 3-5 秒，这对实时推荐类 API 是灾难。Azure Functions 提供了“预热实例”（Always On）选项，虽然会增加基础费用，但实测下来，将冷启动从平均 2.8 秒压到 0.3 秒以内，且这个功能在 Portal 界面里勾选即可，不用改一行代码。再看依赖包：K-means 看似简单，但 scikit-learn 依赖 numpy 、 scipy ，而 scipy 编译又牵扯到 BLAS 库版本。AWS Lambda 的自定义运行时需要你手动打包 .so 文件，稍有不慎就报 ImportError: libopenblas.so.0: cannot open shared object file 。Azure Functions 则直接支持 requirements.txt ，它会在部署时自动解析依赖树，用 pip install --target ./bin 安装到函数目录，连 wheel 编译失败的兜底方案（如 --find-links https://download.pytorch.org/whl/torch_stable.html ）都预留了配置入口。最后是数据栈咬合：客户用的是 Azure SQL Database 或 Synapse Analytics，Azure Functions 的托管身份（Managed Identity）能直接授权访问，完全绕过密码明文存储的风险。我试过用 Lambda 访问同一家客户的 Azure SQL，最终不得不妥协于在 Secrets Manager 里存凭证，而 Secrets Manager 的轮换策略又和 Azure AD 的权限生命周期不同步，导致凌晨三点收到告警邮件。所以选 Azure Functions，不是因为它是微软亲儿子，而是因为它把数据科学家最怕的“环境地狱”问题，用平台能力封进了黑盒。

2.2 为什么坚持模块化设计而非单文件巨兽

原文中把 get_data.py 、 preprocess.py 、 train.py 、 app.py 拆成四个文件，有人觉得多此一举。但我在重构一个金融风控模型 API 时，就栽在这上面：最初所有逻辑堆在 __init__.py 里，2000 行代码，每次改一个特征工程逻辑，都要重新部署整个函数，CI/CD 流水线跑满 8 分钟。后来拆成模块后， preprocess.py 单元测试可以独立运行，用 pytest tests/test_preprocess.py 12 秒内完成，错误定位精准到 ValueError: Tenure column contains negative values 。模块化不是为了炫技，而是为了 隔离变更影响域 。比如客户突然要求把 MonetaryValue 的 log 变换改成 np.log1p （处理零值），你只需要改 preprocess.py 里的两行代码， train.py 和 get_data.py 完全不动。更关键的是调试效率：本地测试时，我可以单独运行 python -m preprocess test_data.csv ，把原始 CSV 文件喂进去，直接在终端看到 Tenure 列是否正确生成，而不是在 HTTP 响应里扒拉 JSON 字段。这种“可插拔”的设计，让每个模块都能成为独立的质量门禁—— get_data.py 的单元测试必须覆盖数据库连接超时、空结果集、字段缺失三种异常； preprocess.py 的测试必须校验 Tenure 计算精度（毫秒级时间差是否四舍五入到天）； train.py 的测试则要断言 kmeans_pipe.predict() 返回的标签数严格等于 n_clusters=4 。当所有模块的单元测试覆盖率超过 85%，整个函数的稳定性才真正有了根基。否则，所谓“Serverless”只是把单点故障从服务器转移到了函数实例上。

2.3 为什么用 HTTP 触发器而非 Blob 或 Queue 触发器

原文选择了 HTTP Trigger，这看似最直白的选择，但背后有明确的业务语义。RFM 分群不是后台批处理任务，而是 按需触发的决策支持服务 。市场部同事在 Dashboard 上点“刷新客户分群”，前端发一个 GET 请求到 /api/CusCluster ；CRM 系统在新客户注册后，调用 POST 接口传入 {"customer_id": "CUST-789"} 获取其初始分群标签。这种“请求-响应”模式天然匹配 HTTP。如果换成 Blob Trigger（监听 Azure Storage 中新上传的 CSV），就会引入不必要的异步复杂度：谁负责上传文件？上传失败如何重试？文件格式错误怎么通知业务方？我见过一个团队用 Blob Trigger 做模型更新，结果因为上传脚本没加 --fail-fast 参数，一个损坏的 CSV 把整个分群流水线卡死三天。HTTP Trigger 的优势在于 端到端可观测性 ：Azure Monitor 里能直接看到每个请求的耗时、状态码、入参（开启 Application Insights 后），甚至能下钻到 get_data.py 里 pd.read_sql 的执行时间。当某次请求耗时飙升到 5 秒，我打开 Log Analytics，输入 traces | where message contains "Connection success" ，立刻发现是数据库连接池耗尽——因为 get_connection() 每次都新建引擎，没复用。这种问题在异步触发器里会被层层掩盖。所以，别被“Serverless 支持多种触发器”的宣传迷惑，选触发器的本质是选 业务契约 ：HTTP = “我现在就要结果”，Blob = “等我准备好数据再处理”，你的模型服务到底属于哪一种？

3. 核心细节解析与实操要点

3.1 数据库连接的安全实践：告别明文密码

原文中 app.py 直接从 db.cfg 读取密码，这在本地开发没问题，但一旦部署到 Azure，就是严重安全隐患。Azure Functions 提供了两种更安全的方案，我强烈推荐后者：

方案一：应用设置（App Settings）+ 环境变量（推荐）
在 Azure Portal 的 Function App 设置里，添加应用设置：

DB_USER = "prod_reader"
DB_PASSWORD = "@@SecureString@@"
DB_HOST = "my-sql-server.database.windows.net"
DB_PORT = "1433"
DB_NAME = "retail_db"

注意 DB_PASSWORD 的值不要填真实密码，而是用 Azure Key Vault 的机密 URI，格式为 @Microsoft.KeyVault(SecretUri=https://mykeyvault.vault.azure.net/secrets/db-password/xxxxx) 。然后在代码里这样用：

import os
from sqlalchemy import create_engine

def get_connection():
    user = os.environ.get("DB_USER")
    password = os.environ.get("DB_PASSWORD")  # Azure 自动解密
    host = os.environ.get("DB_HOST")
    port = os.environ.get("DB_PORT")
    database = os.environ.get("DB_NAME")
    return create_engine(f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}")

这样做的好处是：密码永远不会出现在代码仓库或部署包里，Key Vault 的访问权限可以精细控制到具体函数应用，且密码轮换只需在 Key Vault 里操作，函数代码零修改。

方案二：托管身份（Managed Identity）——终极方案
如果数据库是 Azure SQL，直接禁用密码认证，启用 Azure AD 集成。在 Function App 的“标识”设置里开启系统分配的托管身份，然后在 Azure SQL 中执行：

CREATE USER [my-function-app] FROM EXTERNAL PROVIDER;
ALTER ROLE db_datareader ADD MEMBER [my-function-app];

连接字符串变成：

from azure.identity import DefaultAzureCredential
from sqlalchemy import create_engine

def get_connection():
    credential = DefaultAzureCredential()
    token = credential.get_token("https://database.windows.net/.default").token
    return create_engine(
        f"mssql+pyodbc:///?odbc_connect="
        f"DRIVER={{ODBC Driver 17 for SQL Server}};"
        f"SERVER={host};DATABASE={database};"
        f"Authentication=ActiveDirectoryAccessToken;"
        f"AccessToken={token}"
    )

这种方式连密码概念都不存在，彻底杜绝泄露风险。我在一个医疗客户项目中强制推行此方案，审计时直接通过了 HIPAA 合规检查。

3.2 特征工程中的时间陷阱： datetime.now() 的坑

原文 preprocess.py 里这行代码很危险：

data['Tenure'] = (datetime.now() - data['Customer_Activation_date']).dt.days

问题在于 datetime.now() 返回的是函数执行时的本地时间，而 Azure Functions 的运行时区域（Region）可能和你的业务时区不一致。比如函数部署在 East US ，但客户数据是 Asia/Shanghai 时区， datetime.now() 返回的是美国东部时间，计算出的 Tenure 会少 13 小时，导致刚激活的客户 Tenure 为 0 天，而实际上已过 13 小时。正确做法是显式指定时区：

from datetime import datetime
import pytz

def pre_process(data):
    # 明确指定业务时区
    tz = pytz.timezone("Asia/Shanghai")
    now = datetime.now(tz)
    # 确保 Customer_Activation_date 也是同一时区
    if data['Customer_Activation_date'].dt.tz is None:
        data['Customer_Activation_date'] = data['Customer_Activation_date'].dt.tz_localize(tz)
    else:
        data['Customer_Activation_date'] = data['Customer_Activation_date'].dt.tz_convert(tz)
    data['Tenure'] = (now - data['Customer_Activation_date']).dt.days
    # ... 其余逻辑

更进一步， Tenure 应该用 pd.Timedelta 计算，避免跨日历月份的误差。比如 1 月 31 日到 2 月 28 日，简单减法可能算成 28 天，但实际是 28 天（闰年）或 27 天（平年）。用 pd.Timedelta 能保证精度：

data['Tenure'] = (now - data['Customer_Activation_date']).dt.total_seconds() // 86400

3.3 模型训练的确定性保障： optimal_init 的持久化

原文把 optimal_init 数组硬编码在 app.py 里，这在快速原型阶段可行，但生产环境必须升级。K-means 的 init 参数决定了聚类中心的初始位置，直接影响最终结果。如果每次部署都用新生成的随机种子，客户分群结果会漂移，业务方无法建立稳定预期。我的做法是：

1. 训练时保存最优 init ：在离线训练脚本中，用 KMeans(n_init=100).fit(X) 找出 inertia_ 最小的那次初始化，并将 kmeans.cluster_centers_ 保存为 optimal_init.npy 。
2. 部署时加载 ：在 train.py 开头加入：

import numpy as np

def load_optimal_init():
    try:
        # 优先从 Azure Blob Storage 加载
        from azure.storage.blob import BlobServiceClient
        blob_service = BlobServiceClient.from_connection_string(
            os.environ["STORAGE_CONNECTION_STRING"]
        )
        blob_client = blob_service.get_blob_client(
            container="ml-models", blob="kmeans/optimal_init.npy"
        )
        init_bytes = blob_client.download_blob().readall()
        return np.load(io.BytesIO(init_bytes))
    except:
        # 降级到本地文件
        return np.load("optimal_init.npy")

3. 版本控制 ： optimal_init.npy 文件名带上时间戳，如 optimal_init_20230801.npy ，并在应用设置里配置 OPTIMAL_INIT_VERSION = "20230801" 。这样模型迭代时，旧版本 API 仍可用旧 init，新版本 API 用新 init，实现灰度发布。

4. 实操过程与核心环节实现

4.1 本地开发环境搭建：VS Code + Azure Functions Extension 的避坑指南

安装 Azure Functions Extension 后，创建新函数的流程看似简单，但有三个极易踩的坑：

坑一：Python 解释器选择
VS Code 提示选择 Interpreter 时，千万别选系统自带的 /usr/bin/python3 。Azure Functions 运行时要求 Python 3.8 或 3.9（截至 2023 年），而 macOS 自带的是 3.9，Ubuntu 22.04 是 3.10，都会报错 Runtime version '3.10' is not supported 。正确做法是用 pyenv 安装指定版本：

pyenv install 3.9.16
pyenv local 3.9.16

然后在 VS Code 的 Interpreter 选择里，找到 ~/.pyenv/versions/3.9.16/bin/python 。这样创建的虚拟环境才是兼容的。

坑二： func host start 启动失败
常见报错 Could not find a version that satisfies the requirement azure-functions ，这是因为 VS Code 创建的项目默认用 pip 安装依赖，但 Azure Functions Core Tools 要求 azure-functions 必须是 >=4.0.0,<5.0.0 。解决方案是在项目根目录创建 requirements-dev.txt ：

azure-functions==4.12.0
azure-storage-blob==12.18.0
pymysql==1.0.2

然后执行：

pip install -r requirements-dev.txt
func host start

坑三：本地测试数据库连接
get_data.py 里的 pd.read_sql 在本地跑不通，因为 db.cfg 里的 HOST 是内网地址。我的做法是：在 app.py 顶部加一个开关：

import os
IS_LOCAL = os.environ.get("IS_LOCAL", "false").lower() == "true"

if IS_LOCAL:
    # 本地用 SQLite 模拟
    import sqlite3
    def get_connection():
        conn = sqlite3.connect("test_rfm.db")
        # 创建测试表并插入模拟数据
        conn.execute("""
            CREATE TABLE IF NOT EXISTS RFM_table (
                Customer_id TEXT,
                Recency INTEGER,
                Frequency INTEGER,
                MonetaryValue REAL,
                Customer_Activation_date TEXT
            )
        """)
        return conn
else:
    # 生产用 Azure SQL
    def get_connection():
        # 原有逻辑

然后在终端启动时：

export IS_LOCAL=true
func host start

这样本地开发完全脱离生产数据库，测试速度提升 10 倍。

4.2 函数代码改造：从脚本到 HTTP API 的三步手术

将 app.py 改造成 Azure Function 的 __init__.py ，不是简单替换函数签名，而是三步深度重构：

第一步：HTTP 请求解析与参数校验
原文 main() 函数没有输入，但 HTTP API 必须处理请求体。我增加了对 GET/POST 的兼容：

import json
import logging
import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('CusCluster function processed a request.')
    
    # 解析请求参数
    customer_id = req.params.get('customer_id')  # GET 参数
    if not customer_id:
        try:
            req_body = req.get_json()
            customer_id = req_body.get('customer_id')  # POST JSON
        except ValueError:
            pass
    
    # 强制校验
    if not customer_id or not isinstance(customer_id, str):
        return func.HttpResponse(
            json.dumps({"error": "Missing or invalid customer_id"}),
            status_code=400,
            mimetype="application/json"
        )
    
    # ... 后续逻辑

第二步：模型输出的健壮序列化
原文 RFM.to_json() 会把 datetime 类型转成 ISO 字符串，但 cluster_labels 是 int64，JSON 不认。必须显式转换：

# 在 train.py 的 train() 函数末尾
def train(data, optimal_init):
    # ... 原有训练逻辑
    data["cluster_labels"] = kmeans_pipe.predict(data)
    # 关键：确保所有列都是 JSON 友好类型
    for col in data.columns:
        if data[col].dtype == 'int64':
            data[col] = data[col].astype(int)
        elif data[col].dtype == 'float64':
            data[col] = data[col].astype(float)
        elif data[col].dtype == 'datetime64[ns]':
            data[col] = data[col].dt.strftime('%Y-%m-%d %H:%M:%S')
    return data

# 在 __init__.py 里
resp = RFM.to_json(orient='records', date_format='iso', date_unit='s')
return func.HttpResponse(
    resp,
    status_code=200,
    mimetype="application/json"
)

第三步：错误处理的分级响应
不能让任何异常穿透到 HTTP 层。我建立了三级错误捕获：

try:
    RFM = get_data(engine)
except Exception as e:
    logging.error(f"Database error: {str(e)}")
    return func.HttpResponse(
        json.dumps({"error": "Database connection failed"}),
        status_code=503,
        mimetype="application/json"
    )

try:
    RFM = pre_process(RFM)
except ValueError as e:
    logging.error(f"Preprocessing error: {str(e)}")
    return func.HttpResponse(
        json.dumps({"error": "Invalid data format"}),
        status_code=400,
        mimetype="application/json"
    )

try:
    RFM = train(RFM, optimal_init)
except Exception as e:
    logging.error(f"Training error: {str(e)}")
    return func.HttpResponse(
        json.dumps({"error": "Model training failed"}),
        status_code=500,
        mimetype="application/json"
    )

这样业务方能根据状态码精准判断问题根源，而不是收到一个笼统的 500。

4.3 部署全流程：从 func deploy 到生产监控

部署不是点一下“Deploy”按钮就完事，而是包含五个关键动作：

动作一：生成精准的 requirements.txt
在激活的虚拟环境中，不要用 pip freeze > requirements.txt ，它会把 azure-functions-core-tools 这类开发依赖也写进去。正确命令是：

pipreqs . --force --ignore venv,tests

pipreqs 会静态分析代码中的 import ，只列出真正用到的包，且自动加上版本号，如 pandas==1.5.3 。

动作二：配置部署时的环境变量
在 local.settings.json 里写的配置，不会自动同步到 Azure。必须在 Portal 的 Function App → “配置” → “应用程序设置”里手动添加。特别注意：

• WEBSITE_RUN_FROM_PACKAGE = 1 ：启用从 ZIP 包运行，提升启动速度。
• PYTHONPATH = /home/site/wwwroot ：确保模块导入路径正确。
• SCM_DO_BUILD_DURING_DEPLOYMENT = true ：让 Kudu 构建引擎在部署时自动运行 pip install 。

动作三：部署命令的黄金组合
VS Code 的图形化部署有时会失败，我坚持用 CLI：

# 登录 Azure
az login

# 设置订阅
az account set --subscription "My Production Subscription"

# 部署（指定资源组、函数应用名、存储账户）
func azure functionapp publish my-function-app \
  --resource-group my-rg \
  --storage-account mystorage \
  --build-native-deps

--build-native-deps 参数至关重要，它会在 Linux 容器里编译 numpy 、 scipy 的 C 扩展，避免运行时报 ImportError: libgfortran.so.5: cannot open shared object file 。

动作四：部署后验证
部署成功不等于 API 可用。必须立即验证：

1. 在 Portal 的 Function App → “函数”里，点击你的函数，看“代码 + 测试”页签是否显示 Function app is running 。
2. 点击“获取函数 URL”，复制链接，在 Postman 里发 GET 请求，检查响应体是否为有效 JSON。
3. 查看“监视” → “日志流”，确认没有 ModuleNotFoundError 或 ConnectionRefusedError 。

动作五：启用 Application Insights
在 Function App → “监视” → “Application Insights”里启用。然后在代码里加一行：

import logging
import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    # 这行会自动上报到 App Insights
    logging.info(f"Processing customer_id: {customer_id}")
    # ... 其余逻辑

在 App Insights 的 Logs 里，可以查：

requests 
| where timestamp > ago(1h) 
| project timestamp, name, resultCode, duration, customDimensions 
| order by timestamp desc

一眼看出哪个请求慢、哪个失败。

5. 常见问题与排查技巧实录

5.1 冷启动耗时过长：从 3.2 秒降到 0.4 秒的实操记录

现象 ：首次调用 API，响应时间 3200ms，后续调用降至 800ms。业务方投诉“第一次用太慢”。

排查路径 ：

1. 在 Application Insights 的 dependencies 表里查：

   dependencies 
   | where type == "SQL" and timestamp > ago(1h) 
   | project timestamp, name, duration, resultCode 
   | order by duration desc
   

   发现 get_data.py 的 pd.read_sql 耗时 2100ms，远超正常值（应 < 300ms）。

2. 检查数据库连接字符串，发现 ?connect_timeout=30 缺失，默认是 15 秒，但网络抖动时会重试。

3. 在 get_connection() 里显式加超时：

   from sqlalchemy import create_engine
   engine = create_engine(
       f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}",
       connect_args={"connect_timeout": 5}
   )
   

4. 更关键的是，发现 get_data.py 每次都新建连接，没复用。改为连接池：

   from sqlalchemy import create_engine, text
   from sqlalchemy.pool import QueuePool
   
   engine = create_engine(
       f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}",
       poolclass=QueuePool,
       pool_size=5,
       max_overflow=10,
       pool_timeout=30,
       pool_recycle=3600
   )
   

效果 ：冷启动耗时从 3200ms 降至 420ms，其中数据库连接从 2100ms 降至 180ms。根本原因是连接池复用了 TCP 连接，避免了三次握手开销。

5.2 模型预测结果不一致： n_init=1 的隐藏陷阱

现象 ：同一 customer_id ，连续两次调用 API，返回的 cluster_labels 不同（如第一次是 2，第二次是 3）。

排查路径 ：

1. 在 train.py 的 train() 函数里加日志：

   logging.info(f"optimal_init shape: {optimal_init.shape}")
   logging.info(f"optimal_init first row: {optimal_init[0]}")
   

   发现日志里 optimal_init 的值每次都不一样。

2. 检查 app.py ，发现 optimal_init 是在 if __name__ == '__main__': 块里定义的，但 Azure Functions 的 main() 函数每次调用都会重新导入模块，导致 optimal_init 被重复初始化。

3. 正确做法是把 optimal_init 提到模块顶层，且用 @lru_cache 缓存：

   from functools import lru_cache
   import numpy as np
   
   @lru_cache(maxsize=1)
   def get_optimal_init():
       return np.array([[0.53110261, -1.65631567, -0.46662182, -0.36120566],
                        [0.36156456, 0.35716547, -0.50222171, -0.47586815],
                        [-0.09470791, 0.41834683, 1.39116768, 1.17794181],
                        [-1.66140645, 0.42296787, -0.10251636, 0.02357708]])
   

效果 ： cluster_labels 100% 一致。 @lru_cache 确保 get_optimal_init() 只执行一次，后续调用直接返回缓存值。

5.3 依赖包冲突： scipy 版本引发的 ImportError

现象 ：部署后函数报错 ImportError: libgfortran.so.5: cannot open shared object file 。

排查路径 ：

1. 在本地用 func azure functionapp publish 部署时，加 --build-native-deps 参数，但依然失败。

2. 登录到 Azure Function 的 Kudu 控制台（ https://<function-app-name>.scm.azurewebsites.net/DebugConsole ），进入 site/wwwroot 目录，执行：

   ls -la /home/site/wwwroot/.python_packages/lib/site-packages/scipy/
   

   发现 scipy 目录下没有 libgfortran.so.5 。

3. 查 requirements.txt ，发现 scipy==1.10.1 ，而 Azure Functions 的 Python 3.9 运行时预装的是 scipy==1.9.3 ，版本不兼容。

解决方案 ：

• 方案 A（推荐）：降级 scipy 到 1.9.3 ，在 requirements.txt 里写死：

  scipy==1.9.3
  numpy==1.23.5
  

• 方案 B：用 --no-cache-dir 强制重新编译：

  func azure functionapp publish my-function-app --no-cache-dir
  

效果 ：方案 A 成功率 100%，部署时间缩短 2 分钟，因为跳过了 scipy 的编译步骤。

5.4 生产环境监控：用 Log Analytics 定制告警

需求 ：当 API 错误率超过 5% 时，自动发邮件给值班工程师。

实操步骤 ：

1. 在 Azure Monitor → “日志”里，创建新查询：

   requests
   | where timestamp > ago(5m)
   | where resultCode startswith "5"
   | summarize failCount = count(), totalCount = count() by bin(timestamp, 1m)
   | extend errorRate = (failCount * 100.0) / totalCount
   | where errorRate > 5
   

2. 点击“新建警报规则”，设置：

   • 条件：当查询结果 > 0 时触发
   • 严重性：严重
   • 操作组：选择已配置的邮件通知组

3. 在函数代码里，主动打点关键指标：

   import logging
   from opencensus.ext.azure.log_exporter import AzureLogHandler
   
   logger = logging.getLogger(__name__)
   logger.addHandler(AzureLogHandler(
       connection_string='InstrumentationKey=xxxxx'
   ))
   
   def main(req: func.HttpRequest) -> func.HttpResponse:
       start_time = time.time()
       # ... 业务逻辑
       duration_ms = (time.time() - start_time) * 1000
       logger.info(f"Cluster prediction completed", extra={
           'custom_dimensions': {
               'duration_ms': duration_ms,
               'customer_id': customer_id
           }
       })
   

效果 ：告警延迟 < 2 分钟，比传统基于日志文件的监控快 10 倍。工程师手机收到邮件时，问题还在发生中，能立刻介入。

6. 模型评估与持续迭代：让 Serverless API 活起来

6.1 在线评估：把 Assess.py 改造成健康检查端点

原文的 Assess.py 是离线脚本，但生产环境需要实时健康度反馈。我把它改造成一个 /api/health 端点：

# __init__.py 里新增 health 函数
def health(req: func.HttpRequest) -> func.HttpResponse:
    try:
        # 1. 检查数据库连通性
        engine = get_connection()
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        
        # 2. 检查模型加载
        optimal_init = get_optimal_init()
        assert optimal_init.shape == (4, 4)
        
        # 3. 抽样评估（用最近 100 条数据）
        RFM = get_data(engine).tail(100)
        RFM = pre_process(RFM)
        RFM = train(RFM, optimal_init)
        
        # 计算轮廓系数（Silhouette Score）
        from sklearn.metrics import silhouette_score
        score = silhouette_score(
            RFM[['Recency', 'Frequency', 'MonetaryValue', 'Tenure']],
            RFM['cluster_labels']
        )
        
        return func.HttpResponse(
            json.dumps({
                "status": "healthy",
                "silhouette_score": round(score, 3),
                "sample_size": len(RFM)
            }),
            status_code=200,
            mimetype="application/json"
        )
    except Exception as e:
        return func.HttpResponse(
            json.dumps({"status": "unhealthy", "error": str(e)}),
            status_code=503,
            mimetype="application/json"
        )

这个端点被 Prometheus 抓取，当 silhouette_score < 0.3 时，触发告警，提示模型可能需要重新训练。

6.2 自动化再训练：用 Timer Trigger 触发周期性更新

HTTP Trigger 是按需的，但模型需要定期更新。我添加了一个 Timer Trigger 函数：

import datetime
import logging
import azure.functions as func
from train import train
from get_data import get_data
from preprocess import pre_process
from app import get_connection, get_optimal_init

def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)
    
    try:
        engine = get_connection()
        RFM = get_data(engine)
        RFM = pre_process(RFM)
        # 用新数据重新训练，但保持 init 不变
        RFM = train(RFM, get_optimal_init())
        
        # 将新模型保存回 Blob Storage
        from azure.storage.blob import BlobServiceClient
        import io
        import joblib
        
        model_bytes = io.BytesIO()
        joblib.dump(RFM, model_bytes)
        model_bytes.seek(0)
        
        blob_service = BlobServiceClient.from_connection_string(
            os.environ["STORAGE_CONNECTION_STRING"]
        )
        blob_client = blob_service.get_blob_client(
            container="ml-models", blob="kmeans/latest.pkl"
        )
        blob_client.upload_blob(model_bytes, overwrite=True)
        
        logging.info("Model retrained and saved successfully")
    except Exception as e:
        logging.error(f"Retraining failed: {e}")

在 function.json 里配置 cron 表达式 "0 0 2 * * *"