SQLite3表列数超上限问题求优雅解决方案

我有一个工具，用于将CSV文件的数据加载到SQLite3表中。该CSV文件包含2000多列，而SQLite3表的列数上限为2000列。目前我采用的流程是将CSV加载到DataFrame中，删除不必要的列后再导入数据库表。但CSV文件一旦新增列，就会超出列数上限。该上限为硬编码设置，重新编译SQLite3超出我的能力范围。请问有什么优雅的处理方法？

现有加载流程如下：

df_survey = pd.read_csv(data_full_path)
print('File {} loaded into dataframe'.format(data_file))

# drop uneccessary columns
number_columns = len(df_survey.columns)
print('Number of columns in dataframe={}'.format(number_columns))

column_list = list(df_survey.columns.values)
for column in column_list:
    if column.startswith('Tag:'):
        df_survey.pop(column)

number_columns = len(df_survey.columns)
print('Number of columns after purge in dataframe={}'.format(number_columns))

table_name = 'wave'
df_survey.to_sql(table_name, gvars.db_conn, schema=None, if_exists='replace', index=False)

我目前还未尝试任何方案，希望先寻求建议再行动。

可行的优雅处理方案 #

1. 动态筛选+冗余预留，自动控制列数 #

先明确必须保留的核心列（比如主键、关键业务字段），剩余非核心列按规则筛选后，只保留到总列数不超过1990（留10个冗余位应对新增列）。即使CSV新增列，也只会自动截断优先级最低的非核心列，不会触发上限。

示例代码：

df_survey = pd.read_csv(data_full_path)
print(f'File {data_file} loaded into dataframe')

# 定义必须保留的核心列（替换为你的实际字段）
core_columns = ['user_id', 'survey_timestamp', 'response_status']
# 筛选非核心列，先剔除Tag:开头的字段
non_core_columns = [col for col in df_survey.columns if col not in core_columns and not col.startswith('Tag:')]

# 计算最多能保留的非核心列数：留10个冗余位避免新增列触发上限
max_allowed = 1990 - len(core_columns)
final_non_core = non_core_columns[:max_allowed]

# 合并核心列与最终非核心列
final_df = df_survey[core_columns + final_non_core]

print(f'Number of columns after processing: {len(final_df.columns)}')

table_name = 'wave'
final_df.to_sql(table_name, gvars.db_conn, schema=None, if_exists='replace', index=False)

2. 宽表转窄表（键值存储），彻底规避列数限制 #

将原本的多列宽表转换成行标识-列名-列值的窄表结构，完全不受SQLite列数限制，还能灵活应对任意新增列。

示例代码：

df_survey = pd.read_csv(data_full_path)
print(f'File {data_file} loaded into dataframe')

# 为每一行原始数据添加唯一标识（用于关联查询）
df_survey['row_identifier'] = df_survey.index

# 宽表转窄表：将所有非标识列转为键值对
df_long = pd.melt(
    df_survey,
    id_vars=['row_identifier'],  # 可添加其他核心关联字段如user_id
    var_name='field_name',
    value_name='field_value'
)

# 拆分存储：核心信息表+键值对表
core_df = df_survey[['row_identifier', 'user_id', 'survey_timestamp']]
core_df.to_sql('wave_core', gvars.db_conn, if_exists='replace', index=False)
df_long.to_sql('wave_field_values', gvars.db_conn, if_exists='replace', index=False)

查询原始数据时，通过row_identifier关联两张表：

SELECT c.row_identifier, c.user_id, f.field_name, f.field_value
FROM wave_core c
JOIN wave_field_values f ON c.row_identifier = f.row_identifier
WHERE c.row_identifier = 456;

3. 按类别拆分表，多表关联存储 #

将CSV的列按业务类别（如基础信息、扩展标签、自定义字段等）拆分到多个SQLite表，所有表通过同一主键关联，确保每个表的列数都在2000以内。

示例代码：

df_survey = pd.read_csv(data_full_path)
print(f'File {data_file} loaded into dataframe')

# 按类别拆分列
base_columns = ['user_id', 'survey_timestamp', 'response_status']
tag_columns = [col for col in df_survey.columns if col.startswith('Tag:')]
extra_columns = [col for col in df_survey.columns if col not in base_columns + tag_columns]

# 每个表保留关联主键，确保可以关联查询
df_base = df_survey[base_columns]
df_tags = df_survey[['user_id'] + tag_columns]
df_extra = df_survey[['user_id'] + extra_columns]

# 分别存入不同表
df_base.to_sql('wave_base', gvars.db_conn, if_exists='replace', index=False)
df_tags.to_sql('wave_tags', gvars.db_conn, if_exists='replace', index=False)
df_extra.to_sql('wave_extra', gvars.db_conn, if_exists='replace', index=False)

查询完整数据时通过主键关联：

SELECT *
FROM wave_base b
JOIN wave_tags t ON b.user_id = t.user_id
JOIN wave_extra e ON b.user_id = e.user_id
WHERE b.user_id = 123;

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