Python - 多线程帮助 - 读取多个文件 - ETL 到 SQL Server

Question

我正在开发一个从本地驱动器读取 DBF 文件并将数据加载到 SQL Server 表中的程序。我对 Python 很陌生，我发现了一些有关多线程的细节，其中大部分都令人困惑。读取和插入的性能很慢，看看我的CPU使用率，我有足够的容量。我也在运行 SSD。

此代码将扩展为读取约 400 个 zip 中的约 20 个 DBF 文件。所以我们讨论的总共是 8000 个 DBF 文件。

我很难做到这一点。可以指点一下吗？

这是我的代码(有点乱，但我稍后会清理它)，

import os, pyodbc, datetime, shutil
from dbfread import DBF
from zipfile import ZipFile

# SQL Server Connection Test
cnxn = pyodbc.connect('DRIVER={SQL Server};SERVER=localhost\test;DATABASE=TEST_DBFIMPORT;UID=test;PWD=test')
cursor = cnxn.cursor()

dr = 'e:\\Backups\\dbf\\'
work = 'e:\\Backups\\work\\'
archive = 'e:\\Backups\\archive\\'


for r in os.listdir(dr):

    curdate = datetime.datetime.now()
    filepath = dr + r
    process = work + r
    arc = archive + r

    pth = r.replace(".sss","")
    zipfolder = work + pth
    filedateunix = os.path.getctime(filepath)
    filedateconverted=datetime.datetime.fromtimestamp(int(filedateunix)
                                                  ).strftime('%Y-%m-%d %H:%M:%S')
    shutil.move(filepath,process)
    with ZipFile(process) as zf:
        zf.extractall(zipfolder)


    cursor.execute(
        "insert into tblBackups(backupname, filedate, dateadded) values(?,?,?)",
    pth, filedateconverted, curdate)
    cnxn.commit()

    for dirpath, subdirs, files in os.walk (zipfolder):

        for file in files:
            dateadded = datetime.datetime.now()

            if file.endswith(('.dbf','.DBF')):
                dbflocation = os.path.abspath(os.path.join(dirpath,file)).lower()

                if dbflocation.__contains__("\\bk.dbf"):
                    table = DBF(dbflocation, lowernames=True, char_decode_errors='ignore')
                    for record in table.records:
                        rec1 = str(record['code'])
                        rec2 = str(record['name'])
                        rec3 = str(record['addr1'])
                        rec4 = str(record['addr2'])
                        rec5 = str(record['city'])
                        rec6 = str(record['state'])
                        rec7 = str(record['zip'])
                        rec8 = str(record['tel'])
                        rec9 = str(record['fax'])
                        cursor.execute(
                       "insert into tblbk(code,name,addr1,addr2,city,state,zip,tel,fax) values(?,?,?,?,?,?,?,?,?)",
                        rec1, rec2, rec3, rec4, rec5, rec6, rec7, rec8, rec9, rec10, rec11, rec12, rec13)
                cnxn.commit()


                if dbflocation.__contains__("\\cr.dbf"):
                    table = DBF(dbflocation, lowernames=True, char_decode_errors='ignore')
                    for record in table.records:
                        rec2 = str(record['cal_desc'])
                        rec3 = str(record['b_date'])
                        rec4 = str(record['b_time'])
                        rec5 = str(record['e_time'])
                        rec6 = str(record['with_desc'])
                        rec7 = str(record['recuruntil'])
                        rec8 = record['notes']
                        rec9 = dateadded
                        cursor.execute(
                        "insert into tblcalendar(cal_desc,b_date,b_time,e_time,with_desc,recuruntil,notes,dateadded) values(?,?,?,?,?,?,?,?)",
                        rec2, rec3, rec4, rec5, rec6, rec7, rec8, rec9)
                cnxn.commit() 

    shutil.move(process, archive)
    shutil.rmtree(zipfolder)

Answer 1

tl;dr:先测量，再修复!

<小时/>

请注意，在最常见的 Python 实现 (CPython) 中，一次只能有一个线程执行 Python 字节码。因此，线程并不是提高 CPU 密集型性能的好方法。如果工作受 I/O 限制，它们可以很好地工作。

但是您首先应该做的是测量。这一点怎么强调都不为过。如果您不知道导致性能下降的原因，则无法修复它!

编写完成这项工作的单线程代码，并在分析器下运行它。首先尝试内置的cProfile。如果这不能为您提供足够的信息，请尝试例如一个line profiler .

分析应该告诉您哪些步骤消耗的时间最多。一旦了解了这一点，您就可以开始改进。

例如，如果将数据填充到 SQL Server 的操作花费最多时间，那么使用多处理来读取 DBF 文件是没有意义的!这甚至可能会减慢速度，因为多个进程会争夺 SQL 服务器的注意力。

如果 SQL 服务器不是瓶颈，并且它可以处理多个连接，我会使用多处理，可能是池.map() 并行读取 DBF 并将数据填充到 SQL 服务器中。在这种情况下，您应该 Pool.map 覆盖 DBF 文件名列表，以便在工作进程中打开这些文件。