My Practical Insights of Using Pandas Library

import pandas as pd import numpy as np import time # Memory-efficient data loading strategies # 1. Use appropriate data types to reduce memory def optimize_dtypes(df): """Automatically optimize data types to reduce memory usage""" original_memory = df.memory_usage(deep=True).sum() # Optimize integers for col in df.select_dtypes(include=['int64']).columns: if df[col].min() >= -128 and df[col].max() <= 127: df[col] = df[col].astype('int8') elif df[col].min() >= -32768 and df[col].max() <= 32767: df[col] = df[col].astype('int16') elif df[col].min() >= -2147483648 and df[col].max() <= 2147483647: df[col] = df[col].astype('int32') # Optimize floats for col in df.select_dtypes(include=['float64']).columns: df[col] = pd.to_numeric(df[col], downcast='float') # Convert to category for strings with low cardinality for col in df.select_dtypes(include=['object']).columns: if df[col].nunique() / len(df) < 0.5: # Less than 50% unique values df[col] = df[col].astype('category') optimized_memory = df.memory_usage(deep=True).sum() print(f"Memory usage reduced from {original_memory/1024**2:.1f}MB to {optimized_memory/1024**2:.1f}MB") print(f"Reduction: {(1 - optimized_memory/original_memory)*100:.1f}%") return df # 2. Chunked reading for large files def process_large_file(filename, chunksize=10000): """Process large CSV files in chunks to avoid memory issues""" results = [] for chunk in pd.read_csv(filename, chunksize=chunksize): # Process each chunk processed_chunk = chunk.groupby('category').agg({ 'value': ['sum', 'mean', 'count'] }).reset_index() results.append(processed_chunk) # Combine all chunks final_result = pd.concat(results, ignore_index=True) return final_result.groupby('category').sum().reset_index() # 3. Selective column reading columns_to_read = ['user_id', 'timestamp', 'event_type', 'value'] df_selective = pd.read_csv('large_dataset.csv', usecols=columns_to_read) # 4. Date parsing during read df_with_dates = pd.read_csv('time_series.csv', parse_dates=['timestamp'], date_parser=pd.to_datetime) print("Demonstrating memory optimization techniques...") # Create sample data for demonstration sample_data = { 'category': np.random.choice(['A', 'B', 'C'], 10000), 'large_int': np.random.randint(1, 100, 10000), 'float_col': np.random.randn(10000), 'text_col': np.random.choice(['low', 'medium', 'high'], 10000) } df = pd.DataFrame(sample_data) print(f"Original DataFrame memory usage: {df.memory_usage(deep=True).sum()/1024**2:.1f}MB") # Apply optimizations df_optimized = optimize_dtypes(df.copy()) print(f"Data types after optimization: {dict(df_optimized.dtypes)}")

# Advanced indexing and selection patterns # Create sample time series data dates = pd.date_range('2023-01-01', periods=100000, freq='1min') df_ts = pd.DataFrame({ 'timestamp': dates, 'sensor_id': np.random.choice(['sensor_1', 'sensor_2', 'sensor_3'], 100000), 'temperature': 20 + 5 * np.random.randn(100000), 'humidity': 50 + 10 * np.random.randn(100000), 'status': np.random.choice(['online', 'offline'], 100000, p=[0.95, 0.05]) }) # 1. Set proper index for time series operations df_ts.set_index('timestamp', inplace=True) print(f"DataFrame shape: {df_ts.shape}") # 2. Boolean indexing with multiple conditions start_time = time.time() # Inefficient - multiple separate conditions condition1 = df_ts['temperature'] > 25 condition2 = df_ts['humidity'] < 60 condition3 = df_ts['status'] == 'online' filtered_slow = df_ts[condition1 & condition2 & condition3] slow_time = time.time() - start_time # Efficient - combined query start_time = time.time() filtered_fast = df_ts.query("temperature > 25 and humidity < 60 and status == 'online'") fast_time = time.time() - start_time print(f"Traditional filtering: {slow_time:.4f}s") print(f"Query method: {fast_time:.4f}s") print(f"Speed improvement: {slow_time/fast_time:.1f}x") # 3. Advanced time-based indexing # Select data for specific time periods march_data = df_ts['2023-03'] # All March data morning_data = df_ts.between_time('06:00', '12:00') # Morning hours only recent_data = df_ts.last('7D') # Last 7 days print(f"March data points: {len(march_data)}") print(f"Morning data points: {len(morning_data)}") # 4. MultiIndex operations for complex data structures df_multi = df_ts.reset_index().set_index(['sensor_id', 'timestamp']) print(f"MultiIndex levels: {df_multi.index.names}") # Efficient cross-section selection sensor_1_data = df_multi.xs('sensor_1', level='sensor_id') print(f"Sensor 1 data points: {len(sensor_1_data)}") # 5. Using .loc and .iloc effectively # Select specific sensors and time range specific_selection = df_ts.loc[ (df_ts.index >= '2023-01-15') & (df_ts.index <= '2023-01-20'), ['temperature', 'humidity'] ] print(f"Specific selection shape: {specific_selection.shape}") # 6. Conditional selection with np.where for performance df_ts['temperature_category'] = np.where( df_ts['temperature'] > 25, 'hot', np.where(df_ts['temperature'] > 15, 'warm', 'cold') ) category_counts = df_ts['temperature_category'].value_counts() print(f"Temperature categories: {dict(category_counts)}")

