如何每小时调度SQL查询并实现偏差超限告警？

一、每小时自动执行SQL查询 #

根据你使用的数据库和运维环境，推荐以下几种实现方式：

1. 数据库内置定时任务 #

PostgreSQL（借助pg_cron扩展）

先安装扩展，再创建每小时执行的任务：

-- 安装pg_cron（若未安装）
CREATE EXTENSION IF NOT EXISTS pg_cron;

-- 每小时整点执行，将差值结果存入日志表（便于后续判断偏差）
SELECT cron.schedule(
    'hourly-data-compare',
    '0 * * * *', -- 定时规则：每小时整点执行
    $$
    INSERT INTO data_diff_log (diff_value, execute_time, yesterday_count)
    SELECT 
        -- 今日上一小时数据量 - 昨日同时段数据量
        (SELECT count(*) FROM t1 WHERE YEAR=YEAR(current_date) AND MONTH=MONTH(current_date) AND DAY=Day(current_date) AND HOUR=Hour(CURRENT_TIMESTAMP)-1)
        -
        (SELECT count(*) FROM t2 WHERE YEAR=YEAR(current_date - interval '1' day) AND MONTH=MONTH(current_date - interval '1' day) AND DAY=Day(current_date - interval '1' day) AND HOUR=Hour(CURRENT_TIMESTAMP)-1)
        AS total_count,
        CURRENT_TIMESTAMP,
        -- 同时存入昨日数据量，用于计算偏差率
        (SELECT count(*) FROM t2 WHERE YEAR=YEAR(current_date - interval '1' day) AND MONTH=MONTH(current_date - interval '1' day) AND DAY=Day(current_date - interval '1' day) AND HOUR=Hour(CURRENT_TIMESTAMP)-1)
    $$
);

注：这里把Hour(CURRENT_TIMESTAMP)改为Hour(CURRENT_TIMESTAMP)-1，确保取的是上一小时的统计数据，符合你的需求。

MySQL（使用事件调度器）

先开启调度功能，再创建定时事件：

-- 开启事件调度器
SET GLOBAL event_scheduler = ON;

-- 创建每小时执行的事件
CREATE EVENT hourly_data_compare
ON SCHEDULE EVERY 1 HOUR
STARTS CURRENT_TIMESTAMP + INTERVAL 1 HOUR - MINUTE(CURRENT_TIMESTAMP) MINUTE - SECOND(CURRENT_TIMESTAMP) SECOND
DO
BEGIN
    INSERT INTO data_diff_log (diff_value, execute_time, yesterday_count)
    SELECT 
        (SELECT count(*) FROM t1 WHERE YEAR=YEAR(NOW()) AND MONTH=MONTH(NOW()) AND DAY=DAY(NOW()) AND HOUR=HOUR(NOW())-1)
        -
        (SELECT count(*) FROM t2 WHERE YEAR=YEAR(DATE_SUB(NOW(), INTERVAL 1 DAY)) AND MONTH=MONTH(DATE_SUB(NOW(), INTERVAL 1 DAY)) AND DAY=DAY(DATE_SUB(NOW(), INTERVAL 1 DAY)) AND HOUR=HOUR(NOW())-1)
        AS total_count,
        NOW(),
        (SELECT count(*) FROM t2 WHERE YEAR=YEAR(DATE_SUB(NOW(), INTERVAL 1 DAY)) AND MONTH=MONTH(DATE_SUB(NOW(), INTERVAL 1 DAY)) AND DAY=DAY(DATE_SUB(NOW(), INTERVAL 1 DAY)) AND HOUR=HOUR(NOW())-1)
    ;
END;

2. 操作系统级定时任务 #

如果数据库不支持内置定时，用系统原生定时工具：

Linux（Cron）

编辑crontab配置（执行crontab -e），添加以下内容：

