From Hours to Minutes: A Step-by-Step Journey of Optimizing Large-Scale Data Imports

import pandas as pd
import psycopg2

def import_csv_naive(csv_file, connection):
    """The simplest possible approach - read everything, insert row by row"""
    
    # Read entire CSV into memory
    df = pd.read_csv(csv_file)
    
    cursor = connection.cursor()
    
    # Insert each row one by one
    for index, row in df.iterrows():
        sql = """
            INSERT INTO generation_data 
            (timestamp, generator_id, power_output) 
            VALUES (%s, %s, %s)
        """
        cursor.execute(sql, (row['timestamp'], row['generator_id'], row['power']))
    
    connection.commit()

CSV File (4GB) → Python reads all → RAM (15GB) → Send to DB row-by-row → Database
                  ↑                   ↑           ↑
                  SLOW              WASTEFUL    EXTREMELY SLOW

def import_csv_chunked(csv_file, connection):
    """Read in chunks and batch insert"""
    
    CHUNK_SIZE = 10000  # Process 10k rows at a time
    
    # Read CSV in chunks to manage memory
    for chunk_df in pd.read_csv(csv_file, chunksize=CHUNK_SIZE):
        
        # Prepare batch insert
        records = chunk_df.to_records(index=False)
        
        # Use execute_values for batch insert (much faster!)
        cursor = connection.cursor()
        psycopg2.extras.execute_values(
            cursor,
            """INSERT INTO generation_data 
               (timestamp, generator_id, power_output) 
               VALUES %s""",
            records,
            template="(%s, %s, %s)",
            page_size=1000
        )
        connection.commit()

Before: INSERT, INSERT, INSERT, INSERT... (10,000 database calls)
After:  INSERT 1000 rows, INSERT 1000 rows... (10 database calls)

def import_csv_filtered(csv_file, connection, relevant_generators):
    """Only import data for generators we care about"""
    
    # We only need 50 out of 1000+ generators
    relevant_set = set(relevant_generators)  # O(1) lookup
    
    for chunk_df in pd.read_csv(csv_file, chunksize=10000):
        
        # Filter BEFORE processing - this is key!
        filtered_df = chunk_df[chunk_df['generator_id'].isin(relevant_set)]
        
        if filtered_df.empty:
            continue  # Skip if no relevant data
        
        # Now we're only inserting 5% of the data
        records = filtered_df.to_records(index=False)
        # ... batch insert as before

Original: 100 million rows → Process all → Insert all → Database
Filtered: 100 million rows → Process all → Insert 5 million → Database
                                              ↑
                                    95% reduction in database work!

# Slow: List lookup is O(n)
if generator_id in relevant_list:  # Checks each item one by one
    
# Fast: Set lookup is O(1)  
if generator_id in relevant_set:   # Instant hash table lookup

from multiprocessing import Process, Queue

def worker(file_path, worker_id, result_queue):
    """Each worker processes one file independently"""
    
    # Each worker gets its own database connection
    connection = psycopg2.connect(DATABASE_URL)
    
    # Process the file
    result = import_csv_filtered(file_path, connection)
    result_queue.put(result)

def import_parallel(csv_files):
    """Process multiple files simultaneously"""
    
    processes = []
    result_queue = Queue()
    
    # Start one worker per file
    for i, csv_file in enumerate(csv_files):
        p = Process(target=worker, args=(csv_file, i, result_queue))
        p.start()
        processes.append(p)
    
    # Wait for all workers to complete
    for p in processes:
        p.join()

Sequential (Before):
File 1 ████████████ Done
File 2              ████████████ Done
File 3                           ████████████ Done
File 4                                        ████████████ Done
Time: ──────────────────────────────────────────────────────────→

Parallel (After):
File 1 ████████████ Done
File 2 ████████████ Done
File 3 ████████████ Done
File 4 ████████████ Done
Time: ─────────────→  (4x faster!)

