Python Data Processing: Complete Guide with Pandas and NumPy

Data processing is a fundamental skill for Python developers working with datasets. This comprehensive guide covers data processing techniques using Python's powerful libraries and built-in tools.

Table of Contents #

Data Processing Fundamentals
Working with CSV Data
Data Cleaning Techniques
Data Transformation
Aggregation and Grouping
Time Series Processing
Performance Optimization

Data Processing Fundamentals #

Basic Data Structures #

Understanding Python's core data structures for processing:

🐍 Try it yourself

# Basic data processing with Python built-ins

# Sample dataset
sales_data = [
    {'date': '2024-01-01', 'product': 'laptop', 'quantity': 5, 'price': 999.99},
    {'date': '2024-01-02', 'product': 'mouse', 'quantity': 20, 'price': 25.50},
    {'date': '2024-01-03', 'product': 'keyboard', 'quantity': 15, 'price': 75.00},
    {'date': '2024-01-04', 'product': 'laptop', 'quantity': 3, 'price': 999.99},
    {'date': '2024-01-05', 'product': 'monitor', 'quantity': 8, 'price': 299.99}
]

# Basic data processing operations
print("Original data:")
for record in sales_data[:3]:
    print(f"  {record}")

# Calculate total revenue per item
processed_data = [
    {
        **record,
        'total_revenue': record['quantity'] * record['price']
    }
    for record in sales_data
]

print("\nProcessed data with total revenue:")
for record in processed_data[:3]:
    print(f"  {record['date']}: {record['product']} - ${record['total_revenue']:.2f}")

# Aggregate by product
product_totals = {}
for record in processed_data:
    product = record['product']
    if product not in product_totals:
        product_totals[product] = {'quantity': 0, 'revenue': 0}
    
    product_totals[product]['quantity'] += record['quantity']
    product_totals[product]['revenue'] += record['total_revenue']

print("\nProduct totals:")
for product, totals in product_totals.items():
    print(f"  {product}: {totals['quantity']} units, ${totals['revenue']:.2f}")

Output:

Click "Run Code" to see the output

Data Loading and Parsing #

Load data from various sources:

🐍 Try it yourself

# Simulated data loading and parsing

# CSV-like data as string (simulating file reading)
csv_data = """name,age,city,salary
Alice Johnson,28,New York,85000
Bob Smith,35,London,75000
Charlie Brown,42,Tokyo,90000
Diana Davis,30,Paris,80000"""

# Parse CSV data manually
lines = csv_data.strip().split('\n')
headers = lines[0].split(',')

parsed_data = []
for line in lines[1:]:
    values = line.split(',')
    record = dict(zip(headers, values))
    
    # Type conversion
    try:
        record['age'] = int(record['age'])
        record['salary'] = int(record['salary'])
    except ValueError:
        print(f"Warning: Could not convert data for {record['name']}")
    
    parsed_data.append(record)

print("Parsed CSV data:")
for record in parsed_data:
    print(f"  {record}")

# JSON-like data processing
json_like_data = [
    '{"id": 1, "name": "Alice", "scores": [85, 90, 92]}',
    '{"id": 2, "name": "Bob", "scores": [78, 85, 88]}',
    '{"id": 3, "name": "Charlie", "scores": [92, 95, 89]}'
]

import json

student_data = []
for json_string in json_like_data:
    try:
        record = json.loads(json_string)
        # Calculate average score
        record['average'] = sum(record['scores']) / len(record['scores'])
        student_data.append(record)
    except json.JSONDecodeError:
        print(f"Error parsing: {json_string}")

print("\nProcessed JSON data:")
for student in student_data:
    print(f"  {student['name']}: Average {student['average']:.1f}")

Output:

Click "Run Code" to see the output

Working with CSV Data #

CSV Processing Patterns #

Advanced CSV data handling:

🐍 Try it yourself

# Advanced CSV processing patterns

# Simulated CSV with various data types and missing values
complex_csv = """product_id,name,category,price,stock,last_updated,is_active
1,Laptop Pro,Electronics,1299.99,15,2024-01-15,true
2,Wireless Mouse,Electronics,29.99,,2024-01-10,true
3,Office Chair,Furniture,199.50,8,2024-01-12,false
4,Coffee Mug,Kitchen,12.99,25,,true
5,Notebook,Office,4.99,100,2024-01-08,true"""

