Sử dụng bảng COPY nhị phân TỪ với psycopg2

Question

Tôi có hàng chục triệu hàng để chuyển từ các tệp mảng đa chiều thành cơ sở dữ liệu PostgreSQL. Công cụ của tôi là Python và psycopg2. Cách hiệu quả nhất để xử lý dữ liệu hàng loạt là sử dụng copy_from. Tuy nhiên, dữ liệu của tôi chủ yếu là các số dấu phẩy động 32 bit (thực hoặc float4), vì vậy tôi không muốn chuyển đổi từ văn bản → thực → thực. Dưới đây là một cơ sở dữ liệu ví dụ DDL:

CREATE TABLE num_data 
(
    id serial PRIMARY KEY NOT NULL, 
    node integer NOT NULL, 
    ts smallint NOT NULL, 
    val1 real, 
    val2 double precision 
); 

Đây là nơi tôi đang ở với Python sử dụng chuỗi (text):

# Just one row of data 
num_row = [23253, 342, -15.336734, 2494627.949375] 

import psycopg2 
# Python3: 
from io import StringIO 
# Python2, use: from cStringIO import StringIO 

conn = psycopg2.connect("dbname=mydb user=postgres") 
curs = conn.cursor() 

# Convert floating point numbers to text, write to COPY input 
cpy = StringIO() 
cpy.write('\t'.join([repr(x) for x in num_row]) + '\n') 

# Insert data; database converts text back to floating point numbers 
cpy.seek(0) 
curs.copy_from(cpy, 'num_data', columns=('node', 'ts', 'val1', 'val2')) 
conn.commit() 

Có tương đương mà có thể làm việc bằng cách sử dụng chế độ nhị phân? Tức là, giữ các số dấu chấm động trong nhị phân? Điều này không chỉ duy trì độ chính xác của dấu phẩy động, mà còn có thể nhanh hơn.

(Lưu ý: để xem độ chính xác tương tự như ví dụ này, sử dụng SET extra_float_digits='2')

Answer 1

Dưới đây là tương đương nhị phân của COPY TỪ cho Python 3:

from io import BytesIO 
from struct import pack 
import psycopg2 

# Two rows of data; "id" is not in the upstream data source 
# Columns: node, ts, val1, val2 
data = [(23253, 342, -15.336734, 2494627.949375), 
     (23256, 348, 43.23524, 2494827.949375)] 

conn = psycopg2.connect("dbname=mydb user=postgres") 
curs = conn.cursor() 

# Determine starting value for sequence 
curs.execute("SELECT nextval('num_data_id_seq')") 
id_seq = curs.fetchone()[0] 

# Make a binary file object for COPY FROM 
cpy = BytesIO() 
# 11-byte signature, no flags, no header extension 
cpy.write(pack('!11sii', b'PGCOPY\n\377\r\n\0', 0, 0)) 

# Columns: id, node, ts, val1, val2 
# Zip: (column position, format, size) 
row_format = list(zip(range(-1, 4), 
         ('i', 'i', 'h', 'f', 'd'), 
         (4, 4, 2, 4, 8))) 
for row in data: 
    # Number of columns/fields (always 5) 
    cpy.write(pack('!h', 5)) 
    for col, fmt, size in row_format: 
     value = (id_seq if col == -1 else row[col]) 
     cpy.write(pack('!i' + fmt, size, value)) 
    id_seq += 1 # manually increment sequence outside of database 

# File trailer 
cpy.write(pack('!h', -1)) 

# Copy data to database 
cpy.seek(0) 
curs.copy_expert("COPY num_data FROM STDIN WITH BINARY", cpy) 

# Update sequence on database 
curs.execute("SELECT setval('num_data_id_seq', %s, false)", (id_seq,)) 
conn.commit() 

---

Cập nhật

Tôi viết lại cách tiếp cận trên để viết các tệp cho COPY. Dữ liệu của tôi trong Python là trong mảng NumPy, do đó, nó có ý nghĩa để sử dụng chúng. Dưới đây là một số ví dụ data với với hàng 1M, 7 cột:

import psycopg2 
import numpy as np 
from struct import pack 
from io import BytesIO 
from datetime import datetime 

conn = psycopg2.connect("dbname=mydb user=postgres") 
curs = conn.cursor() 

# NumPy record array 
shape = (7, 2000, 500) 
print('Generating data with %i rows, %i columns' % (shape[1]*shape[2], shape[0])) 

dtype = ([('id', 'i4'), ('node', 'i4'), ('ts', 'i2')] + 
     [('s' + str(x), 'f4') for x in range(shape[0])]) 
data = np.empty(shape[1]*shape[2], dtype) 
data['id'] = np.arange(shape[1]*shape[2]) + 1 
data['node'] = np.tile(np.arange(shape[1]) + 1, shape[2]) 
data['ts'] = np.repeat(np.arange(shape[2]) + 1, shape[1]) 
data['s0'] = np.random.rand(shape[1]*shape[2]) * 100 
prv = 's0' 
for nxt in data.dtype.names[4:]: 
    data[nxt] = data[prv] + np.random.rand(shape[1]*shape[2]) * 10 
    prv = nxt 