Before: CPU Core 1: 100% | Core 2: 0% | Core 3: 0% | Core 4: 0%
After:  CPU Core 1: 95%  | Core 2: 93% | Core 3: 96% | Core 4: 94%

import pandas as pd
import time

# Reading Excel file - SLOW
start = time.time()
df_excel = pd.read_excel('large_data.xlsx', engine='openpyxl')
print(f"Excel read time: {time.time() - start:.2f} seconds")
# Output: Excel read time: 45.23 seconds

# Reading CSV file - FAST
start = time.time()
df_csv = pd.read_csv('large_data.csv')
print(f"CSV read time: {time.time() - start:.2f} seconds")
# Output: CSV read time: 2.14 seconds

def optimize_excel_import(excel_file):
    """Convert Excel to CSV first for 20x faster processing"""
    
    # One-time conversion cost
    df = pd.read_excel(excel_file, engine='openpyxl')
    csv_file = excel_file.replace('.xlsx', '.csv')
    df.to_csv(csv_file, index=False)
    
    # Now use the CSV for all subsequent operations
    # This is 20x faster for repeated reads
    return csv_file

# Even better: Convert Excel to CSV using command line (faster)
# $ in2csv large_data.xlsx > large_data.csv

# Write compressed CSV (80% smaller than original)
df.to_csv('data.csv.gz', compression='gzip', index=False)

# Read compressed CSV (still 10x faster than Excel)
df = pd.read_csv('data.csv.gz', compression='gzip')

# For maximum speed with compression
import polars as pl
df = pl.read_csv('data.csv.gz')  # Handles compression automatically

import polars as pl  # Rust-based, columnar dataframe library

def import_with_polars(csv_file, connection):
    """Use Polars for 5-10x faster CSV reading"""
    
    # Polars reads CSV much faster than Pandas
    df = pl.read_csv(
        csv_file,
        has_header=True,
        n_threads=4  # Multi-threaded CSV parsing!
    )
    
    # Filter using Polars (columnar operations are faster)
    filtered_df = df.filter(
        pl.col('generator_id').is_in(relevant_generators)
    )
    
    # Convert to pandas for database insertion
    pandas_df = filtered_df.to_pandas()

Pandas (Row-based):
Row 1: [ID: 1, Time: "2024-01-01", Power: 100]
Row 2: [ID: 2, Time: "2024-01-01", Power: 150]
↑ Must read entire row even for one column

Polars (Column-based):
ID:    [1, 2, 3, 4, ...]
Time:  ["2024-01-01", "2024-01-01", ...]
Power: [100, 150, 120, ...]
↑ Can read just the columns you need

# Using Arrow format for even better performance
df = pd.read_csv(
    csv_file,
    engine='pyarrow',  # C++ implementation, 3x faster
    dtype_backend='pyarrow'  # Efficient memory format
)

import asyncpg  # Async PostgreSQL driver
from io import StringIO

async def import_with_copy(csv_file, connection):
    """Use PostgreSQL COPY command - the fastest way to bulk load"""
    
    # Read and filter data
    df = pl.read_csv(csv_file)
    filtered_df = df.filter(...)
    
    # Convert to CSV format in memory
    output = StringIO()
    filtered_df.write_csv(output)
    output.seek(0)
    
    # Use COPY command - bypasses SQL parsing entirely!
    await connection.copy_from_table(
        'generation_data',
        source=output,
        columns=['timestamp', 'generator_id', 'power_output'],
        format='csv'
    )

INSERT (even batched):
SQL Parse → Plan → Execute → Write → SQL Parse → Plan → Execute → Write...
~1,000 rows/second per connection

COPY:
Stream data directly → Write to disk
~50,000 rows/second per connection

-- Disable synchronous commit for bulk loads (faster but less safe)
SET synchronous_commit = OFF;

-- Increase memory for maintenance operations
SET maintenance_work_mem = '2GB';

-- Disable triggers and indexes during import
ALTER TABLE generation_data DISABLE TRIGGER ALL;
-- ... import data ...
ALTER TABLE generation_data ENABLE TRIGGER ALL;

