用于 CSV 导入的 SQLite 事务答案

【问题标题】：SQLite transaction for CSV importing用于 CSV 导入的 SQLite 事务
【发布时间】：2013-08-06 10:15:47
【问题描述】：

我对 python 非常陌生，并且一直在研究我的树莓派，以启动并运行一个脚本，以将数百万传感器数据记录导入 sqlite。我想在交易中这样做，以提高流程效率。我正在尝试将事务分解为 10k 块，如下所示：Python CSV to SQLite

目前为止

import csv, sqlite3, time

def chunks(data, rows=10000):
    for i in range (0, len(data), rows):
            yield data[i:i+rows]

if __name__ == "__main__":

    t = time.time()

con = sqlite3.connect('test.db')
cur = con.cursor()
cur.execute("DROP TABLE IF EXISTS sensor;")
cur.execute("CREATE TABLE sensor(key INT, reading REAL);")

filename = 'dummy.csv'
reader = csv.reader(open(filename,"r"))
divdata = chunks(reader)

for chunk in divdata:
    cur.execute('BEGIN TRANSACTION')

    for col1, col2 in chunk:
            cur.execute('INSERT INTO sensor (key, reading) VALUES (?, ?)', (col1, col2))

    con.execute('COMMIT')

我在 python 3.2.3 中遇到以下错误：

Traceback (most recent call last):
File "/home/pi/test1.py", line 20, in <module>
for chunk in divdata:
File "/home/pi/test1.py", line 4, in chunks
for i in range (0, len(data), rows):
TypeError: object of type '_csv.reader' has no len()

我显然在某处搞砸了块部分，因为没有块和事务，一切（基本插入）都可以正常工作。任何帮助表示赞赏。

【问题讨论】：

标签： python csv sqlite raspberry-pi

【解决方案1】：

您的 SQL 看起来不错。不过，我确实看到您的 CSV 阅读器存在问题：它不支持您在 chunks() 中使用它的方式len()。

您可以使用更典型的for row in data 循环，或者如果您需要将文件分成块，则使用one of the techniques described in this thread。

【讨论】：

在您链接到的线程中使用生成器版本而不是列表版本有什么好处吗？
在生成器版本中，for chunk in gen_chunks(reader): print chunk 行出现错误“无效语法”，第二个“块”给出了问题。为什么会这样？

【解决方案2】：

quest中的代码sn-p有两个问题：

调用chunks 中的读者应该被包裹在list() 中
“提交”本应使用连接的commit() 方法

查看固定代码：

import csv, sqlite3, time

def chunks(data, rows=10000):
    for i in range (0, len(data), rows):
            yield data[i:i+rows]

if __name__ == "__main__":

    t = time.time()

con = sqlite3.connect('test.db')
cur = con.cursor()
cur.execute("DROP TABLE IF EXISTS sensor;")
cur.execute("CREATE TABLE sensor(key INT, reading REAL);")

filename = 'dummy.csv'
reader = csv.reader(open(filename,"r"))
divdata = chunks(list(reader))

for chunk in divdata:
    cur.execute('BEGIN TRANSACTION')

    for col1, col2 in chunk:
            cur.execute('INSERT INTO sensor (key, reading) VALUES (?, ?)', (col1, col2))

    con.commit()

【讨论】：

虽然此代码有效，但它会将文件的全部内容放入内存中，这对于非常大的文件可能是个问题。

【解决方案3】：

问题是你从csv.reader方法得到的对象不支持len()函数。事实上，这个 CSV 阅读器只有在收到指示时才会读取，因此它不知道文件中有多少条记录。

因此，您需要更新chunks 方法来处理不知道要分块的事物的数量。用这个替换那个函数：

def chunks(data, n=10000):
    buffer = [None] * n
    idx = 0
    for record in data:
        buffer[idx] = record
        idx += 1
        if idx == n:
            yield buffer
            buffer = [None] * n
            idx = 0
    if idx > 0:
        yield buffer[:idx]

它的作用是：只要有要检索的记录，它就会不断地从您的文件中检索记录，并且每个 n 行都会发出一个 n 记录列表。例如：

>>> for c in chunks(range(10), 3):
...     print(c)
...
[0, 1, 2]
[3, 4, 5]
[6, 7, 8]
[9]

【讨论】：