# Advanced GroupBy operations and aggregations # Create sample sales data np.random.seed(42) sales_data = pd.DataFrame({ 'date': pd.date_range('2023-01-01', periods=50000, freq='1H'), 'product_id': np.random.choice(['P001', 'P002', 'P003', 'P004', 'P005'], 50000), 'category': np.random.choice(['Electronics', 'Clothing', 'Books'], 50000), 'sales_amount': np.random.exponential(100, 50000), 'quantity': np.random.poisson(3, 50000) + 1, 'customer_type': np.random.choice(['new', 'returning', 'premium'], 50000, p=[0.3, 0.5, 0.2]) }) print(f"Sales data shape: {sales_data.shape}") # 1. Multiple aggregations with named outputs start_time = time.time() sales_summary = sales_data.groupby(['category', 'customer_type']).agg({ 'sales_amount': ['sum', 'mean', 'std', 'count'], 'quantity': ['sum', 'mean'], 'product_id': 'nunique' }).round(2) # Flatten column names for easier access sales_summary.columns = ['_'.join(col).strip() for col in sales_summary.columns.values] print(f"Aggregation completed in: {time.time() - start_time:.4f}s") print(f"Summary shape: {sales_summary.shape}") # 2. Custom aggregation functions def sales_metrics(series): """Custom aggregation function for sales analysis""" return pd.Series({ 'total': series.sum(), 'avg': series.mean(), 'median': series.median(), 'top_10_pct': series.quantile(0.9), 'cv': series.std() / series.mean() if series.mean() > 0 else 0 }) custom_agg = sales_data.groupby('category')['sales_amount'].apply(sales_metrics) print("Custom aggregation results:") print(custom_agg.round(2)) # 3. Rolling operations within groups sales_data['date'] = pd.to_datetime(sales_data['date']) sales_data = sales_data.sort_values(['category', 'date']) # Calculate 7-day rolling average within each category sales_data['rolling_7d_avg'] = ( sales_data.groupby('category')['sales_amount'] .rolling(window=168, min_periods=1) # 168 hours = 7 days .mean() .reset_index(level=0, drop=True) ) print("Rolling averages calculated") # 4. Advanced transformation patterns # Calculate percentage of total sales by category sales_data['category_sales_pct'] = ( sales_data['sales_amount'] / sales_data.groupby('category')['sales_amount'].transform('sum') ) * 100 # Rank products within each category sales_data['product_rank'] = ( sales_data.groupby('category')['sales_amount'] .rank(method='dense', ascending=False) ) # 5. Efficient filtering with groupby # Find top-performing products (top 10% by sales amount) top_products = ( sales_data.groupby('product_id')['sales_amount'] .sum() .nlargest(int(len(sales_data['product_id'].unique()) * 0.1)) ) print(f"Top products identified: {len(top_products)}") # 6. GroupBy with multiple operations in one pass daily_insights = sales_data.groupby(sales_data['date'].dt.date).agg({ 'sales_amount': ['sum', 'count'], 'quantity': 'sum', 'customer_type': lambda x: x.value_counts().index[0], # Most common customer type 'product_id': 'nunique' }).round(2) daily_insights.columns = ['total_sales', 'num_transactions', 'total_quantity', 'dominant_customer_type', 'unique_products'] print(f"Daily insights calculated for {len(daily_insights)} days") print(daily_insights.head()) # 7. Memory-efficient groupby operations for large datasets def efficient_groupby_agg(df, group_cols, agg_dict, chunksize=10000): """Memory-efficient groupby for large datasets""" results = [] for i in range(0, len(df), chunksize): chunk = df.iloc[i:i+chunksize] chunk_result = chunk.groupby(group_cols).agg(agg_dict) results.append(chunk_result) # Combine results combined = pd.concat(results) return combined.groupby(level=group_cols).sum() # Demonstrate efficient groupby efficient_result = efficient_groupby_agg( sales_data, ['category'], {'sales_amount': 'sum', 'quantity': 'sum'}, chunksize=10000 ) print("Efficient groupby completed") print(efficient_result)

