Note
Go to the end to download the full example code.
Benchmarking preprocessing with parallelization and serialization#
In this example, we compare the execution time and memory requirements of
preprocessing data with the parallelization and serialization functionalities
available in braindecode.preprocessing.preprocess().
We compare 4 cases:
Sequential, no serialization
Sequential, with serialization
Parallel, no serialization
Parallel, with serialization
Case 1 is the simplest approach, in which all recordings in a
braindecode.datasets.BaseConcatDataset are preprocessed one after the
other. In this scenario, braindecode.preprocessing.preprocess() acts
inplace, which means memory usage will likely stay stable (depending on the
preprocessing operations) if recordings have been preloaded. However, two
potential issues arise when working with large datasets: (1) if recordings have
not been preloaded before preprocessing, preprocess() will need to load them
and keep them in memory, in which case memory can become a bottleneck, and (2)
sequential preprocessing can take a considerable amount of time to run when
working with many recordings.
A solution to the first issue (memory usage) is to save the preprocessed data
to a file so it can be cleared from memory before moving on to the next
recording (case 2). The recordings can then be reloaded with preload=False
once they have all been saved to disk. This enables using the lazy loading
capabilities of braindecode.datasets.BaseConcatDataset and avoids
potential memory bottlenecks. The downside is that the writing to disk can take
some time and of course requires disk space.
A solution to the second issue (slow preprocessing) is to parallelize the preprocessing over multiple cores whenever possible (case 3). This can speed up preprocessing significantly. However, this approach will increase memory usage because of the way parallelization is implemented internally (with joblib, copies of (part of) the data must be made when sending arguments to parallel processes).
Finally, case 4 (combining parallelization and serialization) is likely to be both fast and memory efficient. As shown in this example, this remains a tradeoff though, and the selected configuration should depend on the size of the dataset and the specific operations applied to the recordings.
# Authors: Hubert Banville <hubert.jbanville@gmail.com>
#
# License: BSD (3-clause)
import time
import tempfile
from itertools import product
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.preprocessing import scale
from memory_profiler import memory_usage
from braindecode.datasets import SleepPhysionet
from braindecode.preprocessing import (
preprocess,
Preprocessor,
create_fixed_length_windows,
)
We create a function that goes through the usual three steps of data
preparation: (1) data loading, (2) continuous data preprocessing,
(3) windowing and (4) windowed data preprocessing. We use the
braindecode.datasets.SleepPhysionet dataset for testing purposes.
def prepare_data(n_recs, save, preload, n_jobs):
if save:
tmp_dir = tempfile.TemporaryDirectory()
save_dir = tmp_dir.name
else:
save_dir = None
# (1) Load the data
concat_ds = SleepPhysionet(
subject_ids=range(n_recs), recording_ids=[1], crop_wake_mins=30, preload=preload
)
sfreq = concat_ds.datasets[0].raw.info["sfreq"]
# (2) Preprocess the continuous data
preprocessors = [
Preprocessor("crop", tmin=10),
Preprocessor("filter", l_freq=None, h_freq=30),
]
preprocess(
concat_ds, preprocessors, save_dir=save_dir, overwrite=True, n_jobs=n_jobs
)
# (3) Window the data
windows_ds = create_fixed_length_windows(
concat_ds,
0,
None,
int(30 * sfreq),
int(30 * sfreq),
True,
preload=preload,
n_jobs=n_jobs,
)
# Preprocess the windowed data
preprocessors = [Preprocessor(scale, channel_wise=True)]
preprocess(
windows_ds, preprocessors, save_dir=save_dir, overwrite=True, n_jobs=n_jobs
)
Next, we can run our function and measure its run time and peak memory usage for each one of our 4 cases above. We call the function multiple times with each configuration to get better estimates.
Note
To better characterize the run time vs. memory usage tradeoff for your specific configuration (as this will differ based on available hardware, data size and preprocessing operations), we recommend adapting this example to your use case and running it on your machine.
n_repets = 2 # Number of repetitions
all_n_recs = 2 # Number of recordings to load and preprocess
all_n_jobs = [1, 2] # Number of parallel processes
results = list()
for _, n_recs, save, n_jobs in product(
range(n_repets), [all_n_recs], [True, False], all_n_jobs
):
start = time.time()
mem = max(memory_usage(proc=(prepare_data, [n_recs, save, False, n_jobs], {})))
time_taken = time.time() - start
results.append(
{
"n_recs": n_recs,
"max_mem": mem,
"save": save,
"n_jobs": n_jobs,
"time": time_taken,
}
)
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Reading 0 ... 2507000 = 0.000 ... 25070.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
Writing /tmp/tmpqcwqb2hl/0/0-raw.fif
Closing /tmp/tmpqcwqb2hl/0/0-raw.fif
[done]
Reading 0 ... 3260000 = 0.000 ... 32600.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
Writing /tmp/tmpqcwqb2hl/1/1-raw.fif
Closing /tmp/tmpqcwqb2hl/1/1-raw.fif
[done]
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:788: UserWarning: Chosen directory /tmp/tmpqcwqb2hl contains other subdirectories or files ['0'].
warnings.warn(
Opening raw data file /tmp/tmpqcwqb2hl/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpqcwqb2hl/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fc1d4e30>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
Reading 0 ... 2507000 = 0.000 ... 25070.000 secs...
Writing /tmp/tmpqcwqb2hl/0/0-raw.fif
Closing /tmp/tmpqcwqb2hl/0/0-raw.fif
[done]
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:777: UserWarning: The number of saved datasets (1) does not match the number of existing subdirectories (2). You may now encounter a mix of differently preprocessed datasets!
warnings.warn(
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:788: UserWarning: Chosen directory /tmp/tmpqcwqb2hl contains other subdirectories or files ['1'].
warnings.warn(
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fc1d4e30>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
Reading 0 ... 3260000 = 0.000 ... 32600.000 secs...
Writing /tmp/tmpqcwqb2hl/1/1-raw.fif
Closing /tmp/tmpqcwqb2hl/1/1-raw.fif
[done]
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:788: UserWarning: Chosen directory /tmp/tmpqcwqb2hl contains other subdirectories or files ['0'].
warnings.warn(
Opening raw data file /tmp/tmpqcwqb2hl/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpqcwqb2hl/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Opening raw data file /tmp/tmpkgl8aox0/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpkgl8aox0/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
Opening raw data file /tmp/tmpkgl8aox0/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpkgl8aox0/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Reading 0 ... 2507000 = 0.000 ... 25070.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
Reading 0 ... 3260000 = 0.000 ... 32600.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fc360770>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fc360770>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Reading 0 ... 2507000 = 0.000 ... 25070.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
Writing /tmp/tmpdoa51ob8/0/0-raw.fif
Closing /tmp/tmpdoa51ob8/0/0-raw.fif
[done]
Reading 0 ... 3260000 = 0.000 ... 32600.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
Writing /tmp/tmpdoa51ob8/1/1-raw.fif
Closing /tmp/tmpdoa51ob8/1/1-raw.fif
[done]
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:788: UserWarning: Chosen directory /tmp/tmpdoa51ob8 contains other subdirectories or files ['0'].
warnings.warn(
Opening raw data file /tmp/tmpdoa51ob8/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpdoa51ob8/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fb911ee0>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
Reading 0 ... 2507000 = 0.000 ... 25070.000 secs...
Writing /tmp/tmpdoa51ob8/0/0-raw.fif
Closing /tmp/tmpdoa51ob8/0/0-raw.fif
[done]
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:777: UserWarning: The number of saved datasets (1) does not match the number of existing subdirectories (2). You may now encounter a mix of differently preprocessed datasets!
warnings.warn(
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:788: UserWarning: Chosen directory /tmp/tmpdoa51ob8 contains other subdirectories or files ['1'].
warnings.warn(
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fb911ee0>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
Reading 0 ... 3260000 = 0.000 ... 32600.000 secs...
Writing /tmp/tmpdoa51ob8/1/1-raw.fif
Closing /tmp/tmpdoa51ob8/1/1-raw.fif
[done]
/home/runner/work/braindecode/braindecode/braindecode/datasets/base.py:788: UserWarning: Chosen directory /tmp/tmpdoa51ob8 contains other subdirectories or files ['0'].
warnings.warn(
Opening raw data file /tmp/tmpdoa51ob8/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpdoa51ob8/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Opening raw data file /tmp/tmpm8n9ns19/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpm8n9ns19/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
Opening raw data file /tmp/tmpm8n9ns19/0/0-raw.fif...
Isotrak not found
Range : 2884000 ... 5391000 = 28840.000 ... 53910.000 secs
Ready.
Opening raw data file /tmp/tmpm8n9ns19/1/1-raw.fif...
Isotrak not found
Range : 1975000 ... 5235000 = 19750.000 ... 52350.000 secs
Ready.
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Reading 0 ... 2507000 = 0.000 ... 25070.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
Reading 0 ... 3260000 = 0.000 ... 32600.000 secs...
Filtering raw data in 1 contiguous segment
Setting up low-pass filter at 30 Hz
FIR filter parameters
---------------------
Designing a one-pass, zero-phase, non-causal lowpass filter:
- Windowed time-domain design (firwin) method
- Hamming window with 0.0194 passband ripple and 53 dB stopband attenuation
- Upper passband edge: 30.00 Hz
- Upper transition bandwidth: 7.50 Hz (-6 dB cutoff frequency: 33.75 Hz)
- Filter length: 45 samples (0.450 s)
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fbbeb170>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
/home/runner/work/braindecode/braindecode/braindecode/preprocessing/preprocess.py:246: UserWarning: Applying preprocessors [<braindecode.preprocessing.preprocess.Preprocessor object at 0x7f48fbbeb170>] to the mne.io.Raw of an EEGWindowsDataset.
warn(
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4001E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Extracting EDF parameters from /home/runner/mne_data/physionet-sleep-data/SC4011E0-PSG.edf...
EDF file detected
Setting channel info structure...
Creating raw.info structure...
Finally, we can plot the results:
df = pd.DataFrame(results)
fig, ax = plt.subplots(figsize=(6, 4))
colors = {True: "tab:orange", False: "tab:blue"}
markers = {n: m for n, m in zip(all_n_jobs, ["o", "x", "."])}
for (save, n_jobs), sub_df in df.groupby(["save", "n_jobs"]):
ax.scatter(
x=sub_df["time"],
y=sub_df["max_mem"],
color=colors[save],
marker=markers[n_jobs],
label=f"save={save}, n_jobs={n_jobs}",
)
ax.legend()
ax.set_xlabel("Execution time (s)")
ax.set_ylabel("Memory usage (MiB)")
ax.set_title(f"Loading and preprocessing {all_n_recs} recordings from Sleep Physionet")
plt.show()
We see that parallel preprocessing without serialization (blue crosses) is faster than simple sequential processing (blue circles), however it uses more memory.
Combining parallel preprocessing and serialization (orange crosses) reduces memory usage significantly, however it increases run time by a few seconds. Depending on available resources (e.g. in limited memory settings), it might therefore be more advantageous to use both parallelization and serialization together.
Total running time of the script: (0 minutes 47.182 seconds)
Estimated memory usage: 817 MB