如何查询Athena历史数据结合实时数据调用SageMaker推理做预测

整个链路核心要保证线上线下特征逻辑完全一致，毕竟离线训练已经验证了生命体征变化趋势的重要性，一旦线上取数、特征计算逻辑和训练时有偏差，预测效果会直接跳水。

前置校验（上线前必须完成） #

• 确认Athena体征表已按patient_id、vital_record_time设置分区，单患者查询不要扫全表，否则查询延迟会高到无法满足实时推理要求，分区粒度按数据量选小时/天级即可
• 对齐离线训练的所有规则：包括历史数据取数窗口长度、需要的体征字段列表、缺失值处理逻辑、特征计算方式、数值归一化参数，所有逻辑直接复用离线训练的代码，不要在线上单独重写一套
• 给推理服务配置的IAM角色只开必要权限：对应Athena表的查询权限、Athena查询结果所在S3路径的读权限、目标SageMaker推理端点的调用权限，不要配多余权限

Athena指定患者历史体征查询实现 #

用参数化预编译SQL查询，避免SQL注入同时降低查询延迟，参考SQL如下：

-- 预编译语句提前在Athena创建，线上请求直接传参执行
SELECT vital_record_time, heart_rate, systolic_bp, diastolic_bp, blood_oxygen, body_temperature
FROM patient_vitals_athena_table
WHERE patient_id = ?
  AND vital_record_time BETWEEN DATE_ADD('hour', -?, ?) AND ?
ORDER BY vital_record_time ASC

SQL里的4个占位符按顺序传入：目标患者ID、离线训练对齐的历史窗口小时数、当前实时体征的上报时间、当前实时体征的上报时间即可。
查询侧两个优化点必须做：

• 对高频就诊/重症患者，提前把最近窗口的历史体征预热缓存到Redis，实时请求过来优先读缓存，缓存未命中时再查Athena回源，能把端到端延迟压到百毫秒级
• 给Athena查询设置2秒超时，超时后直接走降级逻辑，不要无限等待拖垮整个推理服务

特征拼接与SageMaker推理调用 #

• 拿到历史体征序列后，直接复用离线的特征工程代码计算趋势特征：包括窗口内各体征的均值、极值、变化斜率、波动标准差等，如果用的是端到端时序模型，直接按模型要求的序列长度截断/补全历史序列即可
• 把计算好的趋势特征、当前上报的实时体征值，按模型训练时的输入字段顺序组装成请求体，字段类型、数值范围必须和训练时完全一致
• 推理调用参考逻辑（boto3实现，客户端初始化一次即可，不要每次请求新建连接）：

import boto3
import json

# 初始化客户端，服务启动时执行一次
sm_runtime = boto3.client('sagemaker-runtime', region_name='ap-east-1')
athena_client = boto3.client('athena', region_name='ap-east-1')

def run_patient_prediction(patient_id: str, current_vital_data: dict, report_timestamp: str):
    # 1. 优先读缓存，缓存miss则执行预编译Athena查询获取历史体征
    history_vitals = get_history_vitals(patient_id, report_timestamp)
    # 2. 复用离线特征逻辑生成输入特征
    model_input = build_model_input(history_vitals, current_vital_data)
    # 3. 调用SageMaker端点
    resp = sm_runtime.invoke_endpoint(
        EndpointName='patient-health-pred-endpoint',
        ContentType='application/json',
        Body=json.dumps(model_input)
    )
    return json.loads(resp['Body'].read().decode())

上线后注意事项 #

• 每次推理的输入数据、返回结果异步落盘到S3，后续做模型迭代、数据漂移检测可以直接复用
• 不要在实时推理链路里做重计算操作，比如历史数据重采样、异常值批量过滤，这类逻辑提前在数仓侧或者缓存预热阶段完成
• 定期抽线上请求的特征和离线同时段计算的特征做比对，一旦出现偏差及时修正，避免线上线下特征不一致导致预测效果下跌

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