# Advanced time series operations and date handling # Create comprehensive time series dataset date_range = pd.date_range('2022-01-01', '2024-01-01', freq='1H') ts_data = pd.DataFrame({ 'timestamp': date_range, 'value': 100 + 10 * np.sin(np.arange(len(date_range)) * 2 * np.pi / 24) + np.random.randn(len(date_range)) * 5, 'category': np.random.choice(['A', 'B', 'C'], len(date_range)), 'is_weekend': date_range.weekday >= 5, 'hour': date_range.hour, 'month': date_range.month }) ts_data.set_index('timestamp', inplace=True) print(f"Time series data shape: {ts_data.shape}") print(f"Date range: {ts_data.index.min()} to {ts_data.index.max()}") # 1. Resampling with custom aggregations # Resample to daily data with multiple aggregations daily_stats = ts_data.resample('D').agg({ 'value': ['mean', 'min', 'max', 'std'], 'category': lambda x: x.mode().iloc[0] if len(x.mode()) > 0 else 'Unknown', 'is_weekend': 'first' }) daily_stats.columns = ['avg_value', 'min_value', 'max_value', 'std_value', 'mode_category', 'is_weekend'] print(f"Daily aggregation shape: {daily_stats.shape}") # 2. Time-based feature engineering ts_data['hour_sin'] = np.sin(2 * np.pi * ts_data['hour'] / 24) ts_data['hour_cos'] = np.cos(2 * np.pi * ts_data['hour'] / 24) ts_data['month_sin'] = np.sin(2 * np.pi * ts_data['month'] / 12) ts_data['month_cos'] = np.cos(2 * np.pi * ts_data['month'] / 12) print("Cyclical features created for hour and month") # 3. Advanced rolling operations # Multiple rolling window calculations rolling_stats = ts_data['value'].rolling(window=24).agg({ 'mean': 'mean', 'std': 'std', 'min': 'min', 'max': 'max' }) rolling_stats.columns = ['rolling_24h_mean', 'rolling_24h_std', 'rolling_24h_min', 'rolling_24h_max'] ts_data = pd.concat([ts_data, rolling_stats], axis=1) # 4. Lag features for time series modeling for lag in [1, 6, 24, 168]: # 1h, 6h, 1d, 1w lags ts_data[f'value_lag_{lag}'] = ts_data['value'].shift(lag) print("Lag features created") # 5. Seasonal decomposition patterns def calculate_seasonal_stats(df, value_col, period='D'): """Calculate seasonal statistics""" if period == 'D': grouper = df.index.hour period_name = 'hour' elif period == 'W': grouper = df.index.dayofweek period_name = 'dayofweek' elif period == 'M': grouper = df.index.day period_name = 'day' else: grouper = df.index.month period_name = 'month' seasonal_stats = df.groupby(grouper)[value_col].agg(['mean', 'std']) seasonal_stats.index.name = period_name return seasonal_stats hourly_patterns = calculate_seasonal_stats(ts_data, 'value', 'D') weekly_patterns = calculate_seasonal_stats(ts_data, 'value', 'W') print("Seasonal patterns calculated") print("Hourly patterns (first 5):") print(hourly_patterns.head()) # 6. Time series outlier detection def detect_time_series_outliers(series, window=24, threshold=3): """Detect outliers using rolling statistics""" rolling_mean = series.rolling(window=window, center=True).mean() rolling_std = series.rolling(window=window, center=True).std() z_scores = np.abs((series - rolling_mean) / rolling_std) outliers = z_scores > threshold return outliers ts_data['is_outlier'] = detect_time_series_outliers(ts_data['value']) outlier_count = ts_data['is_outlier'].sum() print(f"Outliers detected: {outlier_count} ({outlier_count/len(ts_data)*100:.2f}%)") # 7. Business day operations # Filter only business hours (9 AM to 5 PM, weekdays) business_hours = ts_data[ (ts_data.index.hour >= 9) & (ts_data.index.hour <= 17) & (ts_data.index.weekday < 5) ] print(f"Business hours data points: {len(business_hours)}") # 8. Time zone handling # Convert to different time zones ts_data_utc = ts_data.copy() ts_data_utc.index = ts_data_utc.index.tz_localize('UTC') ts_data_ny = ts_data_utc.tz_convert('America/New_York') ts_data_tokyo = ts_data_utc.tz_convert('Asia/Tokyo') print("Time zone conversions completed") # 9. Gap detection and filling def detect_time_gaps(df, expected_freq='1H'): """Detect gaps in time series data""" expected_index = pd.date_range( start=df.index.min(), end=df.index.max(), freq=expected_freq ) missing_dates = expected_index.difference(df.index) return missing_dates # Artificially create some gaps ts_sample = ts_data.iloc[::100].copy() # Sample every 100th point gaps = detect_time_gaps(ts_sample) print(f"Time gaps detected: {len(gaps)}") # Fill gaps using interpolation ts_filled = ts_sample.reindex( pd.date_range(ts_sample.index.min(), ts_sample.index.max(), freq='1H') ) ts_filled['value'] = ts_filled['value'].interpolate(method='time') print("Time gaps filled using interpolation")

