如何将MySQL存储过程中更新语句的执行结果保存至表或外部文件

• 步骤1：创建ETL执行日志表，用来存储每条语句的执行结果

CREATE TABLE etl_execution_log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    exec_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '执行时间',
    exec_sql TEXT COMMENT '执行的SQL语句',
    rows_matched INT COMMENT '匹配行数',
    rows_changed INT COMMENT '实际变更行数',
    warnings_cnt INT COMMENT '警告数量'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

注意：如果你的ETL任务包含事务回滚逻辑，又需要保留所有执行记录，可将上表引擎改为MyISAM，该引擎不支持事务，日志写入不会随主事务回滚而丢失。

• 步骤2：修改现有存储过程，每条UPDATE执行后立刻捕获状态并写入日志表
  示例存储过程代码如下：

DELIMITER //
CREATE PROCEDURE your_etl_procedure()
BEGIN
    -- 声明存储执行结果的变量
    DECLARE v_matched INT;
    DECLARE v_changed INT;
    DECLARE v_warn INT;
    
    -- 执行UPDATE语句
    UPDATE table1 SET column1=3 WHERE column2 = 4;
    -- 捕获匹配行数
    SHOW SESSION STATUS LIKE 'Rows_matched' INTO @dummy, v_matched;
    -- 捕获变更行数
    SHOW SESSION STATUS LIKE 'Rows_changed' INTO @dummy, v_changed;
    -- 捕获警告数量
    SHOW SESSION STATUS LIKE 'Warnings' INTO @dummy, v_warn;
    -- 写入日志表
    INSERT INTO etl_execution_log(exec_sql, rows_matched, rows_changed, warnings_cnt)
    VALUES ('UPDATE table1 SET column1=3 WHERE column2 = 4', v_matched, v_changed, v_warn);
    
    -- 其余UPDATE语句均按照上述逻辑，执行后紧跟捕获+写入操作即可
END //
DELIMITER ;

• 步骤3（可选）：将日志导出到外部文件
  如需将日志保存到本地文件，可执行如下SQL导出：

SELECT * FROM etl_execution_log 
WHERE DATE(exec_time) = CURDATE()
INTO OUTFILE '/var/lib/mysql-files/etl_daily_log.csv'
FIELDS TERMINATED BY ',' ENCLOSED BY '"'
LINES TERMINATED BY '\n';

需确保MySQL配置项secure_file_priv允许写入对应目录。

如果不想改动已上线的存储过程代码，可通过Python逐条执行SQL并捕获结果，直接写入本地文件或数据库。

• 首先安装依赖库：
  pip install mysql-connector-python
• 示例代码：

import mysql.connector
import datetime
from mysql.connector import Error

# 数据库连接配置
db_config = {
    "host": "你的数据库IP",
    "user": "数据库账号",
    "password": "数据库密码",
    "database": "业务库名",
    "client_flags": [mysql.connector.ClientFlag.FOUND_ROWS]
}

# 待执行的UPDATE语句列表，可从现有存储过程中提取
update_sqls = [
    "UPDATE table1 SET column1=3 WHERE column2 = 4",
    # 其余UPDATE语句依次添加
]

if __name__ == "__main__":
    conn = None
    cursor = None
    try:
        conn = mysql.connector.connect(**db_config)
        if conn.is_connected():
            cursor = conn.cursor()
            for sql in update_sqls:
                # 执行SQL
                cursor.execute(sql)
                # 获取变更行数
                changed = cursor.rowcount
                # 获取匹配行数
                cursor.execute("SHOW SESSION STATUS LIKE 'Rows_matched'")
                matched = int(cursor.fetchone()[1])
                # 获取警告数
                cursor.execute("SHOW SESSION STATUS LIKE 'Warnings'")
                warn_cnt = int(cursor.fetchone()[1])
                # 写入本地日志文件
                log_content = f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] \n执行SQL：{sql}\nRows matched: {matched} Changed: {changed} Warnings: {warn_cnt}\n---\n"
                with open("etl_execution_log.txt", "a", encoding="utf-8") as f:
                    f.write(log_content)
                # 提交事务
                conn.commit()
    except Error as e:
        print(f"执行出错：{e}")
        if conn:
            conn.rollback()
    finally:
        if cursor:
            cursor.close()
        if conn and conn.is_connected():
            conn.close()

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