import polars as pl
import asyncpg
import numpy as np
from multiprocessing import Process
import psutil

class OptimizedImporter:
    def __init__(self):
        # Dynamic configuration based on system resources
        self.memory = psutil.virtual_memory().available
        self.cpu_count = psutil.cpu_count()
        self.chunk_size = self.calculate_optimal_chunk_size()
    
    def calculate_optimal_chunk_size(self):
        """Dynamically determine chunk size based on available RAM"""
        # Use 10% of available memory per chunk
        memory_per_chunk = self.memory * 0.1
        # Estimate: 200 bytes per row
        return int(memory_per_chunk / 200)
    
    async def import_file(self, file_path, relevant_ids):
        """Combines all optimizations"""
        
        # 1. Use Polars for fast reading
        df = pl.scan_csv(file_path)  # Lazy evaluation
        
        # 2. Filter at scan time (before loading into memory)
        filtered = df.filter(
            pl.col('generator_id').is_in(relevant_ids)
        ).collect(streaming=True)  # Stream processing
        
        # 3. Remove duplicates in memory (faster than database)
        unique_df = filtered.unique(
            subset=['generator_id', 'timestamp']
        )
        
        # 4. Use PostgreSQL COPY for insertion
        conn = await asyncpg.connect(DATABASE_URL)
        
        # Convert to CSV format
        csv_buffer = StringIO()
        unique_df.write_csv(csv_buffer)
        csv_buffer.seek(0)
        
        # 5. Direct COPY (50x faster than INSERT)
        await conn.copy_from_table(
            'generation_data',
            source=csv_buffer,
            format='csv'
        )
        
        await conn.close()
    
    def run_parallel(self, files):
        """Process all files in parallel"""
        
        with ProcessPoolExecutor(max_workers=self.cpu_count) as executor:
            futures = [
                executor.submit(self.import_file, file, relevant_ids)
                for file in files
            ]
            
            # Progress bar
            for future in tqdm(as_completed(futures), total=len(files)):
                result = future.result()

# Specify data types explicitly to avoid inference overhead
dtype_spec = {
    'generator_id': 'int32',  # Use int32 instead of int64 if IDs < 2 billion
    'power_output': 'float32',  # float32 is often enough for measurements
    'status': 'category'  # Use category for repeated strings
}

# Pandas with explicit dtypes (3x faster reading, 50% less memory)
df = pd.read_csv('data.csv', dtype=dtype_spec)

# Polars with schema (even faster)
schema = {
    'generator_id': pl.Int32,
    'power_output': pl.Float32,
    'timestamp': pl.Datetime
}
df = pl.read_csv('data.csv', schema=schema)

def incremental_import(csv_file, last_timestamp):
    """Only import data newer than last import"""
    
    # Track last processed timestamp
    query = "SELECT MAX(timestamp) FROM generation_data"
    last_imported = connection.execute(query).fetchone()[0]
    
    # Filter during read using Polars lazy evaluation
    df = pl.scan_csv(csv_file).filter(
        pl.col('timestamp') > last_imported
    ).collect()
    
    # Result: Process only 1% of file on subsequent runs
    return df

def split_large_csv(input_file, chunk_size_mb=100):
    """Split large CSV into smaller chunks for parallel processing"""
    
    file_size = os.path.getsize(input_file) / (1024 * 1024)  # MB
    num_chunks = int(file_size / chunk_size_mb) + 1
    
    # Use GNU split for fastest splitting (command line)
    os.system(f"split -n {num_chunks} {input_file} chunk_")
    
    # Or use Pandas with chunks
    for i, chunk in enumerate(pd.read_csv(input_file, chunksize=1000000)):
        chunk.to_csv(f'chunk_{i}.csv', index=False)

import numpy as np
import pandas as pd