# Advanced data transformation and cleaning strategies # Create messy dataset to demonstrate cleaning techniques np.random.seed(42) messy_data = pd.DataFrame({ 'customer_id': ['CUST_' + str(i).zfill(4) for i in range(1000)], 'name': ['Customer ' + str(i) for i in range(1000)], 'email': [f'customer{i}@example.com' if i % 10 != 0 else np.nan for i in range(1000)], 'age': np.random.randint(18, 80, 1000), 'income': np.random.exponential(50000, 1000), 'phone': [f'555-{np.random.randint(1000000, 9999999)}' if i % 15 != 0 else '' for i in range(1000)], 'registration_date': pd.date_range('2020-01-01', periods=1000, freq='1D') + pd.to_timedelta(np.random.randint(0, 24*60, 1000), unit='m'), 'category': np.random.choice(['Premium', 'Standard', 'Basic', 'premium', 'STANDARD', ''], 1000, p=[0.2, 0.3, 0.2, 0.1, 0.1, 0.1]) }) # Introduce some outliers and inconsistencies messy_data.loc[np.random.choice(1000, 50), 'age'] = np.random.randint(150, 200, 50) # Impossible ages messy_data.loc[np.random.choice(1000, 30), 'income'] = np.random.randint(1000000, 10000000, 30) # Extreme incomes messy_data.loc[np.random.choice(1000, 20), 'name'] = np.nan # Missing names print(f"Original messy data shape: {messy_data.shape}") print("Data quality issues:") print(messy_data.isnull().sum()) # 1. Comprehensive data profiling def data_quality_report(df): """Generate comprehensive data quality report""" report = pd.DataFrame({ 'dtype': df.dtypes, 'missing_count': df.isnull().sum(), 'missing_percentage': (df.isnull().sum() / len(df)) * 100, 'unique_count': df.nunique(), 'unique_percentage': (df.nunique() / len(df)) * 100 }) # Add memory usage report['memory_usage_mb'] = df.memory_usage(deep=True) / 1024**2 # Add additional stats for numeric columns numeric_cols = df.select_dtypes(include=[np.number]).columns for col in numeric_cols: report.loc[col, 'mean'] = df[col].mean() report.loc[col, 'std'] = df[col].std() report.loc[col, 'min'] = df[col].min() report.loc[col, 'max'] = df[col].max() report.loc[col, 'outlier_count'] = len(df[(df[col] < df[col].quantile(0.01)) | (df[col] > df[col].quantile(0.99))]) return report quality_report = data_quality_report(messy_data) print("\nData Quality Report:") print(quality_report.round(2)) # 2. Advanced missing value handling def handle_missing_values(df): """Advanced missing value imputation strategies""" df_clean = df.copy() # Email: Fill with pattern-based generation missing_emails = df_clean['email'].isnull() df_clean.loc[missing_emails, 'email'] = ( df_clean.loc[missing_emails, 'name'].str.lower().str.replace(' ', '') + '@generated.com' ) # Phone: Fill with 'Unknown' for empty strings df_clean['phone'] = df_clean['phone'].replace('', 'Unknown') # Name: Forward fill or use customer ID pattern df_clean['name'] = df_clean['name'].fillna( 'Customer_' + df_clean['customer_id'].str.replace('CUST_', '') ) return df_clean # 3. Outlier detection and treatment def detect_and_treat_outliers(df, columns, method='iqr', action='cap'): """Detect and treat outliers using various methods""" df_treated = df.copy() for col in columns: if method == 'iqr': Q1 = df[col].quantile(0.25) Q3 = df[col].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR elif method == 'zscore': z_scores = np.abs((df[col] - df[col].mean()) / df[col].std()) lower_bound = df[col].mean() - 3 * df[col].std() upper_bound = df[col].mean() + 3 * df[col].std() outliers_mask = (df[col] < lower_bound) | (df[col] > upper_bound) outliers_count = outliers_mask.sum() print(f"{col}: {outliers_count} outliers detected") if action == 'cap': df_treated.loc[df_treated[col] < lower_bound, col] = lower_bound df_treated.loc[df_treated[col] > upper_bound, col] = upper_bound elif action == 'remove': df_treated = df_treated[~outliers_mask] return df_treated # 4. Text data standardization def standardize_categorical_data(df, column): """Standardize categorical data with fuzzy matching""" df_clean = df.copy() # Convert to lowercase and strip whitespace df_clean[column] = df_clean[column].astype(str).str.strip().str.lower() # Handle empty strings df_clean[column] = df_clean[column].replace('', 'unknown') # Standardize common variations standardization_map = { 'premium': 'Premium', 'standard': 'Standard', 'basic': 'Basic', 'unknown': 'Unknown' } df_clean[column] = df_clean[column].map(standardization_map).fillna('Unknown') return df_clean # Apply cleaning pipeline print("\nApplying cleaning pipeline...") # Step 1: Handle missing values cleaned_data = handle_missing_values(messy_data) print(f"After missing value treatment: {cleaned_data.isnull().sum().sum()} missing values") # Step 2: Treat outliers cleaned_data = detect_and_treat_outliers( cleaned_data, ['age', 'income'], method='iqr', action='cap' ) # Step 3: Standardize categorical data cleaned_data = standardize_categorical_data(cleaned_data, 'category') print(f"Category distribution after standardization:") print(cleaned_data['category'].value_counts()) # 4. Data validation def validate_data_quality(df): """Validate data quality after cleaning""" validation_results = {} # Check for remaining missing values validation_results['missing_values'] = df.isnull().sum().sum() # Check for duplicate records validation_results['duplicates'] = df.duplicated().sum() # Check data ranges validation_results['invalid_ages'] = len(df[(df['age'] < 0) | (df['age'] > 120)]) validation_results['invalid_incomes'] = len(df[df['income'] < 0]) # Check email format email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' validation_results['invalid_emails'] = len(df[~df['email'].str.match(email_pattern, na=False)]) return validation_results validation_results = validate_data_quality(cleaned_data) print(f"\nValidation Results: {validation_results}") # 5. Advanced data transformation pipeline def create_transformation_pipeline(): """Create a reusable data transformation pipeline""" pipeline_steps = { 'missing_values': handle_missing_values, 'outliers': lambda df: detect_and_treat_outliers(df, ['age', 'income'], 'iqr', 'cap'), 'categorical': lambda df: standardize_categorical_data(df, 'category'), 'validation': validate_data_quality } return pipeline_steps # Apply complete pipeline pipeline = create_transformation_pipeline() final_data = messy_data.copy() for step_name, step_func in pipeline.items(): if step_name != 'validation': print(f"Applying {step_name} transformation...") final_data = step_func(final_data) # Final validation final_validation = validate_data_quality(final_data) print(f"\nFinal Data Quality: {final_validation}") print(f"Final dataset shape: {final_data.shape}") print(f"Data types: {final_data.dtypes.to_dict()}")

