如何通过GCP Cloud Function读取BigQuery数据并触发阈值告警邮件

• 给Cloud Function关联的服务账号授予BigQuery数据查看权限（roles/bigquery.dataViewer）以及BigQuery作业用户权限（roles/bigquery.jobUser）
• 环境变量配置：除了现有的EMAIL_API_KEY，还可以新增BQ_PROJECT_ID、THRESHOLD_VALUE、ALERT_RECEIVER这类参数，避免硬编码到代码里后续维护麻烦
• 依赖配置：requirements.txt文件中需要添加以下依赖包：

google-cloud-bigquery==3.12.0
sendgrid==6.10.0
python-http-client==3.3.7

整体逻辑顺序为：执行BigQuery查询→提取结果和阈值对比→超过阈值则调用你已有的发邮件逻辑，整合后的参考代码如下：

import os
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, Email
from python_http_client.exceptions import HTTPError
from google.cloud import bigquery

# 初始化BigQuery客户端，Cloud Function运行时会自动调用关联服务账号鉴权，无需手动配置密钥
bq_client = bigquery.Client()

def threshold_alert(request):
    # 1. 替换为你自己的业务查询语句
    query = """
        SELECT COUNT(*) as today_active_user
        FROM `你的项目ID.你的数据集名.你的表名`
        WHERE DATE(register_time) = CURRENT_DATE()
    """
    # 执行查询
    query_job = bq_client.query(query)
    results = query_job.result()
    
    # 2. 提取查询结果，这里示例取第一行第一个字段值，你可以根据自己的查询结果结构调整
    current_value = list(results)[0][0]
    # 阈值从环境变量读取，调整时无需修改代码
    threshold = int(os.environ.get('THRESHOLD_VALUE', 100))
    
    # 3. 阈值比对，未触发条件直接返回
    if current_value < threshold:
        return f"当前指标值{current_value}未超过阈值{threshold}，无需发送告警"
    
    # 4. 超过阈值时调用邮件发送逻辑
    sg = SendGridAPIClient(os.environ['EMAIL_API_KEY'])
    # 邮件内容动态插入实际数值，可读性更高
    html_content = f"<p>告警通知：监测指标已超过预设阈值！当前值：{current_value}，阈值：{threshold}</p>"
    
    message = Mail(
        to_emails=os.environ.get('ALERT_RECEIVER', "[Destination]@email.com"),
        from_email=Email('[YOUR]@gmail.com', "告警通知账号"),
        subject="BigQuery指标阈值告警",
        html_content=html_content
    )
    message.add_bcc("[YOUR]@gmail.com")
    
    try:
        response = sg.send(message)
        return f"告警邮件已发送，状态码：{response.status_code}，当前指标值：{current_value}"
    except HTTPError as e:
        return f"邮件发送失败：{e.message}"

• 如果你的BigQuery查询返回多个监测指标，需要对应修改结果提取逻辑，逐个和对应阈值做比对
• 可以配合Cloud Scheduler给Cloud Function配置定时触发器，按照业务需要的频率（每小时、每天等）自动执行监测，不需要手动调用
• 如果查询的数据量较大，建议提前优化查询语句，避免函数运行超时，Cloud Function默认超时时间可以手动调整到最长9分钟
• 测试阶段可以先把阈值调至低于当前指标值，验证告警逻辑正常后再切换为实际业务阈值

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