使用Python和SQLAlchemy导入CSV数据至数据库的问题排查及50列数据表创建与导入方法

一、为什么CSV数据没插入到数据库？ #

你的代码里有几个关键问题导致数据没写入：

• 错误地用csv.reader读取DataFrame
  当你用pd.read_csv(train_url)已经把数据加载成了DataFrame，但接下来把这个DataFrame传给csv.reader是完全错误的——csv.reader需要的是文件对象或每行是字符串的可迭代对象，而DataFrame不是这种类型。结果就是你的for i in data_reader循环根本没执行，自然没有数据被添加到会话里。

• 列名大小写不匹配（潜在问题）
  你的表定义里列名是小写的y2、y3，但构造record时用了大写的Y2、Y3。虽然SQLite默认大小写不敏感，但为了代码规范和避免其他数据库的兼容性问题，最好保持列名一致。

修正后的高效写法（用pandas原生to_sql） #

其实完全不用手动写循环插入，pandas的to_sql方法可以直接把DataFrame写入数据库，代码更简洁高效：

import sqlalchemy as db
import pandas as pd

def main():
    # 初始化引擎
    engine = db.create_engine('sqlite:///train_database.db')
    
    # 创建表（如果不存在）
    meta_data = db.MetaData()
    train_table = db.Table(
        "train_table", meta_data,
        db.Column("x", db.Float),
        db.Column("y1", db.Float),
        db.Column("y2", db.Float),
        db.Column("y3", db.Float),
        db.Column("y4", db.Float)
    )
    meta_data.create_all(engine)
    
    # 加载CSV并写入数据库
    try:
        train_url = 'https://raw.githubusercontent.com/Jacmanski/dataset/main/train.csv'
        train_data = pd.read_csv(train_url)
        # to_sql直接写入，if_exists='append'表示追加数据（表已存在时）
        train_data.to_sql(
            name='train_table',
            con=engine,
            if_exists='append',
            index=False,  # 不写入DataFrame的索引列
            dtype={
                'x': db.Float,
                'y1': db.Float,
                'y2': db.Float,
                'y3': db.Float,
                'y4': db.Float
            }
        )
        print("数据插入成功！")
    except Exception as e:
        print(f"插入失败：{str(e)}")
    finally:
        engine.dispose()  # 关闭引擎连接

if __name__ == '__main__':
    main()

如果坚持要用ORM会话方式，应该遍历DataFrame的行而非用csv.reader：

# 替换原代码中的try块内容
train_url = 'https://raw.githubusercontent.com/Jacmanski/dataset/main/train.csv'
train_data = pd.read_csv(train_url)

session = sessionmaker(bind=engine)()
try:
    for _, row in train_data.iterrows():
        record = train_table(
            x=row['x'],
            y1=row['y1'],
            y2=row['y2'],
            y3=row['y3'],
            y4=row['y4']
        )
        session.add(record)
    session.commit()
    print("数据插入成功！")
except Exception as e:
    session.rollback()
    print(f"插入失败：{str(e)}")
finally:
    session.close()

二、导入50列的CSV到数据表 #

对于50列的场景，手动写50个db.Column太繁琐，直接用pandas的to_sql自动创建表是最优解——它会根据CSV的列名和数据类型自动生成对应的表结构：

import sqlalchemy as db
import pandas as pd

def import_ideal_data():
    engine = db.create_engine('sqlite:///train_database.db')
    ideal_url = 'https://raw.githubusercontent.com/Jacmanski/dataset/main/ideal.csv'
    
    try:
        ideal_data = pd.read_csv(ideal_url)
        # 自动创建表并插入数据，if_exists='replace'覆盖已有表，'append'则追加
        ideal_data.to_sql(
            name='ideal_table',
            con=engine,
            if_exists='replace',
            index=False
        )
        print("50列数据导入成功！")
    except Exception as e:
        print(f"导入失败：{str(e)}")
    finally:
        engine.dispose()

if __name__ == '__main__':
    import_ideal_data()

如果需要统一指定所有列的数据类型（比如全部为Float），可以先获取列名构造dtype字典：

ideal_data = pd.read_csv(ideal_url)
# 构造所有列的dtype为Float
dtype_dict = {col: db.Float for col in ideal_data.columns}
ideal_data.to_sql(
    name='ideal_table',
    con=engine,
    if_exists='replace',
    index=False,
    dtype=dtype_dict
)

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