# High-performance join and merge operations # Create sample datasets for demonstration np.random.seed(42) # Main transaction dataset transactions = pd.DataFrame({ 'transaction_id': range(1, 100001), 'customer_id': np.random.randint(1, 50000, 100000), 'product_id': np.random.randint(1, 1000, 100000), 'amount': np.random.exponential(50, 100000), 'timestamp': pd.date_range('2023-01-01', periods=100000, freq='3min') }) # Customer information customers = pd.DataFrame({ 'customer_id': range(1, 50001), 'customer_type': np.random.choice(['Premium', 'Standard', 'Basic'], 50000), 'registration_date': pd.date_range('2020-01-01', periods=50000, freq='1H'), 'region': np.random.choice(['North', 'South', 'East', 'West'], 50000) }) # Product catalog products = pd.DataFrame({ 'product_id': range(1, 1001), 'category': np.random.choice(['Electronics', 'Clothing', 'Books', 'Home'], 1000), 'price': np.random.exponential(100, 1000), 'brand': np.random.choice(['Brand_A', 'Brand_B', 'Brand_C', 'Brand_D'], 1000) }) print(f"Transactions: {transactions.shape}") print(f"Customers: {customers.shape}") print(f"Products: {products.shape}") # 1. Performance comparison of different join methods def benchmark_joins(left_df, right_df, on_column, join_type='inner'): """Benchmark different join approaches""" # Method 1: Standard merge start_time = time.time() result1 = left_df.merge(right_df, on=on_column, how=join_type) merge_time = time.time() - start_time # Method 2: Using index (if beneficial) start_time = time.time() left_indexed = left_df.set_index(on_column) right_indexed = right_df.set_index(on_column) result2 = left_indexed.join(right_indexed, how=join_type).reset_index() index_join_time = time.time() - start_time # Method 3: Using categorical for repeated joins start_time = time.time() left_cat = left_df.copy() right_cat = right_df.copy() left_cat[on_column] = left_cat[on_column].astype('category') right_cat[on_column] = right_cat[on_column].astype('category') result3 = left_cat.merge(right_cat, on=on_column, how=join_type) categorical_time = time.time() - start_time return { 'merge_time': merge_time, 'index_join_time': index_join_time, 'categorical_time': categorical_time, 'result_shape': result1.shape } # Benchmark customer joins print("\nBenchmarking join performance:") customer_join_benchmark = benchmark_joins(transactions, customers, 'customer_id') print(f"Customer join benchmark: {customer_join_benchmark}") # 2. Complex multi-way joins def perform_multi_way_join(transactions, customers, products): """Efficiently perform multi-way joins""" # Step 1: Start with the largest table (transactions) # Step 2: Join with smaller tables in order of join selectivity # First join: transactions + customers (high selectivity) start_time = time.time() tx_customers = transactions.merge(customers, on='customer_id', how='left') step1_time = time.time() - start_time # Second join: result + products (lower selectivity) start_time = time.time() final_result = tx_customers.merge(products, on='product_id', how='left') step2_time = time.time() - start_time total_time = step1_time + step2_time print(f"Multi-way join completed in {total_time:.4f}s") print(f"Step 1 (customers): {step1_time:.4f}s") print(f"Step 2 (products): {step2_time:.4f}s") print(f"Final shape: {final_result.shape}") return final_result enriched_transactions = perform_multi_way_join(transactions, customers, products) # 3. Memory-efficient join for large datasets def memory_efficient_join(left_df, right_df, on_column, chunk_size=10000): """Join large datasets in chunks to manage memory""" results = [] for i in range(0, len(left_df), chunk_size): chunk = left_df.iloc[i:i+chunk_size] joined_chunk = chunk.merge(right_df, on=on_column, how='left') results.append(joined_chunk) if i % (chunk_size * 10) == 0: print(f"Processed {i + len(chunk):,} records...") return pd.concat(results, ignore_index=True) # 4. Fuzzy joining for approximate matches def fuzzy_string_join(left_df, right_df, left_col, right_col, threshold=0.8): """Perform fuzzy string matching join""" from difflib import SequenceMatcher def similarity(a, b): return SequenceMatcher(None, a.lower(), b.lower()).ratio() matches = [] for left_idx, left_val in enumerate(left_df[left_col]): best_match = None best_score = 0 for right_idx, right_val in enumerate(right_df[right_col]): score = similarity(str(left_val), str(right_val)) if score > best_score and score >= threshold: best_score = score best_match = right_idx if best_match is not None: matches.append({ 'left_idx': left_idx, 'right_idx': best_match, 'similarity_score': best_score }) return pd.DataFrame(matches) # 5. Optimized join strategies for different scenarios def optimize_join_strategy(left_df, right_df, join_cols, join_type='inner'): """Automatically choose the best join strategy""" left_size = len(left_df) right_size = len(right_df) # Calculate join selectivity left_unique = left_df[join_cols].nunique() if isinstance(join_cols, str) else left_df[join_cols].drop_duplicates().shape[0] right_unique = right_df[join_cols].nunique() if isinstance(join_cols, str) else right_df[join_cols].drop_duplicates().shape[0] print(f"Join optimization analysis:") print(f"Left table size: {left_size:,}") print(f"Right table size: {right_size:,}") print(f"Left unique keys: {left_unique:,}") print(f"Right unique keys: {right_unique:,}") # Strategy selection logic if left_size * right_size > 10**8: # Very large join strategy = "chunked_join" print("Recommendation: Use chunked processing") elif left_unique / left_size > 0.9 and right_unique / right_size > 0.9: strategy = "index_join" print("Recommendation: Use index-based join") else: strategy = "standard_merge" print("Recommendation: Use standard merge") return strategy # Apply join optimization optimization_result = optimize_join_strategy(transactions, customers, 'customer_id') # 6. Advanced join patterns # Time-based joins (asof joins) def time_based_join_example(): """Demonstrate time-based asof joins""" # Create time series data with different frequencies prices = pd.DataFrame({ 'timestamp': pd.date_range('2023-01-01', periods=1000, freq='1min'), 'symbol': np.random.choice(['AAPL', 'GOOGL', 'MSFT'], 1000), 'price': np.random.randn(1000) * 10 + 100 }).sort_values('timestamp') trades = pd.DataFrame({ 'timestamp': pd.date_range('2023-01-01', periods=500, freq='2min') + pd.Timedelta(seconds=30), 'symbol': np.random.choice(['AAPL', 'GOOGL', 'MSFT'], 500), 'volume': np.random.randint(100, 10000, 500) }).sort_values('timestamp') # Asof join to get the most recent price for each trade merged_data = pd.merge_asof( trades.sort_values('timestamp'), prices.sort_values('timestamp'), on='timestamp', by='symbol', direction='backward' ) print(f"Time-based join result shape: {merged_data.shape}") return merged_data time_join_result = time_based_join_example() # 7. Join validation and quality checks def validate_join_result(original_left, original_right, joined_result, join_type): """Validate join results for data quality""" validation = {} # Check expected row counts if join_type == 'inner': validation['row_count_valid'] = len(joined_result) <= min(len(original_left), len(original_right)) elif join_type == 'left': validation['row_count_valid'] = len(joined_result) >= len(original_left) elif join_type == 'outer': validation['row_count_valid'] = len(joined_result) >= max(len(original_left), len(original_right)) # Check for unexpected nulls left_cols = set(original_left.columns) right_cols = set(original_right.columns) validation['unexpected_nulls_left'] = joined_result[left_cols].isnull().sum().sum() validation['unexpected_nulls_right'] = joined_result[right_cols].isnull().sum().sum() # Check for data type preservation validation['dtypes_preserved'] = all( joined_result[col].dtype == original_left[col].dtype for col in left_cols if col in joined_result.columns ) return validation join_validation = validate_join_result(transactions, customers, enriched_transactions, 'left') print(f"\nJoin validation results: {join_validation}") print(f"\nFinal enriched dataset preview:") print(enriched_transactions.head(3))

