非Pandas+Sybase：多CSV批量插入优化及多进程实现求助

Let’s start by fixing the low-hanging fruit in your current code—there are several obvious bottlenecks that are dragging down performance way more than necessary. Then we’ll cover smarter optimizations, including a correct multi-process implementation (since your first attempt failed) and even a native Sybase tool alternative that might blow Python out of the water.

First: Fix the Critical Flaws in Your Existing Code #

Looking at your code, I spot a few mistakes that are probably causing most of the 4-5 hour runtime:

• Reading the same file twice
  You open each CSV once to detect the region, then again to read the data. That’s double the disk I/O, which is a massive slowdown for 500MB files. We can grab the region from the filename once, then process the file in a single pass.
• Broken row iteration
  In your second file block, you’re looping over read—but that’s the csv.reader from the first (already closed) file handle. This is either skipping rows entirely or throwing silent errors that slow things down.
• Loading the entire file into memory
  You’re reading the full CSV into a list with data = list(reader) just to get total = len(data), then immediately setting total = -1. Loading 500MB into RAM is a huge waste—process rows line-by-line instead.
• Placeholder query mistake
  Your insert query has hardcoded values (1,2,3,4,5)—I assume that’s a placeholder, but Sybase uses ? as parameter placeholders for executemany, not numbered values. Using the wrong syntax can cause slowdowns or errors.

Step-by-Step Optimization Plan #

1. Streamline File Processing (Single Pass) #

Let’s rewrite the core logic to process each file once, with clean region detection:

import glob
import csv

# Assume you've already initialized your Sybase connection/cursor properly
conn.autocommit(False)  # Disable auto-commit to speed up batches

list_of_files = glob.glob('./*csv')
for file_name in list_of_files:
    # Grab region once, before opening the file
    if "cityA" in file_name:
        region = "cityA"
    elif "cityB" in file_name:
        region = "cityB"
    elif "cityC" in file_name:
        region = "cityC"
    else:
        print(f"Skipping unrecognized file: {file_name}")
        continue

    batch_size = 1000  # We can tweak this later
    row_count = 0
    batch_data = []

    # Single pass through the file with larger buffer for faster reading
    with open(file_name, 'r', buffering=1024*1024) as csvfile:
        reader = csv.reader(csvfile)
        next(reader)  # Skip header row

        for row in reader:
            row.append(region)
            batch_data.append(tuple(row))
            row_count += 1

            # Insert batch when we hit our limit
            if row_count >= batch_size:
                insert_query = """INSERT INTO table_name(A,B,C,D,E) 
                                  VALUES (?, ?, ?, ?, ?)"""
                cursor.executemany(insert_query, batch_data)
                conn.commit()
                row_count = 0
                batch_data = []

        # Insert any remaining rows after the loop ends
        if batch_data:
            cursor.executemany(insert_query, batch_data)
            conn.commit()

cursor.callproc('any_proc')
cursor.close()
conn.close()

2. Tune the Batch Size #

1000 rows per batch is a safe starting point, but you can likely go bigger. Test with 5000 or 10000 rows—larger batches reduce the number of round-trips to the database, which is one of the biggest bottlenecks. Just don’t go so big that you hit Sybase’s max batch size or run out of RAM.

3. Correct Multi-Process Implementation (Why Your First Attempt Failed) #

If your multi-process try didn’t work, it’s almost certainly because you shared a database connection across processes—Sybase connections aren’t thread/process-safe. Each process needs its own independent connection. Here’s how to do it right with multiprocessing.Pool:

import multiprocessing
import glob
import csv

def process_single_file(file_path):
    # Initialize connection/cursor INSIDE the process (critical!)
    conn = your_sybase_connection_setup()  # Replace with your actual connection code
    cursor = conn.cursor()
    conn.autocommit(False)

    # Detect region
    if "cityA" in file_path:
        region = "cityA"
    elif "cityB" in file_path:
        region = "cityB"
    elif "cityC" in file_path:
        region = "cityC"
    else:
        cursor.close()
        conn.close()
        return

    batch_size = 1000
    row_count = 0
    batch_data = []

    with open(file_path, 'r', buffering=1024*1024) as csvfile:
        reader = csv.reader(csvfile)
        next(reader)
        for row in reader:
            row.append(region)
            batch_data.append(tuple(row))
            row_count += 1
            if row_count >= batch_size:
                insert_query = "INSERT INTO table_name(A,B,C,D,E) VALUES (?, ?, ?, ?, ?)"
                cursor.executemany(insert_query, batch_data)
                conn.commit()
                row_count = 0
                batch_data = []
        if batch_data:
            cursor.executemany(insert_query, batch_data)
            conn.commit()

    cursor.callproc('any_proc')
    cursor.close()
    conn.close()

if __name__ == "__main__":
    files_to_process = glob.glob('./*csv')
    # Use number of processes equal to your CPU cores (or 2-3 if DB is the bottleneck)
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        pool.map(process_single_file, files_to_process)

Rule of thumb*: Never share DB connections between processes. Each process must create its own.*

4. Alternative: Use Sybase’s Native Bulk Load Tools #

If Python optimizations still aren’t fast enough, Sybase has built-in tools designed for massive data loads—like bcp (Bulk Copy Program) or dbisql with BULK INSERT commands. These are way faster than any Python script because they’re optimized for the database’s internal structure. You can even call them from Python using subprocess if you need to automate the workflow.

Quick Testing Tip #

Always test with a small sample of your data first to make sure the logic works, then scale up. This saves you from waiting hours only to find a bug.

内容的提问来源于stack exchange，提问作者Babru