# Advanced CSV parser with type handling
def parse_csv_advanced(csv_text):
    lines = csv_text.strip().split('\n')
    headers = lines[0].split(',')
    
    parsed_records = []
    for line_num, line in enumerate(lines[1:], start=2):
        values = line.split(',')
        
        if len(values) != len(headers):
            print(f"Warning: Line {line_num} has incorrect number of columns")
            continue
        
        record = {}
        for header, value in zip(headers, values):
            # Type conversion based on header name and value
            if value == '':
                record[header] = None
            elif header.endswith('_id') or header == 'stock':
                try:
                    record[header] = int(value)
                except ValueError:
                    record[header] = None
            elif header == 'price':
                try:
                    record[header] = float(value)
                except ValueError:
                    record[header] = None
            elif header == 'is_active':
                record[header] = value.lower() == 'true'
            else:
                record[header] = value
        
        parsed_records.append(record)
    
    return parsed_records

# Process the CSV
products = parse_csv_advanced(complex_csv)

print("Processed product data:")
for product in products:
    print(f"  {product}")

# Data validation and cleaning
def validate_product_data(products):
    valid_products = []
    errors = []
    
    for i, product in enumerate(products):
        # Check required fields
        if not product.get('name'):
            errors.append(f"Product {i}: Missing name")
            continue
        
        if product.get('price') is None:
            errors.append(f"Product {i}: Missing price")
            continue
        
        if product.get('price', 0) <= 0:
            errors.append(f"Product {i}: Invalid price")
            continue
        
        # Set defaults for missing optional fields
        if product.get('stock') is None:
            product['stock'] = 0
        
        if product.get('last_updated') is None:
            product['last_updated'] = '2024-01-01'
        
        valid_products.append(product)
    
    return valid_products, errors

valid_products, validation_errors = validate_product_data(products)

print(f"\nValidation results: {len(valid_products)} valid, {len(validation_errors)} errors")
for error in validation_errors:
    print(f"  {error}")

Output:

Click "Run Code" to see the output

Data Cleaning Techniques #

Handling Missing Data #

Strategies for dealing with missing values:

🐍 Try it yourself

# Missing data handling strategies

# Dataset with missing values
survey_data = [
    {'name': 'Alice', 'age': 28, 'income': 85000, 'education': 'Bachelor'},
    {'name': 'Bob', 'age': None, 'income': 75000, 'education': 'Master'},
    {'name': 'Charlie', 'age': 42, 'income': None, 'education': 'PhD'},
    {'name': 'Diana', 'age': 30, 'income': 80000, 'education': None},
    {'name': 'Eve', 'age': 35, 'income': 95000, 'education': 'Bachelor'},
    {'name': None, 'age': 25, 'income': 60000, 'education': 'Bachelor'}
]

print("Original data with missing values:")
for i, record in enumerate(survey_data):
    print(f"  {i}: {record}")

# Strategy 1: Remove records with any missing values
complete_records = [
    record for record in survey_data
    if all(value is not None for value in record.values())
]

print(f"\nComplete records only: {len(complete_records)} out of {len(survey_data)}")

# Strategy 2: Fill missing values with defaults/averages
def fill_missing_values(data):
    # Calculate averages for numerical fields
    ages = [record['age'] for record in data if record['age'] is not None]
    incomes = [record['income'] for record in data if record['income'] is not None]
    
    avg_age = sum(ages) / len(ages) if ages else 0
    avg_income = sum(incomes) / len(incomes) if incomes else 0
    
    # Most common education level
    educations = [record['education'] for record in data if record['education'] is not None]
    most_common_education = max(set(educations), key=educations.count) if educations else 'Unknown'
    
    filled_data = []
    for record in data:
        new_record = record.copy()
        
        if new_record['name'] is None:
            new_record['name'] = 'Unknown'
        
        if new_record['age'] is None:
            new_record['age'] = int(avg_age)
        
        if new_record['income'] is None:
            new_record['income'] = int(avg_income)
        
        if new_record['education'] is None:
            new_record['education'] = most_common_education
        
        filled_data.append(new_record)
    
    return filled_data

filled_data = fill_missing_values(survey_data)

print("\nData with filled missing values:")
for i, record in enumerate(filled_data):
    print(f"  {i}: {record}")