# Advanced performance optimization techniques # Create large dataset for performance testing np.random.seed(42) large_dataset = pd.DataFrame({ 'id': range(1000000), 'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], 1000000), 'value1': np.random.randn(1000000), 'value2': np.random.randn(1000000), 'timestamp': pd.date_range('2020-01-01', periods=1000000, freq='1min'), 'status': np.random.choice([0, 1], 1000000, p=[0.7, 0.3]) }) print(f"Large dataset shape: {large_dataset.shape}") print(f"Memory usage: {large_dataset.memory_usage(deep=True).sum() / 1024**2:.1f}MB") # 1. Vectorization vs loops performance comparison def compare_vectorization_performance(df): """Compare vectorized operations vs loops""" # Method 1: Python loop (very slow) start_time = time.time() result_loop = [] for i in range(min(10000, len(df))): # Limit for demonstration if df.iloc[i]['value1'] > 0: result_loop.append(df.iloc[i]['value2'] * 2) else: result_loop.append(df.iloc[i]['value2'] * 0.5) loop_time = time.time() - start_time # Method 2: Pandas apply (better) start_time = time.time() result_apply = df.head(10000).apply( lambda row: row['value2'] * 2 if row['value1'] > 0 else row['value2'] * 0.5, axis=1 ) apply_time = time.time() - start_time # Method 3: Vectorized operations (fastest) start_time = time.time() result_vectorized = np.where( df.head(10000)['value1'] > 0, df.head(10000)['value2'] * 2, df.head(10000)['value2'] * 0.5 ) vectorized_time = time.time() - start_time print(f"Loop method: {loop_time:.4f}s") print(f"Apply method: {apply_time:.4f}s") print(f"Vectorized method: {vectorized_time:.4f}s") print(f"Vectorized speedup vs loop: {loop_time/vectorized_time:.1f}x") print(f"Vectorized speedup vs apply: {apply_time/vectorized_time:.1f}x") compare_vectorization_performance(large_dataset) # 2. Memory-efficient operations def memory_efficient_operations(df): """Demonstrate memory-efficient data processing""" print("\nMemory-efficient processing techniques:") # Use inplace operations when possible df_copy = df.copy() # Memory inefficient (creates copy) start_memory = df_copy.memory_usage(deep=True).sum() / 1024**2 start_time = time.time() df_copy['new_col'] = df_copy['value1'] + df_copy['value2'] copy_time = time.time() - start_time # Memory efficient (inplace when possible) df_copy2 = df.copy() start_time = time.time() df_copy2.eval('new_col = value1 + value2', inplace=True) eval_time = time.time() - start_time print(f"Standard assignment: {copy_time:.4f}s") print(f"Eval method: {eval_time:.4f}s") print(f"Eval speedup: {copy_time/eval_time:.1f}x") # Use categorical for memory savings category_memory_before = df['category'].memory_usage(deep=True) / 1024**2 df_categorical = df.copy() df_categorical['category'] = df_categorical['category'].astype('category') category_memory_after = df_categorical['category'].memory_usage(deep=True) / 1024**2 print(f"Category column memory reduction: {category_memory_before:.1f}MB -> {category_memory_after:.1f}MB") print(f"Memory savings: {((category_memory_before - category_memory_after) / category_memory_before * 100):.1f}%") memory_efficient_operations(large_dataset) # 3. Optimized aggregation strategies def optimize_aggregations(df): """Compare different aggregation approaches""" print("\nOptimizing aggregation operations:") # Method 1: Multiple separate groupby operations start_time = time.time() sum_result = df.groupby('category')['value1'].sum() mean_result = df.groupby('category')['value1'].mean() count_result = df.groupby('category')['value1'].count() separate_time = time.time() - start_time # Method 2: Single aggregation call start_time = time.time() combined_result = df.groupby('category')['value1'].agg(['sum', 'mean', 'count']) combined_time = time.time() - start_time # Method 3: Using eval for complex calculations start_time = time.time() df_temp = df.copy() df_temp.eval('combined_value = value1 + value2', inplace=True) complex_result = df_temp.groupby('category')['combined_value'].agg(['sum', 'mean']) eval_time = time.time() - start_time print(f"Separate operations: {separate_time:.4f}s") print(f"Combined aggregation: {combined_time:.4f}s") print(f"With eval: {eval_time:.4f}s") print(f"Combined speedup: {separate_time/combined_time:.1f}x") optimize_aggregations(large_dataset) # 4. Query optimization techniques def optimize_queries(df): """Optimize query and filtering operations""" print("\nQuery optimization techniques:") # Method 1: Chained boolean indexing start_time = time.time() filtered1 = df[df['value1'] > 0] filtered2 = filtered1[filtered1['value2'] < 1] filtered3 = filtered2[filtered2['status'] == 1] chained_time = time.time() - start_time # Method 2: Combined boolean indexing start_time = time.time() combined_filter = (df['value1'] > 0) & (df['value2'] < 1) & (df['status'] == 1) filtered_combined = df[combined_filter] combined_time = time.time() - start_time # Method 3: Using query method start_time = time.time() filtered_query = df.query('value1 > 0 and value2 < 1 and status == 1') query_time = time.time() - start_time print(f"Chained filtering: {chained_time:.4f}s") print(f"Combined boolean: {combined_time:.4f}s") print(f"Query method: {query_time:.4f}s") print(f"Combined speedup: {chained_time/combined_time:.1f}x") print(f"Query speedup: {chained_time/query_time:.1f}x") optimize_queries(large_dataset) # 5. Parallel processing techniques def demonstrate_parallel_processing(df): """Show parallel processing approaches""" print("\nParallel processing techniques:") # Method 1: Standard processing start_time = time.time() standard_result = df.groupby('category').apply( lambda x: x['value1'].rolling(window=100, min_periods=1).mean() ) standard_time = time.time() - start_time # Method 2: Using numba for acceleration (if available) try: import numba @numba.jit(nopython=True) def fast_rolling_mean(arr, window): n = len(arr) result = np.zeros(n) for i in range(n): start_idx = max(0, i - window + 1) result[i] = np.mean(arr[start_idx:i+1]) return result start_time = time.time() numba_results = [] for category in df['category'].unique(): cat_data = df[df['category'] == category]['value1'].values numba_result = fast_rolling_mean(cat_data, 100) numba_results.extend(numba_result) numba_time = time.time() - start_time print(f"Standard processing: {standard_time:.4f}s") print(f"Numba acceleration: {numba_time:.4f}s") print(f"Numba speedup: {standard_time/numba_time:.1f}x") except ImportError: print("Numba not available, showing standard processing only") print(f"Standard processing: {standard_time:.4f}s") # 6. Memory profiling and optimization def profile_memory_usage(df): """Profile memory usage of different operations""" import psutil import os process = psutil.Process(os.getpid()) def get_memory_usage(): return process.memory_info().rss / 1024**2 # MB initial_memory = get_memory_usage() print(f"\nMemory profiling:") print(f"Initial memory usage: {initial_memory:.1f}MB") # Operation 1: Create large intermediate result memory_before = get_memory_usage() temp_result = df.groupby('category').apply(lambda x: x.describe()) memory_after = get_memory_usage() print(f"After groupby describe: {memory_after:.1f}MB (+{memory_after - memory_before:.1f}MB)") # Clean up del temp_result # Operation 2: Memory-efficient alternative memory_before = get_memory_usage() efficient_result = df.groupby('category').agg({ 'value1': ['mean', 'std', 'min', 'max', 'count'], 'value2': ['mean', 'std', 'min', 'max', 'count'] }) memory_after = get_memory_usage() print(f"After efficient aggregation: {memory_after:.1f}MB (+{memory_after - memory_before:.1f}MB)") # Apply optimization techniques demonstrate_parallel_processing(large_dataset.head(100000)) # Use subset for demo profile_memory_usage(large_dataset.head(100000)) # 7. Best practices summary print("\n" + "="*50) print("PERFORMANCE OPTIMIZATION BEST PRACTICES") print("="*50) best_practices = [ "1. Use vectorized operations instead of apply() when possible", "2. Leverage eval() for complex mathematical expressions", "3. Set categorical data type for low-cardinality string columns", "4. Use query() method for complex filtering operations", "5. Combine multiple aggregations in single groupby call", "6. Use inplace=True for operations that don't need to return copies", "7. Profile memory usage and optimize data types", "8. Consider chunked processing for very large datasets", "9. Use appropriate index strategies for frequent lookups", "10. Leverage numba or other acceleration libraries for compute-intensive operations" ] for practice in best_practices: print(practice)