python - 如何使用 pandas 和 sqlalchemy 进行分块读写

Question

我正在使用 python(版本 3.4.4)、pandas(版本 0.19.1)和 sqlalchemy(版本 1.1.4)以便从大型 SQL 表中按 block 读取，预处理这些 block 并将它们写入不同的SQL 表。连续分块读取 pd.read_sql_query(verses_sql, conn, chunksize=10) ，其中pd是 pandas 导入，verses_sql是 SQL 查询并且 conn是 DB-API 连接，如果我这样做就可以正常工作:

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine('mssql+pymssql://<username>:<password>@<database>:1433/<FirstTable>')
conn = engine.connect()

verses_sql = '''SELECT [KA_Lang] FROM [dbo].[<FirstTable>]'''

for chunk in pd.read_sql_query(verses_sql, conn, chunksize=10):
    chunk['KA_Lang'] = chunk['KA_Lang'].str.replace(r'[^a-zA-Z\u00C0-\u02AF]'," ")
    chunk['KA_Lang'] = chunk['KA_Lang'].str.replace(r'\s\s+', " ")
    chunk['KA_Lang'] = chunk['KA_Lang'].str.lower()
    print(chunk['KA_Lang'].head(1))

问题是:如果我尝试写入预处理的 block chunk['KA_Lang']在第二个 SQL 表中，我们将其称为 SecondTable ，仅传递 10 个元素的第一个 block 。迭代就此停止。这是改编后的代码:

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy import Table, Column, Integer, String, MetaData

engine = create_engine('mssql+pymssql://<username>:<password>@<database>:1433/<FirstTable>')
conn = engine.connect()

verses_sql = '''SELECT [KA_Lang] FROM [dbo].[<FirstTable>]'''

for chunk in pd.read_sql_query(verses_sql, conn, chunksize=10):
    chunk['KA_Lang'] = chunk['KA_Lang'].str.replace(r'[^a-zA-Z\u00C0-\u02AF]'," ")
    chunk['KA_Lang'] = chunk['KA_Lang'].str.replace(r'\s\s+', " ")
    chunk['KA_Lang'] = chunk['KA_Lang'].str.lower()
    print(chunk['KA_Lang'].head(1))

    chunk.to_sql('<SecondTable>', conn, if_exists= 'append', index= False)

conn.close()

如何连续从一个 SQL 表中读取一个 block 并将其写入到另一个 SQL 表中？如果我包含:chunk.to_sql('<SecondTable>', conn, if_exists= 'append', index= False)，为什么所有 block 的迭代都会停止？

Answer 1

经过几天尝试不同的解决方法，我解决了这个问题。这相当容易。为了连续地从一个 SQL 表中读取一个 block 并将其写入到不同的 SQL 表中，需要定义两个不同的连接:

engine = create_engine('mssql+pymssql://<username>:<password>@<database>:1433/<FirstTable>')
engine1 = create_engine('mssql+pymssql://<username>:<password>@<database>:1433/<FirstTable>')
conn = engine.connect()
conn1 = engine1.connect()

第二个表中写入chunk的代码行需要调整为:

chunk.to_sql('<SecondTable>', conn1, if_exists= 'append', index= False)

完成!