[Python] Efficient way to deal with large files

I am in need of advice please. I am working on a Conditioned Based Maintenance school project for marine engines using AI and ML techniques. The data that i have contains both normal and faulty files for different load conditions. They all contain the same number of columns(engine parameters: cylinder pressure etc) but the number of rows are massive. On average I have 30000000 rows per column(29 columns per file) for a total of 56 files. For the life of me, no matter how much i try to move forward with my project; be it pandas or dask, I always get stuck at some point due to memory error. Any advice would be appreciated.

Code block example:

import dask.dataframe as dd
import cupy as cp
import pandas as pd
import numpy as np
from scipy.stats import skew, kurtosis
from scipy.fft import rfft
from tqdm import tqdm
import gc

# Function to Compute Advanced Statistical Features (Std Dev, Skewness, Kurtosis, etc.) using CuPy
def compute_advanced_features(values):
"""
Compute Advanced Statistical Features:
- Std Dev
- Skewness
- Kurtosis
- Rate of Change (ΔCR = Mean - Std Dev)
- First 5 FFT Components (using real FFT)
"""
values = values.dropna() # Drop NaNs to avoid issues
if len(values) == 0:
return pd.Series([np.nan] * 9, index=['Std_Dev', 'Skew', 'Kurtosis', 'Delta_CR'] + [f'FFT{i+1}' for i in range(5)])

# Convert to CuPy array for GPU computation
values_gpu = cp.asarray(values.values)

# Compute standard features using CuPy
std_dev = cp.std(values_gpu)
mean_val = cp.mean(values_gpu)
delta_cr = mean_val - std_dev

# For skewness and kurtosis, use numpy functions for compatibility
skewness = skew(values)
kurt = kurtosis(values)

# Compute FFT using real FFT (memory-efficient for real data)
fft_vals = rfft(values_gpu.get())[:5] # Only keep first 5 FFT components
fft_vals = np.pad(fft_vals, (0, max(0, 5 - len(fft_vals))), mode='constant', constant_values=np.nan)

# Return as a pandas series
return pd.Series([std_dev.get(), skewness, kurt, delta_cr.get()] + list(fft_vals),
index=['Std_Dev', 'Skew', 'Kurtosis', 'Delta_CR'] + [f'FFT{i+1}' for i in range(5)])

# Function to Process Features in Chunks using Dask and CuPy
def process_features_in_chunks(df, pressure_col, encoder_col='encoder_in_degrees', chunk_size=50000):
"""
Extracts features like Pmax, Mean Pressure, and Advanced Statistical Features (Std Dev, Skew, Kurtosis, FFT).
Processes data in chunks using Dask and CuPy for GPU-based acceleration.
"""
print(f"\nProcessing {pressure_col}...")

def process_partition(chunk):
# Assign Cycle Numbers based on Encoder Values
chunk['Cycle'] = (chunk[encoder_col] // 360).astype(int) + 1
chunk.dropna(subset=['Cycle'], inplace=True)

# Compute Pmax (Peak Pressure) and Mean Pressure
feature_df = chunk.groupby('Cycle')[pressure_col].agg(Pmax='max', Mean_Pressure='mean').reset_index()

# Apply Advanced Statistical Features
advanced_features = chunk.groupby('Cycle')[pressure_col].apply(compute_advanced_features).reset_index()

# Merge Advanced Features with Pmax and Mean Pressure
feature_df = feature_df.merge(advanced_features, on='Cycle', how='left')

# Compute ΔPmax (Peak Pressure Deviation)
feature_df['Delta_Pmax'] = feature_df['Pmax'] - feature_df['Mean_Pressure']

# Handle NaN Values (Imputation)
feature_df.fillna(method='ffill', inplace=True)
feature_df.fillna(method='bfill', inplace=True)
feature_df.fillna(feature_df.mean(numeric_only=True), inplace=True)

return feature_df

# Apply the function to each partition using map_partitions
result = df.map_partitions(process_partition)
return result

# Efficient Batched Merging with Dask (Lazy Evaluation)
def merge_in_batches(left_df, right_df, on, batch_size=10000):
"""
Efficiently merges two DataFrames in batches to avoid memory overload.
"""
merged_batches = []
num_batches = (len(left_df) // batch_size) + 1 # Calculate batches based on batch size

for i in tqdm(range(num_batches), desc="Merging in Batches", total=num_batches):
# Process smaller batches of data
left_batch = left_df.loc[i * batch_sizei + 1) * batch_size].compute() # Use loc for row indexing
merged_batch = dd.merge(left_batch, right_df, on=on, how='left') # Perform merge with Dask
merged_batches.append(merged_batch)

# Explicitly release memory after processing each batch
del left_batch
gc.collect() # Perform garbage collection to free memory

# Concatenate merged batches into a single DataFrame using Dask
return dd.concat(merged_batches, axis=0)

def process_and_merge_data(normal_data, faulty_data, pressure_columns):
Load_0_to_25_max_pressure_all_normal = None
Load_0_to_25_max_pressure_all_faulty = None

with ProgressBar():
for col in pressure_columns:
# Process Normal Data in batches
normal_features = process_features_in_chunks(normal_data, col)

# Process Faulty Data in batches
faulty_features = process_features_in_chunks(faulty_data, col)

# Rename the columns to avoid duplication during merge
normal_features.columns = [f'{c}_normal_{col}' if c != 'Cycle' else c for c in normal_features.columns]
faulty_features.columns = [f'{c}_faulty_{col}' if c != 'Cycle' else c for c in faulty_features.columns]

# Merge the features from each pressure column into a single DataFrame in batches using Dask
if Load_0_to_25_max_pressure_all_normal is None:
Load_0_to_25_max_pressure_all_normal = normal_features
else:
Load_0_to_25_max_pressure_all_normal = merge_in_batches(Load_0_to_25_max_pressure_all_normal, normal_features, on='Cycle')

if Load_0_to_25_max_pressure_all_faulty is None:
Load_0_to_25_max_pressure_all_faulty = faulty_features
else:
Load_0_to_25_max_pressure_all_faulty = merge_in_batches(Load_0_to_25_max_pressure_all_faulty, faulty_features, on='Cycle')

return Load_0_to_25_max_pressure_all_normal, Load_0_to_25_max_pressure_all_faulty


# Assuming 'Load_0_to_25_normal_data' and 'Load_0_to_25_faulty_data' are already defined as Dask DataFrames
Load_0_to_25_max_pressure_all_normal, Load_0_to_25_max_pressure_all_faulty = process_and_merge_data(
Load_0_to_25_normal_data, Load_0_to_25_faulty_data, pressure_columns
)

# Save results to Parquet files (optional)
Load_0_to_25_max_pressure_all_normal.to_parquet("Load_0_to_25_max_pressure_all_normal.parquet", compression='snappy')
Load_0_to_25_max_pressure_all_faulty.to_parquet("Load_0_to_25_max_pressure_all_faulty.parquet", compression='snappy')

# Display Results
print(Load_0_to_25_max_pressure_all_normal.head())
print(Load_0_to_25_max_pressure_all_faulty.head())

Continue reading...