# Strategy 3: Forward fill (use previous valid value)
def forward_fill(data, field):
    filled_data = data.copy()
    last_valid = None
    
    for record in filled_data:
        if record[field] is not None:
            last_valid = record[field]
        elif last_valid is not None:
            record[field] = last_valid
    
    return filled_data

# Example: forward fill education
ff_data = forward_fill(survey_data, 'education')
print("\nForward filled education field:")
for record in ff_data:
    print(f"  {record['name']}: {record['education']}")

Output:

Click "Run Code" to see the output

Data Validation and Quality Checks #

Implement comprehensive data validation:

🐍 Try it yourself

# Data validation and quality checks

# Employee dataset with potential quality issues
employee_data = [
    {'id': 1, 'name': 'Alice Johnson', 'email': 'alice@company.com', 'age': 28, 'salary': 85000},
    {'id': 2, 'name': 'Bob', 'email': 'bob@invalid', 'age': 150, 'salary': -5000},
    {'id': 3, 'name': '', 'email': 'charlie@company.com', 'age': 42, 'salary': 90000},
    {'id': 1, 'name': 'Diana Davis', 'email': 'diana@company.com', 'age': 30, 'salary': 80000},  # Duplicate ID
    {'id': 5, 'name': 'Eve O\'Connor', 'email': 'eve@company.com', 'age': 35, 'salary': 95000}
]

# Comprehensive validation function
def validate_employee_data(data):
    errors = []
    warnings = []
    valid_records = []
    seen_ids = set()
    
    for i, record in enumerate(data):
        record_errors = []
        record_warnings = []
        
        # Check ID uniqueness
        emp_id = record.get('id')
        if emp_id in seen_ids:
            record_errors.append(f"Duplicate ID: {emp_id}")
        else:
            seen_ids.add(emp_id)
        
        # Validate name
        name = record.get('name', '').strip()
        if not name:
            record_errors.append("Empty name")
        elif len(name) < 2:
            record_warnings.append("Very short name")
        
        # Validate email
        email = record.get('email', '')
        if not email or '@' not in email or '.' not in email.split('@')[-1]:
            record_errors.append("Invalid email format")
        
        # Validate age
        age = record.get('age')
        if not isinstance(age, int) or age < 16 or age > 100:
            record_errors.append("Invalid age (must be 16-100)")
        
        # Validate salary
        salary = record.get('salary')
        if not isinstance(salary, (int, float)) or salary < 0:
            record_errors.append("Invalid salary (must be positive)")
        elif salary > 200000:
            record_warnings.append("Unusually high salary")
        
        # Collect errors and warnings
        if record_errors:
            errors.append(f"Record {i}: {', '.join(record_errors)}")
        
        if record_warnings:
            warnings.append(f"Record {i}: {', '.join(record_warnings)}")
        
        # Add to valid records if no errors
        if not record_errors:
            valid_records.append(record)
    
    return valid_records, errors, warnings

# Run validation
valid_employees, errors, warnings = validate_employee_data(employee_data)

print("Data validation results:")
print(f"  Valid records: {len(valid_employees)}")
print(f"  Errors: {len(errors)}")
print(f"  Warnings: {len(warnings)}")

if errors:
    print("\nErrors found:")
    for error in errors:
        print(f"  {error}")

if warnings:
    print("\nWarnings:")
    for warning in warnings:
        print(f"  {warning}")

print("\nValid employee records:")
for emp in valid_employees:
    print(f"  {emp['id']}: {emp['name']} ({emp['age']}) - ${emp['salary']:,}")

Output:

Click "Run Code" to see the output

Data Transformation #

Reshaping and Restructuring Data #

Transform data structures for analysis:

🐍 Try it yourself

# Data reshaping and restructuring

# Wide format data (multiple measurements per record)
student_scores = [
    {'name': 'Alice', 'math': 85, 'science': 90, 'english': 88},
    {'name': 'Bob', 'math': 78, 'science': 85, 'english': 82},
    {'name': 'Charlie', 'math': 92, 'science': 88, 'english': 90}
]

# Transform to long format (one measurement per record)
def wide_to_long(data, id_col, value_cols):
    long_data = []
    for record in data:
        for col in value_cols:
            long_record = {
                id_col: record[id_col],
                'subject': col,
                'score': record[col]
            }
            long_data.append(long_record)
    return long_data

long_format = wide_to_long(student_scores, 'name', ['math', 'science', 'english'])

