如何将Redshift加载错误提取至Python本地日志文件？

我来帮你搞定这个问题！你现在每次遇到加载错误都得手动查STL_LOAD_ERRORS表，其实可以在Python代码里集成错误查询逻辑，把错误信息提取到变量里，再自动写入本地日志文件，不用再手动查表啦。

实现思路 #

1. 执行完Redshift的COPY命令后，不管命令执行成功还是失败，都查询STL_LOAD_ERRORS表，获取本次加载相关的错误信息
2. 把查询到的错误信息整理成易读的格式，存入变量
3. 将错误信息写入本地日志文件，同时可以根据错误情况决定是否退出程序

具体代码示例 #

下面是完整的代码片段，结合psycopg2和日志模块实现这个功能：

import psycopg2
import logging
from datetime import datetime

# 配置日志记录
logging.basicConfig(
    filename='redshift_load_errors.log',
    level=logging.ERROR,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

def load_data_to_redshift():
    redshift_conn = None
    try:
        # 1. 建立Redshift连接
        redshift_conn = psycopg2.connect(
            dbname='your_db',
            user='your_user',
            password='your_password',
            host='your_redshift_host',
            port='5439'
        )
        cursor = redshift_conn.cursor()

        # 2. 执行COPY命令（这里替换成你的实际COPY语句）
        copy_query = """
            COPY your_target_table
            FROM 's3://your_bucket/your_data_file.csv'
            IAM_ROLE 'arn:aws:iam::123456789012:role/your_redshift_role'
            CSV DELIMITER ',' IGNOREHEADER 1;
        """
        cursor.execute(copy_query)
        redshift_conn.commit()
        print("数据加载成功！")

    except psycopg2.Error as e:
        print(f"加载过程中出现错误: {e}")
        # 3. 查询STL_LOAD_ERRORS获取详细错误信息
        if redshift_conn:
            redshift_conn.rollback()
            error_cursor = redshift_conn.cursor()
            # 用query_id精确匹配本次COPY任务的错误（更精准）
            error_query = """
                SELECT load_time, filename, line_number, error_message, err_code
                FROM stl_load_errors
                WHERE query_id = %s
                ORDER BY load_time DESC;
            """
            error_cursor.execute(error_query, (cursor.query_id,))
            errors = error_cursor.fetchall()

            # 4. 整理错误信息到变量
            error_details = []
            for err in errors:
                err_str = f"加载时间: {err[0]}, 文件: {err[1]}, 行号: {err[2]}, 错误信息: {err[3]}, 错误码: {err[4]}"
                error_details.append(err_str)
                logging.error(err_str)  # 写入日志文件

            # 如果需要，可以把error_details变量用于其他处理（比如发送告警）
            print("详细错误信息已写入日志文件: redshift_load_errors.log")
            # 这里可以选择是否退出程序，比如保留状态1退出
            exit(1)

    finally:
        # 关闭连接
        if redshift_conn:
            redshift_conn.close()

if __name__ == "__main__":
    load_data_to_redshift()

关键细节说明 #

• 精准匹配错误：上面用cursor.query_id来定位本次COPY任务的错误，比时间范围过滤更准确，能确保只获取当前加载产生的错误
• 日志配置：用Python内置的logging模块配置了日志文件，错误信息会自动追加，包含时间戳和完整错误详情，方便后续排查
• 错误处理流程：捕获到psycopg2异常后先回滚事务，再查询错误表，确保能拿到最新的加载错误，最后可以保留你原来的状态1退出逻辑

这样以后再出现加载错误时，你直接看本地的redshift_load_errors.log文件就能看到详细的错误信息，不用再手动登录Redshift查STL_LOAD_ERRORS表啦！

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