# Use memory mapping for huge files
df = pd.read_csv(
    'huge_file.csv',
    engine='c',
    memory_map=True,  # Memory-mapped file access
    low_memory=False
)

# Or use Dask for out-of-core processing
import dask.dataframe as dd
ddf = dd.read_csv('huge_file.csv', blocksize='100MB')
result = ddf.groupby('generator_id').mean().compute()

from psycopg2 import pool

# Create connection pool
connection_pool = pool.ThreadedConnectionPool(
    minconn=1,
    maxconn=20,  # Match your CPU cores
    host='localhost',
    database='energy_db'
)

def get_connection():
    return connection_pool.getconn()

def return_connection(conn):
    connection_pool.putconn(conn)

# Save processed data in Parquet format
df.to_parquet('data.parquet', compression='snappy')

# Reading Parquet is 10x faster than CSV
df = pd.read_parquet('data.parquet')  # 0.5 seconds vs 5 seconds for CSV

# Or use Feather for even faster read/write
df.to_feather('data.feather')
df = pd.read_feather('data.feather')  # 0.2 seconds

# Arrow format for zero-copy reads
import pyarrow as pa
table = pa.Table.from_pandas(df)
pa.parquet.write_table(table, 'data.arrow')

# For database sources - push filters to SQL
query = """
    SELECT * FROM raw_data 
    WHERE generator_id IN ({})
    AND timestamp > '2024-01-01'
""".format(','.join(map(str, relevant_ids)))

df = pd.read_sql(query, connection)

# For Parquet files - use predicate pushdown
df = pd.read_parquet(
    'data.parquet',
    filters=[
        ('generator_id', 'in', relevant_ids),
        ('timestamp', '>', '2024-01-01')
    ]
)

# Method 1: Hash-based deduplication (fastest)
def dedupe_with_hash(df):
    # Create hash of key columns
    df['hash'] = pd.util.hash_pandas_object(
        df[['generator_id', 'timestamp']]
    )
    # Keep first occurrence of each hash
    return df.drop_duplicates(subset=['hash']).drop('hash', axis=1)

# Method 2: Use database UPSERT (no duplicates)
sql = """
    INSERT INTO generation_data (generator_id, timestamp, power)
    VALUES %s
    ON CONFLICT (generator_id, timestamp) 
    DO UPDATE SET power = EXCLUDED.power
"""

# Compress data during network transfer
import zlib

def send_compressed_data(data, connection):
    # Compress data before sending
    compressed = zlib.compress(data.to_csv().encode())
    # 80% smaller = 5x faster network transfer
    
    # On receiving end
    decompressed = zlib.decompress(compressed).decode()
    df = pd.read_csv(StringIO(decompressed))

import psutil
import time

class PerformanceMonitor:
    def __init__(self):
        self.start_time = time.time()
        self.start_memory = psutil.Process().memory_info().rss / 1024 / 1024
    
    def checkpoint(self, label):
        elapsed = time.time() - self.start_time
        current_memory = psutil.Process().memory_info().rss / 1024 / 1024
        memory_used = current_memory - self.start_memory
        
        print(f"{label}:")
        print(f"  Time: {elapsed:.2f}s")
        print(f"  Memory: {memory_used:.0f} MB")
        print(f"  CPU: {psutil.cpu_percent()}%")

# Usage
monitor = PerformanceMonitor()
df = pd.read_csv('data.csv')
monitor.checkpoint("CSV Read")
filtered = df[df['generator_id'].isin(relevant_ids)]
monitor.checkpoint("Filtering")

V1: ████████████████████████████████████████████████ 8 hours
V2: ████████████████████████ 4 hours
V3: ██████ 1 hour
V4: █ 15 minutes
V5: ▋ 6 minutes
V6: ▌ 5 minutes
V7: ▎ 3 minutes (single file)
V8: ▌ 12 minutes (all files parallel)

import cProfile
profiler = cProfile.Profile()
profiler.enable()
# ... your code ...
profiler.disable()
profiler.print_stats(sort='cumulative')