print("Long format data:")
for record in long_format[:6]:  # Show first 6 records
    print(f"  {record}")

# Transform back to wide format
def long_to_wide(data, id_col, category_col, value_col):
    wide_data = {}
    for record in data:
        key = record[id_col]
        if key not in wide_data:
            wide_data[key] = {id_col: key}
        
        wide_data[key][record[category_col]] = record[value_col]
    
    return list(wide_data.values())

wide_format = long_to_wide(long_format, 'name', 'subject', 'score')

print("\nTransformed back to wide format:")
for record in wide_format:
    print(f"  {record}")

# Pivot table creation
def create_pivot_table(data, index_col, column_col, value_col, agg_func=sum):
    pivot_data = {}
    
    for record in data:
        row_key = record[index_col]
        col_key = record[column_col]
        value = record[value_col]
        
        if row_key not in pivot_data:
            pivot_data[row_key] = {}
        
        if col_key not in pivot_data[row_key]:
            pivot_data[row_key][col_key] = []
        
        pivot_data[row_key][col_key].append(value)
    
    # Apply aggregation function
    for row_key in pivot_data:
        for col_key in pivot_data[row_key]:
            pivot_data[row_key][col_key] = agg_func(pivot_data[row_key][col_key])
    
    return pivot_data

# Create pivot table: average scores by student and subject
pivot_table = create_pivot_table(long_format, 'name', 'subject', 'score', lambda x: sum(x)/len(x))

print("\nPivot table (average scores):")
for student, subjects in pivot_table.items():
    print(f"  {student}: {subjects}")

Output:

Click "Run Code" to see the output

Data Normalization and Standardization #

Normalize data for analysis:

🐍 Try it yourself

# Data normalization and standardization
import math

# Sample numerical data
measurements = [
    {'id': 1, 'height': 170, 'weight': 65, 'age': 25},
    {'id': 2, 'height': 180, 'weight': 80, 'age': 30},
    {'id': 3, 'height': 165, 'weight': 55, 'age': 22},
    {'id': 4, 'height': 175, 'weight': 70, 'age': 35},
    {'id': 5, 'height': 185, 'weight': 85, 'age': 28}
]

# Min-Max normalization (0-1 scale)
def min_max_normalize(data, field):
    values = [record[field] for record in data]
    min_val = min(values)
    max_val = max(values)
    range_val = max_val - min_val
    
    normalized_data = []
    for record in data:
        new_record = record.copy()
        if range_val > 0:
            new_record[f'{field}_normalized'] = (record[field] - min_val) / range_val
        else:
            new_record[f'{field}_normalized'] = 0
        normalized_data.append(new_record)
    
    return normalized_data

# Z-score standardization (mean=0, std=1)
def z_score_standardize(data, field):
    values = [record[field] for record in data]
    mean_val = sum(values) / len(values)
    variance = sum((x - mean_val) ** 2 for x in values) / len(values)
    std_val = math.sqrt(variance)
    
    standardized_data = []
    for record in data:
        new_record = record.copy()
        if std_val > 0:
            new_record[f'{field}_standardized'] = (record[field] - mean_val) / std_val
        else:
            new_record[f'{field}_standardized'] = 0
        standardized_data.append(new_record)
    
    return standardized_data

# Apply normalization and standardization
normalized_height = min_max_normalize(measurements, 'height')
standardized_weight = z_score_standardize(normalized_height, 'weight')

print("Normalized and standardized data:")
for record in standardized_weight:
    print(f"  ID {record['id']}: height={record['height']} (norm: {record['height_normalized']:.3f}), "
          f"weight={record['weight']} (std: {record['weight_standardized']:.3f})")

