如何低内存占用为整个SQLite数据库生成汇总统计信息？

假设有一个包含大量表和列的SQLite数据库database.db。Pandas的.describe()方法能够生成我需要的汇总统计信息（示例代码如下），但该方法需要完整读取每张表，这对于大型数据库来说内存占用过高。是否存在内存占用更低的SQL或Python替代方案？手动指定列名在此场景下不可行。

示例代码：

import pandas as pd
import sqlite3

con = sqlite3.connect("file:database.db", uri=True)
tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table'", con)
columns = []
for _, row in tables.iterrows():
    col = pd.read_sql(f"PRAGMA table_info({row['name']})", con)
    col['table'] = row['name']
    stats = pd.read_sql(f"""SELECT * FROM {row['name']}""", con)
    stats = stats.describe(include='all')
    stats = stats.transpose()
    col = col.merge(stats, left_on='name', right_index=True)
    columns.append(col)
columns = pd.concat(columns)

方案1：SQL端计算统计量（内存最优） #

让SQLite引擎直接在数据库内部计算统计量，仅返回最终结果，完全避免加载整张表到Python内存。核心是根据列的数据类型生成针对性的统计SQL：

import sqlite3
import pandas as pd

def get_table_columns(con, table_name):
    # 获取表的列名和数据类型
    cols = pd.read_sql(f"PRAGMA table_info({table_name})", con)
    cols['table'] = table_name
    return cols

def generate_stats_query(table_name, cols):
    stats_clauses = []
    for _, col in cols.iterrows():
        col_name = col['name']
        col_type = col['type'].upper()
        
        if col_type in ('INTEGER', 'REAL'):
            # 数值列统计：count/mean/std/分位数/max/min
            clauses = [
                f"COUNT({col_name}) AS `{col_name}_count`",
                f"AVG({col_name}) AS `{col_name}_mean`",
                f"CAST(STDDEV({col_name}) AS REAL) AS `{col_name}_std`",
                f"MIN({col_name}) AS `{col_name}_min`",
                f"PERCENTILE_CONT(0.25) WITHIN GROUP (ORDER BY {col_name}) AS `{col_name}_25%`",
                f"PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY {col_name}) AS `{col_name}_50%`",
                f"PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY {col_name}) AS `{col_name}_75%`",
                f"MAX({col_name}) AS `{col_name}_max`"
            ]
        else:
            # 文本/其他类型列统计：count/unique/top值/出现频率
            clauses = [
                f"COUNT({col_name}) AS `{col_name}_count`",
                f"COUNT(DISTINCT {col_name}) AS `{col_name}_unique`",
                f"(SELECT {col_name} FROM {table_name} GROUP BY {col_name} ORDER BY COUNT(*) DESC LIMIT 1) AS `{col_name}_top`",
                f"(SELECT COUNT(*) FROM {table_name} GROUP BY {col_name} ORDER BY COUNT(*) DESC LIMIT 1) AS `{col_name}_freq`"
            ]
        stats_clauses.extend(clauses)
    
    return f"SELECT {', '.join(stats_clauses)} FROM {table_name}"

def unpivot_stats(stats_df, table_name, cols):
    # 将宽格式统计结果转成与原代码一致的长格式
    unpivoted = []
    for _, col in cols.iterrows():
        col_name = col['name']
        row = col.to_dict()
        for stat in ['count', 'mean', 'std', 'min', '25%', '50%', '75%', 'max', 'unique', 'top', 'freq']:
            col_stat = f"{col_name}_{stat}"
            if col_stat in stats_df.columns:
                row[stat] = stats_df[col_stat].iloc[0]
        unpivoted.append(row)
    return pd.DataFrame(unpivoted)

# 主逻辑
con = sqlite3.connect("file:database.db", uri=True)
tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table'", con)
all_columns = []

for _, row in tables.iterrows():
    table_name = row['name']
    cols = get_table_columns(con, table_name)
    stats_query = generate_stats_query(table_name, cols)
    stats_df = pd.read_sql(stats_query, con)
    unpivoted_stats = unpivot_stats(stats_df, table_name, cols)
    all_columns.append(unpivoted_stats)

final_df = pd.concat(all_columns, ignore_index=True)
con.close()

优势 #

• 内存占用极低：仅加载表结构和统计结果，不触碰原始表数据
• 统计结果精确，完全匹配describe()的输出格式
• 自动适配列类型，无需手动指定

方案2：Pandas分块读取（折中方案） #

如果坚持使用Pandas的describe()，可以分块读取表数据，逐步合并统计结果，降低单次内存负载：

import pandas as pd
import sqlite3

con = sqlite3.connect("file:database.db", uri=True)
tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table'", con)
columns = []

for _, row in tables.iterrows():
    table_name = row['name']
    col = pd.read_sql(f"PRAGMA table_info({table_name})", con)
    col['table'] = table_name
    
    # 分块读取，chunksize可根据内存调整
    chunk_iter = pd.read_sql(f"SELECT * FROM {table_name}", con, chunksize=1000)
    combined_stats = None
    for chunk in chunk_iter:
        chunk_stats = chunk.describe(include='all').transpose()
        if combined_stats is None:
            combined_stats = chunk_stats
        else:
            # 合并统计逻辑：count直接累加，其他统计量取最新块结果（如需精确值建议用方案1）
            combined_stats['count'] += chunk_stats['count']
            for stat in ['mean', 'std', 'min', '25%', '50%', '75%', 'max', 'unique', 'top', 'freq']:
                if stat in combined_stats.columns:
                    combined_stats[stat] = chunk_stats[stat]
    
    col = col.merge(combined_stats, left_on='name', right_index=True)
    columns.append(col)

columns = pd.concat(columns)
con.close()

注意 #

• 分块合并的统计量为近似值（尤其是分位数、标准差），精度不如SQL端计算
• 适合内存不足以加载整张表，但需要保留describe()输出格式的场景

方案3：SQLite ANALYZE命令（快速近似统计） #

SQLite的ANALYZE命令会收集表的基础统计信息并存储在系统表中，虽然信息有限，但速度极快、内存占用可忽略：

import sqlite3
import pandas as pd

con = sqlite3.connect("file:database.db", uri=True)
con.execute("ANALYZE")  # 收集统计信息

tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table'", con)
all_columns = []

for _, row in tables.iterrows():
    table_name = row['name']
    cols = pd.read_sql(f"PRAGMA table_info({table_name})", con)
    cols['table'] = table_name
    # 读取系统表中的统计数据
    stats = pd.read_sql(f"SELECT * FROM sqlite_stat1 WHERE tbl='{table_name}'", con)
    if not stats.empty:
        cols = cols.merge(stats, left_on='name', right_on='idx', how='left')
    all_columns.append(cols)

final_df = pd.concat(all_columns, ignore_index=True)
con.close()

优势 #

• 执行速度极快，几乎不占用内存
• 适合快速了解表的大致规模（行数、索引分布等）

不足 #

• 统计信息有限，无法替代describe()的完整结果

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