Working with CSV Files in Python

Data Validation and Cleaning

# File: examples/data_validation.py
import csv
import re

def validate_email(email):
    """Simple email validation."""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def validate_and_clean_csv(input_file, output_file):
    valid_rows = []
    invalid_rows = []
    
    with open(input_file, 'r', newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        
        for row_num, row in enumerate(reader, start=2):  # Start from 2 to account for header
            # Create a clean copy of the row
            clean_row = {}
            is_valid = True
            
            # Validate and clean each field
            for field, value in row.items():
                # Strip whitespace
                clean_value = value.strip()
                
                # Field-specific validation and cleaning
                if field == 'email':
                    if not validate_email(clean_value):
                        is_valid = False
                        print(f"Row {row_num}: Invalid email: {clean_value}")
                elif field == 'age':
                    try:
                        age = int(clean_value)
                        if age < 0 or age > 120:
                            is_valid = False
                            print(f"Row {row_num}: Age out of range: {clean_value}")
                    except ValueError:
                        is_valid = False
                        print(f"Row {row_num}: Invalid age format: {clean_value}")
                
                # Store the cleaned value
                clean_row[field] = clean_value
            
            # Add to appropriate list
            if is_valid:
                valid_rows.append(clean_row)
            else:
                invalid_rows.append(clean_row)
    
    # Write valid rows to output file
    with open(output_file, 'w', newline='') as csvfile:
        if valid_rows:
            writer = csv.DictWriter(csvfile, fieldnames=valid_rows[0].keys())
            writer.writeheader()
            writer.writerows(valid_rows)
    
    # Also write invalid rows to a separate file for review
    with open(output_file.replace('.csv', '_invalid.csv'), 'w', newline='') as csvfile:
        if invalid_rows:
            writer = csv.DictWriter(csvfile, fieldnames=invalid_rows[0].keys())
            writer.writeheader()
            writer.writerows(invalid_rows)
    
    return len(valid_rows), len(invalid_rows)

# Example usage
valid_count, invalid_count = validate_and_clean_csv(
    'data/user_data.csv',
    'data/output/validated_users.csv'
)
print(f"Processed {valid_count + invalid_count} rows: {valid_count} valid, {invalid_count} invalid")
                

Working with Large CSV Files

Large CSV files present unique challenges - they may not fit in memory, and processing them can be time-consuming.

# File: examples/large_file_processing.py
import csv
import os

def process_large_csv_in_chunks(input_file, output_file, chunk_size=1000):
    """
    Process a large CSV file in chunks to avoid memory issues.
    """
    # Create a temporary file for each chunk
    temp_files = []
    
    with open(input_file, 'r', newline='') as infile:
        reader = csv.DictReader(infile)
        fieldnames = reader.fieldnames
        
        # Process in chunks
        chunk = []
        chunk_num = 0
        
        for row in reader:
            # Process the row (example: capitalize all string values)
            processed_row = {k: v.upper() if isinstance(v, str) else v 
                            for k, v in row.items()}
            chunk.append(processed_row)
            
            # When chunk is full, write to temp file
            if len(chunk) >= chunk_size:
                temp_file = f"temp_chunk_{chunk_num}.csv"
                with open(temp_file, 'w', newline='') as outfile:
                    writer = csv.DictWriter(outfile, fieldnames=fieldnames)
                    writer.writeheader()
                    writer.writerows(chunk)
                
                temp_files.append(temp_file)
                chunk = []
                chunk_num += 1
        
        # Write any remaining rows
        if chunk:
            temp_file = f"temp_chunk_{chunk_num}.csv"
            with open(temp_file, 'w', newline='') as outfile:
                writer = csv.DictWriter(outfile, fieldnames=fieldnames)
                writer.writeheader()
                writer.writerows(chunk)
            
            temp_files.append(temp_file)
    
    # Merge temp files into final output
    with open(output_file, 'w', newline='') as outfile:
        writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        writer.writeheader()
        
        for temp_file in temp_files:
            with open(temp_file, 'r', newline='') as infile:
                reader = csv.DictReader(infile)
                for row in reader:
                    writer.writerow(row)
            
            # Remove temp file after using it
            os.remove(temp_file)
    
    return len(temp_files)

# Example usage
chunks_processed = process_large_csv_in_chunks(
    'data/very_large_dataset.csv',
    'data/output/processed_large_dataset.csv',
    chunk_size=5000
)
print(f"Processed file in {chunks_processed} chunks")
                

Processing large files in chunks is like dividing a large book into chapters for easier reading - you handle one manageable piece at a time.