如何使用Sqlite3将csv文件中的数据导入我的表中

Question

我希望使用Python (Pandas )从csv文件(具有4列)导入sqlite3数据库中已经创建的表(表具有5列)中的所有数据(约20万行)。csv中的数据类型和表中的数据类型相互满足。问题是，在表中有一个额外的列，即index_of(Primary Key)。

这是我的csv文件的前3行：

​

​

这就是我所能做的，我想如果它能工作的话，大约需要5-6个小时，因为在这段代码中，我使用for循环来读取每一行：

connection = _sqlite3.connect("db_name.sqlite")
cursor = connection.cursor()
with open('path_to_csv', 'r') as file:
    no_records = 0
    for row in file: 
        cursor.execute("INSERT INTO table_name (index_of, high, low, original, ship_date) VALUES (?,?,?,?,?)", row.split(","))
        connection.commit()
        no_records += 1  
        
connection.close()

但是它显示了一个错误:发生了异常:4列的OperationalError 5值

拜托，你能帮我这个忙吗？

1. 如何使用

快速导入20万行

1. 如何将csv文件中的所有列导入表的特定列？

Answer 1

user3380595已经在their answer中指出，您需要为列index_of提供一个值。

cursor.execute("""
    INSERT INTO Quotes (index_of, high, low, original, ship_date)
    VALUES (?, ?, ?, ?, ?)
    """, [index, *row])

我创建了20万行测试数据，它加载得相当快(不到2秒)。参见使用csv和sqlite3的第一个示例。

正如user3380595所提到的，如果您关心内存和性能，可以以块的形式加载数据。实际上，加载该场景的速度稍慢一些。参见使用pandas和sqlalchemy的第二个示例。

使用csv 和 sqlite3的

设置测试环境

import csv
import sqlite3
import contextlib

import pandas as pd

test_data = r"/home/thomas/Projects/Playground/stackoverflow/data/test.csv"
test_db = r"/home/thomas/Projects/Playground/stackoverflow/data/test.db"

with contextlib.closing(sqlite3.connect(test_db)) as connection:
    
    cursor = connection.cursor()

    cursor.execute("DROP TABLE IF EXISTS Quotes;")

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS Quotes (
            index_of INTEGER, -- PRIMARY KEY,
            high REAL,
            low REAL,
            original REAL,
            ship_date TEXT
        );
        """)

    connection.commit()

负载数据

with contextlib.closing(sqlite3.connect(test_db)) as connection:
    
    cursor = connection.cursor()

    with open(test_data, "r") as file:
        
        for index, row in enumerate(csv.reader(file)):
            cursor.execute("""
                INSERT INTO Quotes (index_of, high, low, original, ship_date)
                VALUES (?, ?, ?, ?, ?)
                """, [index, *row])

    connection.commit()

使用pandas 和 sqlalchemy的

设置测试环境

import pandas as pd

from sqlalchemy import create_engine

test_data = r"/home/thomas/Projects/Playground/stackoverflow/data/test.csv"
test_db = r"sqlite:////home/thomas/Projects/Playground/stackoverflow/data/test.db"

engine = create_engine(test_db, echo=True)

with engine.begin() as connection:

    engine.execute("DROP TABLE IF EXISTS Quotes;")

    engine.execute("""
        CREATE TABLE IF NOT EXISTS Quotes (
            index_of INTEGER, -- PRIMARY KEY,
            high REAL,
            low REAL,
            original REAL,
            ship_date TEXT
        );
        """)

加载数据(以块形式)

with engine.begin() as connection:

    reader = pd.read_csv(test_data, iterator=True, chunksize=50000)

    for chunk in reader:
        chunk["index_of"] = chunk.index
        chunk.to_sql("Quotes", con=engine, if_exists="append", index=False)

或者，您也可以使用pandas和处理行块，而不是使用sqlite3.Cursor.executemany。

或者，您可以使用dask.dataframe.to_sql (例如假设一个现有的数据库中有一个表引号)并并行编写数据。但是，我不认为你的数据需要它。

import dask.dataframe as dd

test_data = r"/home/thomas/Projects/Playground/stackoverflow/data/test.csv"
test_db = r"sqlite:////home/thomas/Projects/Playground/stackoverflow/data/test.db"

df = dd.read_csv(test_data)  # , blocksize=2e6
df["index_of"] = df.index
df.to_sql("Quotes", uri=test_db, if_exists="append", index=False, parallel=True)

Answer 2

您需要为第5列提供一个默认值。

如果在每个SQL语句中插入100至200行的块，还可以提高脚本的性能。