在 Python 中使用 ibm_db 从 DB2 提取大型表的性能问题

Question

我使用 ibm_db Python 库从大型 DB2 表 (>50GB) 中提取数据并将其分块保存到 CSV 文件中。我尝试过不同的块大小，但提取过程仍然很慢。跑了大约30分钟后，我还注意到速度明显下降。我不确定问题是否在于我对查询分页的方式（使用 OFFSET）、块大小或我写入 CSV 文件的方式。

这是我的代码的简化版本：

import csv
import ibm_db

chunk =0
offset = 500000

with open(csv_file_path, mode = "w", newline="") as csvfile:
    csv_writer = None
    
    while True:
        query=f"{sql} offset {offset} rows fetch next {chunk} rows only"
        Stmt = ibm_db.exec_immediate(conn, query)
        row = ibm_db.fetch_assoc(stmt)
        
        if not row:
            break
        while row:
        
            if first_chunk:
                header = row.keys()
                csv_writer = csv.DictWriter(csvfile, fieldnames = header)
                first_chunk = False
                
            csv_writer.writerow(row)
            row = ibm_db.fetch_assoc(stmt)
        offset += chunk
        
ibm_db.close(conn)

我尝试过不同的块大小，但在内存使用与提取速度方面似乎回报递减。

Answer 1

提高性能：

1. 您可以使用键集分页进行分页 - 它可以避免像 offset 那样重新扫描所有先前的行。为此，您可以使用任何索引列（例如我的示例中的 id）。

query = f"SELECT * FROM table_name WHERE id > {last_id} ORDER BY id ASC FETCH NEXT {chunk} ROWS ONLY"

2. 您还可以批量写入CSV以减少I/O开销。

import csv
import ibm_db

last_id = 0  # or any value, depending on your data
chunk_size = 50000
buffer_size = 10000  # adjust based on memory limits
rows_buffer = []

with open(csv_file_path, mode="w", newline="") as csvfile:
    csv_writer = None

    while True:
        query = f"SELECT * FROM table_name WHERE id > {last_id} ORDER BY id ASC FETCH NEXT {chunk_size} ROWS ONLY"
        stmt = ibm_db.exec_immediate(conn, query)
        row = ibm_db.fetch_assoc(stmt)

        if not row:
            break
            
        while row:
            if not csv_writer:
                header = row.keys()
                csv_writer = csv.DictWriter(csvfile, fieldnames=header)
                csv_writer.writeheader()

            rows_buffer.append(row)
            
            if len(rows_buffer) >= buffer_size:
                csv_writer.writerows(rows_buffer)
                rows_buffer = []  # clear buffer

            last_id = row['id']  # update last_id for pagination
            row = ibm_db.fetch_assoc(stmt)

    # write remaining rows
    if rows_buffer:
        csv_writer.writerows(rows_buffer)

ibm_db.close(conn)