0 * * * * psql -U your_username -d your_db -c "INSERT INTO data_diff_log (diff_value, execute_time, yesterday_count) SELECT (SELECT count(*) FROM t1 WHERE YEAR=YEAR(current_date) AND MONTH=MONTH(current_date) AND DAY=Day(current_date) AND HOUR=Hour(CURRENT_TIMESTAMP)-1) - (SELECT count(*) FROM t2 WHERE YEAR=YEAR(current_date - interval '1' day) AND MONTH=MONTH(current_date - interval '1' day) AND DAY=Day(current_date - interval '1' day) AND HOUR=Hour(CURRENT_TIMESTAMP)-1) AS total_count, CURRENT_TIMESTAMP, (SELECT count(*) FROM t2 WHERE YEAR=YEAR(current_date - interval '1' day) AND MONTH=MONTH(current_date - interval '1' day) AND DAY=Day(current_date - interval '1' day) AND HOUR=Hour(CURRENT_TIMESTAMP)-1);" >> /var/log/data_diff.log 2>&1

替换your_username、your_db为实际信息，日志路径可自定义。

二、偏差>0.3时发送告警 #

注：这里的偏差指|差值/昨日同时段数据量| > 0.3，需先计算偏差率再判断。

1. Python脚本实现（通用方案） #

写一个脚本完成查询、偏差判断、告警发送：

import psycopg2  # PostgreSQL用这个，MySQL替换为pymysql
import smtplib
from email.mime.text import MIMEText

def check_and_alert():
    # 数据库连接配置
    conn = psycopg2.connect(
        dbname='your_db',
        user='your_username',
        password='your_password',
        host='your_host'
    )
    cur = conn.cursor()

    # 查询核心数据
    cur.execute("""
        SELECT 
            (SELECT count(*) FROM t2 WHERE YEAR=YEAR(current_date - interval '1' day) AND MONTH=MONTH(current_date - interval '1' day) AND DAY=Day(current_date - interval '1' day) AND HOUR=Hour(CURRENT_TIMESTAMP)-1) AS yesterday_count,
            (SELECT count(*) FROM t1 WHERE YEAR=YEAR(current_date) AND MONTH=MONTH(current_date) AND DAY=Day(current_date) AND HOUR=Hour(CURRENT_TIMESTAMP)-1)
            -
            (SELECT count(*) FROM t2 WHERE YEAR=YEAR(current_date - interval '1' day) AND MONTH=MONTH(current_date - interval '1' day) AND DAY=Day(current_date - interval '1' day) AND HOUR=Hour(CURRENT_TIMESTAMP)-1)
            AS total_diff
    """)
    yesterday_count, total_diff = cur.fetchone()

    # 计算偏差率（避免除以0）
    if yesterday_count == 0:
        deviation = float('inf')
    else:
        deviation = abs(total_diff / yesterday_count)

    # 触发告警
    if deviation > 0.3:
        send_email_alert(deviation, total_diff, yesterday_count)

    cur.close()
    conn.close()

def send_email_alert(deviation, diff, yesterday):
    # 邮件配置
    sender = 'alert@yourdomain.com'
    receivers = ['your_notify_email@example.com']
    subject = '数据偏差告警（超过0.3阈值）'
    body = f"""
    数据偏差已超过设定阈值：
    昨日同时段数据量：{yesterday}
    今日上一小时与昨日差值：{diff}
    偏差率：{round(deviation, 2)}
    """

    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = sender
    msg['To'] = ','.join(receivers)

    # 发送邮件（替换为你的SMTP信息）
    with smtplib.SMTP('smtp.yourdomain.com', 587) as server:
        server.starttls()
        server.login(sender, 'your_email_password')
        server.sendmail(sender, receivers, msg.as_string())

if __name__ == '__main__':
    check_and_alert()

然后把脚本加入系统定时任务，每小时执行：

0 * * * * python3 /path/to/your/alert_script.py >> /var/log/alert_script.log 2>&1

如果需要企业微信/钉钉告警，只需修改send_email_alert函数，调用对应机器人API即可。

2. 数据库触发器+外部监听（进阶方案） #

若依赖数据库日志表，可创建触发器，当插入的记录偏差率超过0.3时，通过pg_notify（PostgreSQL）或UDF（MySQL）触发外部脚本告警，但复杂度较高，推荐用脚本直接处理。

注意事项 #

• 确保HOUR字段存储的是0-23的小时数，时间逻辑需覆盖跨月、跨年场景（比如1月1日的0点，要取去年12月31日的0点数据）。
• 数据库连接信息避免硬编码，可改用环境变量或加密配置文件。
• 定时任务需配置足够的执行权限，避免因权限不足导致任务失败。

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