3 백만 개의 레코드에 대해 python postgresql 쿼리를 최적화하는 방법

Question

python 스크립트를 사용하여 postgresql에 3 ~ 4 백만 개의 데이터를 삽입하거나 업데이트하려고합니다. 아래 코드를 참조하십시오. 요구 사항은 새로운 키가 삽입되면 삽입되거나 키가 이미 존재하는 경우 새 값으로 키를 업데이트하는 것입니다. 그러나 아래의 코드는 DB에 너무 많은 왕복 연결을하고 있으며 DB에 3 백만 개의 레코드를 삽입하는 데 약 35-45 분 정도 소요됩니다. 왕복 연결을 피하고 빠른 방법으로 삽입 또는 업데이트하는 방법은 무엇입니까?

도움이 될 것입니다.

미리 도움을 주셔서 감사합니다.

InputFile.txt -이 파일은 준비된 명령문의 약 3 insert.py

productKey1 printer1,printerModel1,printerPrice1,printerDesc1| 
productKey2 sacnner2,scannerModel2,scannerPrice2,scannerDesc2| 
productKey3 mobile3,mobileModel3,mobilePrice3,mobileDesc3| 
productKey4 tv4,tvModel4,tvPrice4,tvDescription4| 
productKey2 sacnner22,scannerModel22,scannerPrice22,scannerDesc22| 

라인 ITESM 만 4-

def insertProduct(filename, conn): 
    seen = set() 
    cursor = conn.cursor() 
    qi = "INSERT INTO productTable (key, value) VALUES (%s, %s);" 
    qu = "UPDATE productTable SET value = CONCAT(value, %s) WHERE key = %s;" 

    with open(filename) as f: 
    for line in f: 
     if line.strip(): 
     key, value = line.split(' ', 1) 
     if key not in seen: 
      seen.add(key) 
      cursor.execute(qi, (key, value)) 
     else: 
      cursor.execute(qu, (value, key)) 

     conn.commit() 

conn = psycopg2.connect("dbname='productDB' user='myuser' host='localhost'") 
insertProduct('InputFile.txt', conn) 

Answer 1

실행 배치를 가지고있다. http://initd.org/psycopg/docs/extras.html#fast-execution-helpers

import psycopg2, psycopg2.extras 
def insertProduct(filename, conn): 

    data = [] 
    with open(filename) as f: 
     for line in f: 
      line = line.strip() 
      if line: 
       key, value = line.split(' ', 1) 
       data.append((key, value)) 

    cursor = conn.cursor() 
    cursor.execute(""" 
     prepare upsert (text, text) as 
     with i as (
      insert into productTable (key, value) 
      select $1, $2 
      where not exists (select 1 from productTable where key = $1) 
      returning * 
     ) 
     update productTable p 
     set value = concat (p.value, $2) 
     where p.key = $1 and not exists (select 1 from i) 
    """) 
    psycopg2.extras.execute_batch(cursor, "execute upsert (%s, %s)", data, page_size=500) 
    cursor.execute("deallocate upsert") 
    conn.commit()   

conn = psycopg2.connect(database='cpn') 
insertProduct('InputFile.txt', conn)