Mở cơ sở dữ liệu của tôi, tôi có hai bảng trông giống như:

CREATE TABLE num_data_binary 
(
    id integer PRIMARY KEY, 
    node integer NOT NULL, 
    ts smallint NOT NULL, 
    s0 real, 
    s1 real, 
    s2 real, 
    s3 real, 
    s4 real, 
    s5 real, 
    s6 real 
) WITH (OIDS=FALSE); 

và một bảng tương tự như tên num_data_text.

Dưới đây là một số chức năng helper đơn giản để chuẩn bị dữ liệu cho COPY (cả văn bản và định dạng nhị phân) bằng cách sử dụng các thông tin trong hồ sơ mảng NumPy:

def prepare_text(dat): 
    cpy = BytesIO() 
    for row in dat: 
     cpy.write('\t'.join([repr(x) for x in row]) + '\n') 
    return(cpy) 

def prepare_binary(dat): 
    pgcopy_dtype = [('num_fields','>i2')] 
    for field, dtype in dat.dtype.descr: 
     pgcopy_dtype += [(field + '_length', '>i4'), 
         (field, dtype.replace('<', '>'))] 
    pgcopy = np.empty(dat.shape, pgcopy_dtype) 
    pgcopy['num_fields'] = len(dat.dtype) 
    for i in range(len(dat.dtype)): 
     field = dat.dtype.names[i] 
     pgcopy[field + '_length'] = dat.dtype[i].alignment 
     pgcopy[field] = dat[field] 
    cpy = BytesIO() 
    cpy.write(pack('!11sii', b'PGCOPY\n\377\r\n\0', 0, 0)) 
    cpy.write(pgcopy.tostring()) # all rows 
    cpy.write(pack('!h', -1)) # file trailer 
    return(cpy) 

này làm thế nào tôi đang sử dụng các chức năng helper để chuẩn định dạng phương pháp hai COPY:

def time_pgcopy(dat, table, binary): 
    print('Processing copy object for ' + table) 
    tstart = datetime.now() 
    if binary: 
     cpy = prepare_binary(dat) 
    else: # text 
     cpy = prepare_text(dat) 
    tendw = datetime.now() 
    print('Copy object prepared in ' + str(tendw - tstart) + '; ' + 
      str(cpy.tell()) + ' bytes; transfering to database') 
    cpy.seek(0) 
    if binary: 
     curs.copy_expert('COPY ' + table + ' FROM STDIN WITH BINARY', cpy) 
    else: # text 
     curs.copy_from(cpy, table) 
    conn.commit() 
    tend = datetime.now() 
    print('Database copy time: ' + str(tend - tendw)) 
    print('  Total time: ' + str(tend - tstart)) 
    return 

time_pgcopy(data, 'num_data_text', binary=False) 
time_pgcopy(data, 'num_data_binary', binary=True) 

đây là kết quả từ hai time_pgcopy lệnh cuối cùng:

Processing copy object for num_data_text 
Copy object prepared in 0:01:15.288695; 84355016 bytes; transfering to database 
Database copy time: 0:00:37.929166 
     Total time: 0:01:53.217861 
Processing copy object for num_data_binary 
Copy object prepared in 0:00:01.296143; 80000021 bytes; transfering to database 
Database copy time: 0:00:23.325952 
     Total time: 0:00:24.622095 

Vì vậy, cả tệp và tệp → NumPy → các bước cơ sở dữ liệu đều nhanh hơn với phương pháp nhị phân. Sự khác biệt rõ ràng là cách Python chuẩn bị tệp COPY, điều này thực sự chậm đối với văn bản. Nói chung, định dạng nhị phân tải vào cơ sở dữ liệu trong 2/3 thời gian dưới dạng định dạng văn bản cho lược đồ này.

Cuối cùng, tôi đã so sánh các giá trị trong cả hai bảng trong cơ sở dữ liệu để xem các số liệu có khác nhau hay không. Khoảng 1,46% các hàng có các giá trị khác nhau cho cột s0 và phần này tăng lên 6,17% cho s6 (có thể liên quan đến phương pháp ngẫu nhiên mà tôi đã sử dụng). Sự khác biệt tuyệt đối không khác giữa tất cả các giá trị float 70-bit 70M nằm trong khoảng 9.3132257e-010 và 7.6293945e-006. Những khác biệt nhỏ giữa văn bản và phương thức tải nhị phân là do mất chính xác từ float → văn bản → chuyển đổi float cần thiết cho phương thức định dạng văn bản.

Answer 2

Here là phiên bản của tôi. Dựa trên phiên bản của Mike.

rất nó ad-hoc nhưng có hai ưu:

• Expect máy phát điện và hoạt động như một dòng suối do quá tải readline
• Ví dụ làm thế nào để viết trong hstore định dạng nhị phân.