使用Polars读取SQLite数据库表时规避SchemaError及相关技术疑问

遇到几百列的表抛出SchemaError却不说是哪列出问题，确实太闹心了！我来一步步帮你拆解这些问题，给你实用的解决办法：

1. 能否让Polars直接使用SQLite类型的合理对应类型？ #

当然可以！而且不用手动写死schema_overrides（毕竟你列不固定），我们可以动态获取SQLite表的列类型，自动生成类型映射：

SQLite的PRAGMA table_info(table_name)能返回表的所有列定义，我们可以用这个结果自动生成schema_overrides，把SQLite的INTEGER/REAL/TEXT分别对应Polars的pl.Int64/pl.Float64/pl.String。具体代码如下：

import polars as pl
import sqlite3

conn = sqlite3.connect('my_database.db')

# 第一步：获取表的列元数据
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(table_to_load)")
col_metadata = cursor.fetchall()  # 格式：(cid, 列名, 类型, notnull, 默认值, 是否主键)

# 第二步：动态生成类型映射
type_map = {
    'INTEGER': pl.Int64,
    'REAL': pl.Float64,
    'TEXT': pl.String
}
schema_overrides = {}
for col in col_metadata:
    col_name = col[1]
    col_type = col[2].upper()  # 统一转大写避免大小写问题
    if col_type in type_map:
        schema_overrides[col_name] = type_map[col_type]

# 第三步：用自动生成的schema_overrides读取数据
df = pl.read_database(
    connection=conn,
    query='SELECT * FROM table_to_load',
    schema_overrides=schema_overrides,
    infer_schema_length=None
)
conn.close()

这种方法完全适配列不固定的场景，不用每次改代码。

2. 为什么infer_schema_length=None还是报类型错误？怎么解决？ #

infer_schema_length=None是让Polars读取所有行来推断类型，但问题出在你的列里实际混合了两种不兼容的类型——比如某个定义为INTEGER的列里，SQLite实际存储了i64（整数）和binary（二进制数据）两种值。Polars找不到这两种类型的共同超类型（总不能把整数转成二进制或者反过来吧），所以直接报错。

除了上面的schema_overrides方法，还有另一种思路：让SQLite先把所有值强制转成列定义的类型，再返回给Polars。我们可以动态生成带CAST的SQL查询：

# 基于之前的col_metadata生成带CAST的查询语句
select_clause = ", ".join(
    f"CAST({col[1]} AS {col[2]}) AS {col[1]}" 
    for col in col_metadata
)
query = f"SELECT {select_clause} FROM table_to_load"

# 直接用这个查询读取，Polars就能拿到统一类型的数据
df = pl.read_database(connection=conn, query=query, infer_schema_length=None)

这样SQLite会先把列里的所有值转成定义的类型，Polars读取时就不会遇到混合类型的问题了。

3. 怎么让Polars提示出问题的具体列？ #

不得不说，Polars当前的这个错误提示确实不够友好，默认不会告诉你哪列出了问题。不过我们可以自己写个小脚本快速排查：

def find_mixed_type_cols(conn, table_name, sample_rows=1000):
    cursor = conn.cursor()
    cursor.execute(f"SELECT * FROM {table_name} LIMIT {sample_rows}")
    rows = cursor.fetchall()
    if not rows:
        return []
    
    col_names = [desc[0] for desc in cursor.description]
    problematic_cols = []
    
    # 逐列检查值的类型
    for col_idx, col_name in enumerate(col_names):
        seen_types = set()
        for row in rows:
            val_type = type(row[col_idx])
            seen_types.add(val_type.__name__)
            if len(seen_types) > 1:
                problematic_cols.append(f"列 {col_name} 包含混合类型：{', '.join(seen_types)}")
                break
    return problematic_cols

# 调用方法排查
bad_cols = find_mixed_type_cols(conn, 'table_to_load')
for col in bad_cols:
    print(col)

这个脚本会抽样检查前1000行（可以调整sample_rows），找出所有包含混合类型的列，帮你快速定位问题根源。

内容来源于stack exchange