# Robust scaling (using median and IQR)
def robust_scale(data, field):
    values = sorted([record[field] for record in data])
    n = len(values)
    
    # Calculate median and IQR
    median = values[n // 2] if n % 2 == 1 else (values[n // 2 - 1] + values[n // 2]) / 2
    q1_idx = n // 4
    q3_idx = 3 * n // 4
    q1 = values[q1_idx]
    q3 = values[q3_idx]
    iqr = q3 - q1
    
    scaled_data = []
    for record in data:
        new_record = record.copy()
        if iqr > 0:
            new_record[f'{field}_robust_scaled'] = (record[field] - median) / iqr
        else:
            new_record[f'{field}_robust_scaled'] = 0
        scaled_data.append(new_record)
    
    return scaled_data

robust_scaled = robust_scale(measurements, 'age')

print("\nRobust scaled ages:")
for record in robust_scaled:
    print(f"  ID {record['id']}: age={record['age']} (robust: {record['age_robust_scaled']:.3f})")

Output:

Click "Run Code" to see the output

Aggregation and Grouping #

Group-by Operations #

Implement SQL-like GROUP BY functionality:

🐍 Try it yourself

# Group-by operations for data aggregation

# Sales transaction data
transactions = [
    {'date': '2024-01-01', 'region': 'North', 'product': 'laptop', 'sales': 10, 'revenue': 15000},
    {'date': '2024-01-01', 'region': 'South', 'product': 'laptop', 'sales': 8, 'revenue': 12000},
    {'date': '2024-01-01', 'region': 'North', 'product': 'mouse', 'sales': 50, 'revenue': 1250},
    {'date': '2024-01-02', 'region': 'North', 'product': 'laptop', 'sales': 12, 'revenue': 18000},
    {'date': '2024-01-02', 'region': 'South', 'product': 'mouse', 'sales': 40, 'revenue': 1000},
    {'date': '2024-01-02', 'region': 'East', 'product': 'laptop', 'sales': 6, 'revenue': 9000}
]

# Generic group-by function
def group_by(data, group_fields, agg_fields):
    groups = {}
    
    for record in data:
        # Create group key
        if isinstance(group_fields, str):
            group_key = record[group_fields]
        else:
            group_key = tuple(record[field] for field in group_fields)
        
        if group_key not in groups:
            groups[group_key] = []
        
        groups[group_key].append(record)
    
    # Aggregate within groups
    aggregated = {}
    for group_key, group_records in groups.items():
        agg_result = {}
        
        # Add group key fields to result
        if isinstance(group_fields, str):
            agg_result[group_fields] = group_key
        else:
            for i, field in enumerate(group_fields):
                agg_result[field] = group_key[i]
        
        # Calculate aggregations
        for agg_field, agg_func in agg_fields.items():
            values = [record[agg_field] for record in group_records]
            if agg_func == 'sum':
                agg_result[f'{agg_field}_sum'] = sum(values)
            elif agg_func == 'avg':
                agg_result[f'{agg_field}_avg'] = sum(values) / len(values)
            elif agg_func == 'count':
                agg_result[f'{agg_field}_count'] = len(values)
            elif agg_func == 'min':
                agg_result[f'{agg_field}_min'] = min(values)
            elif agg_func == 'max':
                agg_result[f'{agg_field}_max'] = max(values)
        
        aggregated[group_key] = agg_result
    
    return list(aggregated.values())

# Example 1: Group by region
region_summary = group_by(
    transactions,
    'region',
    {'sales': 'sum', 'revenue': 'sum', 'sales': 'avg'}
)

print("Sales summary by region:")
for summary in region_summary:
    print(f"  {summary}")

# Example 2: Group by multiple fields
product_region_summary = group_by(
    transactions,
    ['product', 'region'],
    {'sales': 'sum', 'revenue': 'avg'}
)

print("\nSales summary by product and region:")
for summary in product_region_summary:
    print(f"  {summary['product']} in {summary['region']}: "
          f"{summary['sales_sum']} sales, ${summary['revenue_avg']:.0f} avg revenue")

# Example 3: Daily aggregation
daily_summary = group_by(
    transactions,
    'date',
    {'sales': 'sum', 'revenue': 'sum', 'product': 'count'}
)

print("\nDaily sales summary:")
for summary in sorted(daily_summary, key=lambda x: x['date']):
    print(f"  {summary['date']}: {summary['sales_sum']} total sales, "
          f"${summary['revenue_sum']:,} revenue, {summary['product_count']} transactions")

Output:

Click "Run Code" to see the output

Time Series Processing #

Date and Time Handling #

Process time-based data effectively:

🐍 Try it yourself

# Time series data processing
from datetime import datetime, timedelta

# Sample time series data
time_series_data = [
    {'timestamp': '2024-01-01 09:00:00', 'sensor_id': 'A1', 'temperature': 22.5, 'humidity': 45},
    {'timestamp': '2024-01-01 09:15:00', 'sensor_id': 'A1', 'temperature': 23.1, 'humidity': 46},
    {'timestamp': '2024-01-01 09:30:00', 'sensor_id': 'A1', 'temperature': 23.8, 'humidity': 44},
    {'timestamp': '2024-01-01 09:00:00', 'sensor_id': 'B1', 'temperature': 21.2, 'humidity': 50},
    {'timestamp': '2024-01-01 09:15:00', 'sensor_id': 'B1', 'temperature': 21.8, 'humidity': 49},
    {'timestamp': '2024-01-01 09:30:00', 'sensor_id': 'B1', 'temperature': 22.1, 'humidity': 48}
]

# Parse timestamps
def parse_timestamps(data):
    parsed_data = []
    for record in data:
        new_record = record.copy()
        new_record['datetime'] = datetime.strptime(record['timestamp'], '%Y-%m-%d %H:%M:%S')
        parsed_data.append(new_record)
    return parsed_data

parsed_data = parse_timestamps(time_series_data)

# Time-based filtering
def filter_by_time_range(data, start_time, end_time):
    start_dt = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S')
    end_dt = datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
    
    return [
        record for record in data
        if start_dt <= record['datetime'] <= end_dt
    ]

# Filter for specific time range
filtered_data = filter_by_time_range(
    parsed_data,
    '2024-01-01 09:10:00',
    '2024-01-01 09:25:00'
)

print("Filtered time series data:")
for record in filtered_data:
    print(f"  {record['timestamp']} - {record['sensor_id']}: {record['temperature']}°C")

# Resample data (aggregate by time intervals)
def resample_data(data, interval_minutes):
    # Group by time intervals
    intervals = {}
    
    for record in data:
        dt = record['datetime']
        # Round down to nearest interval
        minutes_since_midnight = dt.hour * 60 + dt.minute
        interval_start = (minutes_since_midnight // interval_minutes) * interval_minutes
        interval_key = dt.replace(hour=interval_start // 60, minute=interval_start % 60, second=0, microsecond=0)
        
        key = (interval_key, record['sensor_id'])
        if key not in intervals:
            intervals[key] = []
        
        intervals[key].append(record)
    
    # Aggregate within intervals
    resampled = []
    for (interval_time, sensor_id), records in intervals.items():
        agg_record = {
            'interval_start': interval_time.strftime('%Y-%m-%d %H:%M:%S'),
            'sensor_id': sensor_id,
            'temperature_avg': sum(r['temperature'] for r in records) / len(records),
            'temperature_min': min(r['temperature'] for r in records),
            'temperature_max': max(r['temperature'] for r in records),
            'humidity_avg': sum(r['humidity'] for r in records) / len(records),
            'count': len(records)
        }
        resampled.append(agg_record)
    
    return sorted(resampled, key=lambda x: (x['interval_start'], x['sensor_id']))

# Resample to 30-minute intervals
resampled = resample_data(parsed_data, 30)

print("\nResampled data (30-minute intervals):")
for record in resampled:
    print(f"  {record['interval_start']} - {record['sensor_id']}: "
          f"Temp {record['temperature_avg']:.1f}°C (range: {record['temperature_min']:.1f}-{record['temperature_max']:.1f}), "
          f"Humidity {record['humidity_avg']:.1f}%")

Output:

Click "Run Code" to see the output

Performance Optimization #

Efficient Data Processing Patterns #

Optimize data processing for large datasets:

🐍 Try it yourself

# Performance optimization techniques
import time

# Generate sample large dataset
def generate_large_dataset(size):
    import random
    
    products = ['laptop', 'mouse', 'keyboard', 'monitor', 'tablet']
    regions = ['North', 'South', 'East', 'West']
    
    dataset = []
    for i in range(size):
        record = {
            'id': i + 1,
            'product': random.choice(products),
            'region': random.choice(regions),
            'sales': random.randint(1, 100),
            'revenue': random.uniform(100, 10000)
        }
        dataset.append(record)
    
    return dataset

# Test dataset
large_dataset = generate_large_dataset(10000)

# Method 1: Traditional loop (slower)
def process_traditional(data):
    result = {}
    for record in data:
        key = record['product']
        if key not in result:
            result[key] = {'total_sales': 0, 'total_revenue': 0, 'count': 0}
        
        result[key]['total_sales'] += record['sales']
        result[key]['total_revenue'] += record['revenue']
        result[key]['count'] += 1
    
    return result

# Method 2: Dictionary comprehension with grouping (faster)
def process_optimized(data):
    # Group by product first
    groups = {}
    for record in data:
        product = record['product']
        if product not in groups:
            groups[product] = []
        groups[product].append(record)
    
    # Aggregate using comprehensions
    result = {
        product: {
            'total_sales': sum(r['sales'] for r in records),
            'total_revenue': sum(r['revenue'] for r in records),
            'count': len(records)
        }
        for product, records in groups.items()
    }
    
    return result

# Performance comparison
start_time = time.perf_counter()
result1 = process_traditional(large_dataset)
time1 = time.perf_counter() - start_time

start_time = time.perf_counter()
result2 = process_optimized(large_dataset)
time2 = time.perf_counter() - start_time

print(f"Processing {len(large_dataset)} records:")
print(f"  Traditional method: {time1:.4f} seconds")
print(f"  Optimized method: {time2:.4f} seconds")
print(f"  Speedup: {time1/time2:.2f}x")

# Verify results are identical
print(f"  Results identical: {result1 == result2}")

# Memory-efficient processing for very large datasets
def process_streaming(data_generator, chunk_size=1000):
    """Process data in chunks to save memory"""
    aggregated = {}
    processed_count = 0
    
    chunk = []
    for record in data_generator:
        chunk.append(record)
        
        if len(chunk) >= chunk_size:
            # Process chunk
            chunk_result = process_optimized(chunk)
            
            # Merge with overall results
            for product, stats in chunk_result.items():
                if product not in aggregated:
                    aggregated[product] = {'total_sales': 0, 'total_revenue': 0, 'count': 0}
                
                aggregated[product]['total_sales'] += stats['total_sales']
                aggregated[product]['total_revenue'] += stats['total_revenue']
                aggregated[product]['count'] += stats['count']
            
            processed_count += len(chunk)
            chunk = []
            
            if processed_count % 5000 == 0:
                print(f"  Processed {processed_count} records...")
    
    # Process remaining records
    if chunk:
        chunk_result = process_optimized(chunk)
        for product, stats in chunk_result.items():
            if product not in aggregated:
                aggregated[product] = {'total_sales': 0, 'total_revenue': 0, 'count': 0}
            
            aggregated[product]['total_sales'] += stats['total_sales']
            aggregated[product]['total_revenue'] += stats['total_revenue']
            aggregated[product]['count'] += stats['count']
        
        processed_count += len(chunk)
    
    print(f"  Total processed: {processed_count} records")
    return aggregated

# Simulate streaming processing
print("\nStreaming processing simulation:")
streaming_result = process_streaming(iter(large_dataset), chunk_size=2000)

print("Final aggregated results:")
for product, stats in streaming_result.items():
    avg_revenue = stats['total_revenue'] / stats['count']
    print(f"  {product}: {stats['total_sales']} sales, ${avg_revenue:.2f} avg revenue")

Output:

Click "Run Code" to see the output

Conclusion #

Python data processing involves:

Loading and parsing data from various sources
Cleaning and validating data quality
Transforming and reshaping data structures
Aggregating and grouping for analysis
Handling time series data effectively
Optimizing performance for large datasets

Key takeaways:

Use appropriate data structures for your use case
Implement comprehensive data validation
Choose the right aggregation and grouping strategies
Optimize for performance when dealing with large datasets
Consider memory usage and processing time trade-offs

Master these techniques to become proficient in Python data processing!

PyGuide

PyGuide

Python Data Processing: Complete Guide with Pandas and NumPy

Table of Contents #

Data Processing Fundamentals #

Basic Data Structures #

🐍 Try it yourself

Data Loading and Parsing #

🐍 Try it yourself

Working with CSV Data #

CSV Processing Patterns #

🐍 Try it yourself

Data Cleaning Techniques #

Handling Missing Data #

🐍 Try it yourself

Data Validation and Quality Checks #

🐍 Try it yourself

Data Transformation #

Reshaping and Restructuring Data #

🐍 Try it yourself

Data Normalization and Standardization #

🐍 Try it yourself

Aggregation and Grouping #

Group-by Operations #

🐍 Try it yourself

Time Series Processing #

Date and Time Handling #

🐍 Try it yourself

Performance Optimization #

Efficient Data Processing Patterns #

🐍 Try it yourself

Conclusion #