Python - マルチスレッドヘルプ - 複数のファイルを読む - ETLをSQL Serverに変換

Question

DBFファイルをローカルドライブから読み込み、SQL Serverテーブルにロードするプログラムを作成しています。私はPythonでかなり緑色で、マルチスレッドに関するいくつかの詳細を見つけました。そのほとんどは混乱しています。読み込みと挿入のパフォーマンスが遅く、CPU使用量を見ると、十分な容量があります。 SSDも実行しています。

このコードは、約400個のジップのうち約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を先に試してください。それで十分な情報が得られない場合は、 a line profiler。

プロファイリングでは、最も多くの時間を費やしているステップがわかります。あなたがそれを知ったら、あなたは改善を開始することができます。

たとえば、DBFファイルの読み込みにmultiprocessingを使用するのは意味がありません。これは、SQLサーバーにデータを入れる作業が最も時間がかかります。これにより、SQLサーバーの注意のためにいくつかのプロセスが戦っているため、処理が遅くなる可能性があります。

SQLサーバーがボトルネックでない場合、とそれが複数の接続を処理することができ、私は並列にDBFのを読んで、SQLサーバーにデータを詰め込むためにPool.map()おそらく、multiprocessingを使用します。この場合、DBFファイル名のリストにはPool.mapが必要です。そのため、ファイルはワーカープロセスで開かれます。