从Amazon Redshift取数据触发UnicodeDecodeError，求助可行方案

针对你遇到的stl_querytext.text字段因无效UTF-8字节导致的UnicodeDecodeError，以下是几个可行的解决思路：

1. 在Redshift端将文本转为原始字节，Python端手动处理解码 #

修改SQL查询，把qt.text转为bytea类型获取原始字节数据，绕过驱动自动解码的限制，在Python中手动指定错误处理策略：

SELECT 
    q.query AS query_id,
    qt.text::bytea AS query_text
FROM 
    pg_catalog.stl_query AS q
INNER JOIN
    pg_catalog.stl_querytext AS qt
    ON q.query = qt.query
WHERE TRUE
    AND DATE(q.starttime) = '2023-10-24'
    AND EXTRACT('hour' FROM q.starttime) = 19

对应的Python代码处理：

from sqlalchemy import create_engine

_QUERY = """
SELECT 
    q.query AS query_id,
    qt.text::bytea AS query_text
FROM 
    pg_catalog.stl_query AS q
INNER JOIN
    pg_catalog.stl_querytext AS qt
    ON q.query = qt.query
WHERE TRUE
    AND DATE(q.starttime) = '2023-10-24'
    AND EXTRACT('hour' FROM q.starttime) = 19
"""

uri = "postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}".format(**_secrets)
engine = create_engine(uri)

with engine.connect() as connection:
    result = connection.execute(_QUERY)
    for row in result:
        query_id = row["query_id"]
        # 手动解码，忽略无效字符或替换为占位符
        query_text = row["query_text"].decode("utf-8", errors="ignore")
        # 或者用errors="replace"把无效字符替换为�
        # query_text = row["query_text"].decode("utf-8", errors="replace")
        print(query_id, query_text)

2. 在Redshift端过滤/修复无效UTF-8字符 #

利用Redshift内置函数提前处理无效字符：

• 过滤无效行：使用validate_utf8函数检查文本是否为有效UTF-8，只保留合规行：

SELECT 
    q.query AS query_id,
    qt.text
FROM 
    pg_catalog.stl_query AS q
INNER JOIN
    pg_catalog.stl_querytext AS qt
    ON q.query = qt.query
WHERE TRUE
    AND DATE(q.starttime) = '2023-10-24'
    AND EXTRACT('hour' FROM q.starttime) = 19
    AND validate_utf8(qt.text) IS NOT NULL

• 修复无效字符：通过编码转换丢弃无效字节，将文本先转成latin1再转回utf-8（会丢失部分特殊字符，但能保留大部分内容）：

SELECT 
    q.query AS query_id,
    convert_from(convert_to(qt.text, 'latin1'), 'utf8') AS text
FROM 
    pg_catalog.stl_query AS q
INNER JOIN
    pg_catalog.stl_querytext AS qt
    ON q.query = qt.query
WHERE TRUE
    AND DATE(q.starttime) = '2023-10-24'
    AND EXTRACT('hour' FROM q.starttime) = 19

3. 调整psycopg2客户端编码为latin1，Python端二次转码 #

在连接URL中指定client_encoding=latin1，先以latin1解码所有数据，再在Python中转成utf-8并处理错误：

uri = "postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}?client_encoding=latin1".format(**_secrets)
engine = create_engine(uri)

with engine.connect() as connection:
    result = connection.execute(_QUERY)
    for row in result:
        query_id = row["query_id"]
        # 先按latin1解码，再转成utf-8并忽略错误
        query_text = row["text"].encode("latin1").decode("utf-8", errors="ignore")
        print(query_id, query_text)

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