3_ensembles/solutions/py/dataloaders.py

+"""
+Sample data samples from CSV file with replacement using multiple processors in truly parallel fashion.
+Each processor needs to know
+  - overall number samples
+  - absolute byte positions of data samples' start / end as array byte_pos = [[start0, end0], [start1, end1],...]
+Once I got this array:
+- Perform train-test split by randomly sampling `test_frac` indices from overall number of samples xn = len(byte_pos).
+  Those samples are held out to form test dataset. The other samples form global train dataset.
+- Assume p processors each of which holds sub random forest. Each sub forest should train on `train_frac` * xn/p data
+  samples, where `train_frac` = 1 - `test_frac`. For this, each processor should sample those `train_frac` * xn/p data
+  samples from global train dataset with replacement, i.e., globally, the random forest is trained on `train_frac` * xn
+  samples, however, some of them might occur multiple times while others might not occur at all.
+To do so, each processor needs to have `byte_pos` array!
+"""
+import os.path
+import pathlib
+from typing import BinaryIO, List, Tuple, Union
+
+from mpi4py import MPI
+from mpi4py.util.dtlib import from_numpy_dtype
+import numpy as np
+from sklearn.model_selection import train_test_split
+
+
+def _determine_line_starts(
+    f: BinaryIO, comm: MPI.Comm, displs: np.ndarray, counts: np.ndarray
+) -> Tuple[list, bytes, int]:
+    """
+    Determine line starts in bytes from CSV file.
+
+    Parameters
+    ----------
+    f : BinaryIO
+        The handle of CSV file to read from
+    comm : MPI.Comm
+        The communicator to use.
+    displs : np.ndarray
+        The displacements of byte chunk starts on each rank, length is `comm.size`.
+    counts : np.ndarray
+        The counts of the byte chunk on each rank, length is `comm.size`.
+
+    Returns
+    -------
+    list
+        The line starts in each byte chunk.
+    bytes
+        The bytes chunk read in on each rank.
+    int
+        The number of chars used to indicate the line end in the file.
+    """
+    rank = comm.rank  # Set up communicator stuff.
+
+    # Read bytes chunk and count linebreaks.
+    # \r (Carriage Return) : Move cursor to line start w/o advancing to next line.
+    # \n (Line Feed) : Move cursor down to next line w/o returning to line start.
+    # \r\n (End Of Line) : Combination of \r and \n.
+
+    lineter_len = 1  # Set number of line termination chars.
+    line_starts = []  # Set up list to save line starts.
+    f.seek(displs[rank])  # Jump to pre-assigned position in file.
+    r = f.read(counts[rank])  # Read number of bytes from starting position.
+
+    for pos, l in enumerate(r):  # Determine line breaks in bytes chunk.
+        if chr(l) == "\n":  # Line terminated by '\n' only.
+            if not chr(r[pos - 1]) == "\r":  # No \r\n.
+                line_starts.append(pos + 1)
+        elif chr(l) == "\r":
+            if (
+                pos + 1 < len(r) and chr(r[pos + 1]) == "\n"
+            ):  # Line terminated by '\r\n'.
+                line_starts.append(pos + 2)
+                lineter_len = 2
+            else:  # Line terminated by '\r' only.
+                line_starts.append(pos + 1)
+    return line_starts, r, lineter_len
+
+
+def _get_byte_pos_from_line_starts(
+    line_starts: np.ndarray, file_size: int, lineter_len: int
+) -> np.ndarray:
+    """
+    Get line starts and counts in byte from line starts to read lines via seek and read.
+
+    Parameters
+    ----------
+    line_starts : np.ndarray
+        The absolute positions of line starts in bytes.
+    file_size : int
+        The absolute file size in bytes.
+    lineter_len : int
+        The number of line termination characters.
+
+    Returns
+    -------
+    np.ndarray
+        An array containing vectors of (start, count) for lines in byte.
+    """
+    lines_byte = []  # list for line starts and counts
+    for idx in range(len(line_starts)):  # Loop through all line starts.
+        if idx == len(line_starts) - 1:  # Special case for last line.
+            temp = [
+                line_starts[idx],  # Line start in bytes
+                file_size
+                - line_starts[idx]
+                - lineter_len,  # Bytes count of line length via difference
+            ]
+        else:  # All other lines
+            temp = [
+                line_starts[idx],  # Line start in bytes
+                line_starts[idx + 1]
+                - line_starts[idx]
+                - lineter_len,  # Bytes count of line length via difference
+            ]
+        lines_byte.append(temp)
+    return np.array(lines_byte)
+
+
+def _decode_bytes_array(
+    byte_pos: np.ndarray, f: BinaryIO, sep: str = ",", encoding: str = "utf-8"
+) -> List:
+    """
+    Decode lines from byte positions and counts.
+
+    Parameters
+    ----------
+    byte_pos : np.ndarray
+        The vectors of line starts and lengths in bytes.
+    f : BinaryIO
+        The handle of the CSV file to read from.
+    sep : char
+        The character used in the file to separate entries.
+    encoding : str
+        The encoding used to decode entries from bytes.
+
+    Returns
+    -------
+    list
+        Values read from CSV file as numpy array entries in float format
+    """
+    lines = []  # List for saving decoded lines
+    byte_pos = byte_pos[
+        byte_pos[:, 0].argsort()
+    ]  # Sort line starts of data items to read from file in ascending order.
+    for item in byte_pos:
+        f.seek(item[0])  # Go to line start.
+        line = f.read(item[1]).decode(
+            encoding
+        )  # Read specified number of bytes and decode.
+        if len(line) > 0:
+            sep_values = [
+                float(val) for val in line.split(sep)
+            ]  # Separate values in each line.
+            line = np.array(
+                sep_values
+            )  # Convert list of separated values to numpy array.
+        lines.append(line)  # Append numpy data entry to output list.
+    return lines
+
+
+def _split_indices_train_test(
+    n_samples: int, train_split: float, seed: int
+) -> Tuple[int, int, np.ndarray, np.ndarray]:
+    """
+    Make index-based train test split with shuffle.
+
+    Parameters
+    ----------
+    n_samples : int
+        The overall number of samples in the dataset.
+    train_split : float
+        The fraction of the dataset used for training (0 < `train_split` < 1)
+    seed : int
+        The seed used for the random number generator.
+
+    Returns
+    -------
+    n_train_samples : int
+        The number of train samples.
+    n_test_samples : int
+        The number of test samples.
+    train_indices : np.ndarray
+        The indices of the samples in the dataset used for training.
+    test_indices : np.ndarray
+        The indices of the samples in the dataset used for testing.
+    """
+    n_train_samples = int(
+        train_split * n_samples
+    )  # Determine number of train samples from split.
+    n_test_samples = n_samples - n_train_samples  # Determine number of test samples.
+    rng = np.random.default_rng(
+        seed=seed
+    )  # Set same seed over all ranks for consistent train-test split.
+    all_indices = np.arange(0, n_samples)  # Construct array of all indices.
+    rng.shuffle(all_indices)  # Shuffle them.
+    train_indices = all_indices[
+        :n_train_samples
+    ]  # First `n_train_samples` indices form train set.
+    test_indices = all_indices[n_train_samples:]  # Remaining ones form test set.
+    return n_train_samples, n_test_samples, train_indices, test_indices
+
+
+def load_data_csv_parallel(
+    path_to_data: Union[pathlib.Path, str],
+    header_lines: int,
+    comm: MPI.Comm = MPI.COMM_WORLD,
+    train_split: float = 0.75,
+    verbose: bool = True,
+) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
+    """
+    Load data from CSV file in truly parallel fashion.
+
+    Parameters
+    ----------
+    path_to_data : str | pathlib.Path
+        The path to the CSV file.
+    header_lines : int
+        The number of header lines.
+    comm : MPI.Comm
+        The communicator to use.
+    train_split : float
+        The train-test split fraction.
+    verbose : bool
+        The verbosity level.
+
+    Returns
+    -------
+    train_samples_local : np.ndarray
+        The rank-dependent train samples.
+    train_targets_local : np.ndarray
+        The rank-dependent train targets.
+    test_samples : np.ndarray
+        The global test samples.
+    test_targets : np.ndarray
+        The global test targets.
+    """
+    rank, size = comm.rank, comm.size  # Set up communicator stuff.
+    file_size = os.stat(path_to_data).st_size  # Get file size in bytes.
+
+    # Determine displs + counts of bytes chunk to read on each rank.
+    base = file_size // size  # Determine base chunk size for each process.
+    remainder = file_size % size  # Determine remainder bytes.
+    counts = base * np.ones(  # Construct array with each rank's chunk counts.
+        (size,), dtype=int
+    )
+    if (
+        remainder > 0
+    ):  # Equally distribute remainder over respective ranks to balance load.
+        counts[:remainder] += 1
+    displs = np.concatenate(  # Determine displs via cumulative sum from counts.
+        (np.zeros((1,), dtype=int), np.cumsum(counts, dtype=int)[:-1])
+    )
+
+    if rank == 0:
+        print(f"File size is {file_size} bytes.")
+
+    if rank == 0 and verbose:
+        print(f"Displs {displs}, counts {counts} for reading bytes chunks from file.")
+
+    with open(path_to_data, "rb") as f:  # Open csv file to read from.
+        # Determine line starts in bytes chunks on each rank.
+        line_starts, r, lineter_len = _determine_line_starts(f, comm, displs, counts)
+
+        # On rank 0, add very first line.
+        if rank == 0:
+            line_starts = [0] + line_starts
+
+        if verbose:
+            print(f"[{rank}/{size}]: {len(line_starts)} line starts in chunk.")
+
+        # Find correct starting point, considering header lines.
+        # All-gather numbers of line starts in each chunk in `total_lines` array.
+        total_lines = np.empty(size, dtype=int)
+        comm.Allgather(
+            [np.array(len(line_starts), dtype=int), MPI.INT], [total_lines, MPI.INT]
+        )
+        cumsum = list(np.cumsum(total_lines))
+        # Determine rank where actual data lines start,
+        # i.e. remove ranks only containing header lines.
+        start = next(i for i in range(size) if cumsum[i] > header_lines)
+        if verbose:
+            print(
+                f"[{rank}/{size}]: total_lines is {total_lines}.\n"
+                f"[{rank}/{size}]: cumsum is {cumsum}.\n"
+                f"[{rank}/{size}]: start is {start}."
+            )
+
+        if rank < start:  # Ranks containing only header lines.
+            line_starts = []
+        if rank == start:  # Rank containing header + data lines.
+            rem = header_lines - (0 if start == 0 else cumsum[start - 1])
+            line_starts = line_starts[rem:]
+
+        # Share line starts of data samples across all ranks via Allgatherv.
+        line_starts += displs[
+            rank
+        ]  # Shift line starts on each rank according to displs.
+        if verbose:
+            print(
+                f"[{rank}/{size}]: {len(line_starts)} line starts of shape {line_starts.shape} and type "
+                f"{line_starts.dtype} in local chunk: {line_starts}"
+            )
+        count_linestarts = np.array(  # Determine local number of line starts.
+            len(line_starts), dtype=int
+        )
+        counts_linestarts = (
+            np.empty(  # Initialize array to all-gather local numbers of line starts.
+                size, dtype=int
+            )
+        )
+        comm.Allgather([count_linestarts, MPI.INT], [counts_linestarts, MPI.INT])
+        n_linestarts = np.sum(
+            counts_linestarts
+        )  # Determine overall number of line starts.
+        displs_linestarts = (
+            np.concatenate(  # Determine displacements of line starts from counts.
+                (
+                    np.zeros((1,), dtype=int),
+                    np.cumsum(counts_linestarts, dtype=int)[:-1],
+                ),
+                dtype=int,
+            )
+        )
+        if verbose and rank == 0:
+            print(
+                f"Overall {n_linestarts} linestarts.\n"
+                f"Number of linestarts in each chunk is {counts_linestarts}.\n"
+                f"Displs of linestarts in each chunk is {displs_linestarts}."
+            )
+
+        all_line_starts = (
+            np.empty(  # Initialize array to all-gatherv line starts from all ranks.
+                (n_linestarts,), dtype=line_starts.dtype
+            )
+        )
+        if verbose and rank == 0:
+            print(
+                f"Recvbuf {all_line_starts}, {all_line_starts.shape}, {all_line_starts.dtype}."
+            )
+        comm.Allgatherv(
+            line_starts,
+            [
+                all_line_starts,
+                counts_linestarts,
+                displs_linestarts,
+                from_numpy_dtype(line_starts.dtype),
+            ],
+        )
+        # Line starts were determined as those positions following a line end.
+        # But: There is no line after last line end in file.
+        # Thus, remove last entry from all_line_starts.
+        all_line_starts = all_line_starts[:-1]
+        if rank == 0:
+            print(f"After Allgatherv: All line starts: {all_line_starts}")
+        # Construct array with line starts and lengths in bytes.
+        print(
+            f"[{rank}/{size}]: Construct array with line starts and lengths in bytes."
+        )
+        lines_byte = _get_byte_pos_from_line_starts(
+            all_line_starts, file_size, lineter_len
+        )
+
+        # Make global train-test split.
+        print(f"[{rank}/{size}]: Make global train-test split.")
+        (
+            n_train_samples,
+            n_test_samples,
+            train_indices,
+            test_indices,
+        ) = _split_indices_train_test(
+            n_samples=len(lines_byte), train_split=train_split, seed=size
+        )
+
+        n_train_samples_local = (
+            n_train_samples // size
+        )  # Determine local train dataset size.
+        remainder_train = n_train_samples % size  # Balance load.
+        if rank < remainder_train:
+            n_train_samples_local += 1
+
+        # Construct held-out test dataset (same on each rank).
+        print(f"[{rank}/{size}]: Decode {n_test_samples} test samples from file.")
+        test_lines = _decode_bytes_array(
+            lines_byte[test_indices], f, sep=",", encoding="utf-8"
+        )
+        test_samples = np.array(test_lines)[:, 1:]
+        test_targets = np.array(test_lines)[:, 0]
+        if verbose:
+            print_str = f"[{rank}/{size}]: Test samples: {test_samples[0]}\n"
+            print_str += f"Test targets: {test_targets[0]}\n"
+            print(print_str)
+
+        # Construct train dataset (different on each rank).
+        rng = np.random.default_rng(seed=rank)
+        print(f"[{rank}/{size}]: Draw local {n_train_samples_local} train indices.")
+        train_indices_local = rng.choice(train_indices, size=n_train_samples_local)
+        print(f"[{rank}/{size}]: Decode train lines from file.")
+        train_lines_local = _decode_bytes_array(
+            lines_byte[train_indices_local], f, sep=",", encoding="utf-8"
+        )
+        train_samples_local = np.array(train_lines_local)[:, 1:]
+        train_targets_local = np.array(train_lines_local)[:, 0]
+
+    # Now, each rank holds a local train set (samples + targets)
+    # and the global held-out test set (samples + targets).
+    return train_samples_local, train_targets_local, test_samples, test_targets
+
+
+def load_data_csv_root(
+    path_to_data: Union[str, pathlib.Path],
+    header_lines: int,
+    comm: MPI.Comm,
+    seed: int,
+    train_split: float = 0.75,
+    verbose: bool = True,
+) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
+    """
+    Load data from CSV file via root process.
+
+    Parameters
+    ----------
+    path_to_data : str | pathlib.Path
+        The path to the CSV file.
+    header_lines : int
+        The number of header lines.
+    comm : MPI.Comm
+        The communicator to use.
+    seed : int
+        The seed used for train-test split.
+    train_split : float
+        The train-test split fraction.
+    verbose : bool
+        The verbosity level
+
+    Returns
+    -------
+    np.ndarray
+        The rank-dependent train samples.
+    np.ndarray
+        The rank-dependent train targets.
+    np.ndarray
+        The global test samples.
+    np.ndarray
+        The global test targets.
+    """
+    rank, size = comm.rank, comm.size  # Set up communicator stuff.
+
+    if rank == 0:
+        data = np.loadtxt(
+            path_to_data, dtype=float, delimiter=",", skiprows=header_lines
+        )  # Load data into numpy array.
+        samples, targets = (
+            data[:, 1:],
+            data[:, 0],
+        )  # Divide data into samples and targets.
+        samples_train, samples_test, targets_train, targets_test = train_test_split(
+            samples, targets, test_size=1 - train_split, random_state=seed
+        )  # Perform train-test split.
+        n_train_samples = len(
+            samples_train
+        )  # Determine number of samples in train set.
+        n_test_samples = len(samples_test)  # Determine number of samples in test set.
+        n_features = samples_train.shape[1]  # Determine number of features.
+        n_train_samples_local = (
+            n_train_samples // size
+        )  # Determine rank-local number of train samples.
+        remainder_train = n_train_samples % size
+        train_counts = n_train_samples_local * np.ones(
+            size, dtype=int
+        )  # Determine load-balanced counts and displacements.
+        for idx in range(remainder_train):
+            train_counts[idx] += 1
+        train_displs = np.concatenate(
+            (np.zeros(1, dtype=int), np.cumsum(train_counts, dtype=int)[:-1]), dtype=int
+        )
+        print(
+            f"There are {n_train_samples} train and {n_test_samples} test samples.\n"
+            f"Local train samples: {train_counts}"
+        )
+        # Sample `n_train_samples` indices from train set with replacement.
+        rng = np.random.default_rng(seed=size)
+        train_indices = rng.choice(range(n_train_samples), size=n_train_samples)
+        print(f"train_indices have shape {train_indices.shape}.")
+        # Construct train dataset from drawn indices (repetitions possible!).
+        samples_train_shuffled = samples_train[train_indices]
+        targets_train_shuffled = targets_train[train_indices]
+        send_buf_train_samples = [
+            samples_train_shuffled,
+            train_counts * n_features,
+            train_displs * n_features,
+            from_numpy_dtype(samples_train_shuffled.dtype),
+        ]
+        send_buf_train_targets = [
+            targets_train_shuffled,
+            train_counts,
+            train_displs,
+            from_numpy_dtype(targets_train_shuffled.dtype),
+        ]
+
+    else:
+        train_counts = None
+        n_features = None
+        n_test_samples = None
+        send_buf_train_samples = None
+        send_buf_train_targets = None
+
+    n_features = comm.bcast(n_features, root=0)
+    n_test_samples = comm.bcast(n_test_samples, root=0)
+    train_counts = comm.bcast(train_counts, root=0)
+    samples_train_local = np.empty((train_counts[rank], n_features), dtype=float)
+    targets_train_local = np.empty((train_counts[rank],), dtype=float)
+    recv_buf_train_samples = [
+        samples_train_local,
+        from_numpy_dtype(samples_train_local.dtype),
+    ]
+    recv_buf_train_targets = [
+        targets_train_local,
+        from_numpy_dtype(targets_train_local.dtype),
+    ]
+    if rank != 0:
+        samples_test = np.empty((n_test_samples, n_features), dtype=float)
+        targets_test = np.empty((n_test_samples,), dtype=float)
+    comm.Scatterv(send_buf_train_samples, recv_buf_train_samples, root=0)
+    comm.Scatterv(send_buf_train_targets, recv_buf_train_targets, root=0)
+    comm.Bcast(samples_test, root=0)
+    comm.Bcast(targets_test, root=0)
+    if verbose:
+        print_str = f"[{rank}/{size}]: Train samples after Scatterv have shape {samples_train_local.shape}.\n"
+        print_str += (
+            f"Train targets after Scatterv have shape {targets_train_local.shape}.\n"
+        )
+        print_str += f"Test samples after Bcast have shape {samples_test.shape}.\n"
+        print_str += f"Test targets after Bcast have shape {targets_test.shape}."
+        print(print_str)
+    return samples_train_local, targets_train_local, samples_test, targets_test

3_ensembles/solutions/py/distributed_forest.py

+"Distributed random forest in sklearn with MPI."
+import argparse
+import string
+from typing import Tuple
+import time
+
+from mpi4py import MPI
+import numpy as np
+from sklearn.ensemble import RandomForestClassifier
+
+from dataloaders import load_data_csv_parallel, load_data_csv_root
+
+
+class MPITimer:
+    """A distributed context-manager enabled timer."""
+
+    def __init__(
+        self,
+        comm: MPI.Comm,
+        print_on_exit: bool = True,
+        name: str = "",
+        output_format: str = "Elapsed time {name}: global average {elapsed_time_average:.2g}s, "
+        "local {elapsed_time_local:.2g}s",
+    ) -> None:
+        """
+        Create a new distributed context-manager enabled timer.
+
+        Parameters
+        ----------
+        comm : MPI.Comm
+            The MPI communicator.
+        print_on_exit : bool
+            Whether to print the measured time in __exit__.
+        name : str
+            Label describing what this timer measured, can be used for printing the results.
+        output_format : str
+            Format string template used for printing the output. May reference all attributes of the
+        """
+        self.comm = comm
+        self.output_format = output_format
+        self.print_on_exit = print_on_exit
+        self.name = name
+
+        self.start_time = None
+        self.end_time = None
+        self.elapsed_time_local = None
+        self.elapsed_time_average = None
+
+    def start(self) -> None:
+        """Start the timer by setting the start time."""
+        self.start_time = time.perf_counter()
+
+    def stop(self) -> None:
+        """Stop the timer by setting the end time and updating elapsed_time_local."""
+        self.end_time = time.perf_counter()
+        self.elapsed_time_local = self.end_time - self.start_time
+
+    def allreduce_for_average_time(self) -> None:
+        """Compute the global average using allreduce and update elapsed_time_average."""
+        self.elapsed_time_average = (
+            self.comm.allreduce(self.elapsed_time_local, op=MPI.SUM) / self.comm.size
+        )
+
+    def print(self) -> None:
+        """Print the elapsed time using the given template."""
+        template_keywords = {
+            key for (_, key, _, _) in string.Formatter().parse(self.output_format)
+        }
+        template_kwargs = {
+            key: value for key, value in vars(self).items() if key in template_keywords
+        }
+        print(self.output_format.format(**template_kwargs))
+
+    def __enter__(self) -> "MPITimer":
+        """
+        Called on entering this context (e.g. with a 'with' statement), starts the timer.
+
+        Returns
+        -------
+        MPITimer
+            This timer object.
+        """
+        self.start()
+        return self
+
+    def __exit__(self, *args) -> None:
+        """
+        Called on exiting this context (e.g. after a 'with' statement), stops the timer, computes the global average
+        and optionally prints the result on rank 0.
+
+        Parameters
+        ----------
+        args :
+            Unused, only to fulfill __exit__ interface.
+        """
+        self.stop()
+        self.allreduce_for_average_time()
+        if self.print_on_exit and self.comm.rank == 0:
+            self.print()
+
+
+class DistributedRandomForest:
+    """
+    Distributed random forest class.
+
+    Attributes
+    ----------
+    acc_global : None | float
+        The global accuracy.
+    acc_local : None | float
+        Each local trained model's accuracy.
+    clf : None | RandomForestClassifier
+        The sklearn classifier.
+    comm : MPI.Comm
+        The communicator.
+    n_trees_base : int
+        The base number of rank-local trees.
+    n_trees_global : int
+        The number of trees in the global forest.
+    n_trees_local : int
+        The number of local trees.
+    n_trees_remainder : int
+        The remaining number of trees to distribute.
+    random_state : int
+        The base random state for sklearn random forest.
+
+    Methods
+    -------
+    _distribute_trees()
+        Distribute trees evenly over all processors.
+    _train_local_classifier()
+        Train local random forest classifier.
+    _predict_locally()
+        Get predictions of all sub estimators in random forest.
+    _get_predicted_class_hist()
+        Calculate global sample-wise distributions of predicted classes from rank-local tree-wise predictions.
+    _calc_majority_vote_hist()
+        Calculate majority vote from sample-wise histograms of predicted classes.
+    train()
+        Train random forest in parallel.
+    test()
+        Test trained distributed global random forest.
+    """
+
+    def __init__(
+        self,
+        n_trees_global: int,
+        comm: MPI.Comm,
+        random_state: int,
+    ) -> None:
+        """
+        Initialize distributed random forest object.
+
+        Parameters
+        ----------
+        n_trees_global : int
+            The number of trees in the global forest.
+        comm : MPI.Comm
+            The communicator to use.
+        random_state : int
+            The base random state for sklearn RF.
+        """
+        self.n_trees_global = n_trees_global
+        self.comm = comm
+        (
+            self.n_trees_base,
+            self.n_trees_remainder,
+            self.n_trees_local,
+        ) = self._distribute_trees()
+        self.random_state = random_state + self.comm.rank
+        self.clf = None  # local random forest classifier
+        self.acc_global = None  # accuracy of global classifier
+        self.acc_local = None  # accuracy of each rank-local classifier
+
+    def _distribute_trees(self) -> Tuple[int, int, int]:
+        """
+        Distribute trees evenly over all processors.
+
+        Returns
+        -------
+        int
+            The base number of rank-local trees.
+        int
+            The remaining number of trees to distribute.
+        int
+            The final number of rank-local trees.
+        """
+        size, rank = self.comm.size, self.comm.rank
+        n_trees_base = self.n_trees_global // size  # base number of rank-local trees
+        n_trees_remainder = (
+            self.n_trees_global % size
+        )  # remaining number of trees to distribute
+        n_trees_local = (
+            n_trees_base + 1 if rank < n_trees_remainder else n_trees_base
+        )  # final number of local trees
+        return n_trees_base, n_trees_remainder, n_trees_local
+
+    def _train_local_classifier(
+        self, train_samples: np.ndarray, train_targets: np.ndarray
+    ) -> RandomForestClassifier:
+        """
+        Train local random forest classifier.
+
+        Params
+        ------
+        train_samples : numpy.ndarray
+            samples of train dataset
+        train_targets : numpy.ndarray
+            targets of train dataset
+
+        Returns
+        -------
+        sklearn.ensemble.RandomForestClassifier
+            trained model
+        """
+        clf = RandomForestClassifier(
+            n_estimators=self.n_trees_local, random_state=self.random_state
+        )
+        _ = clf.fit(train_samples, train_targets)
+        return clf
+
+    def _predict_locally(self, samples: np.ndarray) -> np.ndarray:
+        """
+        Get predictions of all sub estimators in random forest.
+
+        Parameters
+        ----------
+        samples : numpy.ndarray
+            The samples whose class to predict.
+
+        Returns
+        -------
+        numpy.ndarray
+            The predictions of each sub estimator on the samples.
+        """
+        return np.array(
+            [tree.predict(samples) for tree in self.clf.estimators_], dtype="H"
+        )
+
+    def _get_predicted_class_hist(
+        self, tree_wise_predictions: np.ndarray, n_classes: int
+    ) -> np.ndarray:
+        """
+        Calculate global sample-wise distributions of predicted classes from rank-local tree-wise predictions.
+
+        Parameters
+        ----------
+        tree_wise_predictions : numpy.ndarray
+            The tree-wise predictions.
+        n_classes : int
+            The number of classes in the dataset.
+
+        Returns
+        -------
+        numpy.ndarray
+            The sample-wise distributions of the predicted classes.
+        """
+        sample_wise_predictions = tree_wise_predictions.transpose()
+        predicted_class_hists_local = np.array(
+            [
+                np.bincount(sample_pred, minlength=n_classes)
+                for sample_pred in sample_wise_predictions
+            ]
+        )
+        # The hist arrays have `n_test_samples` entries with `n_classes` elements each.
+        # The local hist holds the class prediction distribution for each test sample over the local forest.
+        # Now we want to sum up those sample-wise distributions to obtain the global hist over the global forest.
+        # From this global hist, we can determine the global majority vote for each test sample.
+        predicted_class_hists_global = np.zeros_like(predicted_class_hists_local)
+        self.comm.Allreduce(predicted_class_hists_local, predicted_class_hists_global)
+        return predicted_class_hists_global
+
+    @staticmethod
+    def _calc_majority_vote_hist(predicted_class_hists: np.ndarray) -> np.ndarray:
+        """
+        Calculate majority vote from sample-wise histograms of predicted classes.
+
+        Parameters
+        ----------
+        predicted_class_hists : numpy.ndarray
+            sample-wise histograms of predicted classes
+
+        Returns
+        -------
+        numpy.ndarray
+            majority votes for all samples in histogram input
+        """
+        return np.array(
+            [np.argmax(sample_hist) for sample_hist in predicted_class_hists]
+        )
+
+    def train(
+        self,
+        train_samples: np.ndarray,
+        train_targets: np.ndarray,
+    ) -> None:
+        """
+        Train random forest in parallel.
+
+        Parameters
+        ----------
+        train_samples : numpy.ndarray
+            rank-local train samples
+        train_targets : numpy.ndarray
+            corresponding train targets
+        """
+        # Set up communicator.
+        rank, size = self.comm.rank, self.comm.size
+        # Set up and train local forest.
+        print(
+            f"[{rank}/{size}]: Set up and train local random forest with "
+            f"{self.n_trees_local} trees and random state {self.random_state}."
+        )
+        self.clf = self._train_local_classifier(
+            train_samples=train_samples,
+            train_targets=train_targets,
+        )
+
+    def test(
+        self,
+        test_samples: np.ndarray,
+        test_targets: np.ndarray,
+        n_classes: int,
+    ) -> None:
+        """
+        Test trained distributed global random forest.
+
+        Parameters
+        ----------
+        test_samples : numpy.ndarray
+            The rank-local test samples.
+        test_targets : numpy.ndarray
+            The corresponding test targets.
+        n_classes : int
+            The number of classes in the dataset.
+        """
+        rank, size = self.comm.rank, self.comm.size
+        # Get class predictions of sub estimators in each forest.
+        print(f"[{rank}/{size}]: Get predictions of individual sub estimators.")
+        tree_predictions_local = self._predict_locally(test_samples)
+        print(f"[{rank}/{size}]: Calculate majority vote via histograms.")
+        predicted_class_hists = self._get_predicted_class_hist(
+            tree_predictions_local, n_classes
+        )
+        majority_vote = self._calc_majority_vote_hist(predicted_class_hists)
+        # Calculate accuracies.
+        self.acc_global = (test_targets == majority_vote).mean()
+        self.acc_local = self.clf.score(test_samples, test_targets)
+        print(
+            f"[{rank}/{size}]: Local accuracy is {self.acc_local}, global accuracy is {self.acc_global}."
+        )
+
+
+if __name__ == "__main__":
+    # SETTINGS
+    comm = MPI.COMM_WORLD  # Set up communicator.
+    rank, size = comm.rank, comm.size
+
+    if rank == 0:
+        print(
+            "######################################################\n"
+            "# Distributed Random Forest in Scikit-Learn with MPI #\n"
+            "######################################################\n"
+        )
+
+    n_classes = 2
+
+    header_lines = 0  # Data file specifications
+    sep = ","
+    encoding = "utf-8"
+    path_to_data = "/pfs/work7/workspace/scratch/ku4408-VL-ScalableAI/data/SUSY.csv"
+
+    parser = argparse.ArgumentParser(prog="Distributed Random Forest")
+    parser.add_argument(
+        "--dataloader",
+        type=str,
+        choices=["root", "parallel"],
+        default="root",
+        help="The dataloader to use.",
+    )
+    parser.add_argument(
+        "--num_trees",
+        type=int,
+        default=100,
+        help="The global number of trees.",
+    )
+    parser.add_argument(
+        "--train_split",
+        type=float,
+        default=0.75,
+        help="The fraction of the dataset used for training.",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="The verbosity.",
+    )
+    args = parser.parse_args()
+
+    seed = size
+
+    # -------------- Load data. --------------
+    print(f"[{rank}/{size}]: Loading data...")
+    with MPITimer(comm, name="data loading"):
+        if args.dataloader == "parallel":
+            if rank == 0:
+                print("Using truly parallel dataloader...")
+            (
+                train_samples_local,
+                train_targets_local,
+                test_samples,
+                test_targets,
+            ) = load_data_csv_parallel(
+                path_to_data, header_lines, comm, args.train_split, args.verbose
+            )
+
+        elif args.dataloader == "root":
+            if rank == 0:
+                print("Using root-based dataloader with Scatterv...")
+            (
+                train_samples_local,
+                train_targets_local,
+                test_samples,
+                test_targets,
+            ) = load_data_csv_root(
+                path_to_data, header_lines, comm, seed, args.train_split, args.verbose
+            )
+        else:
+            raise ValueError
+
+    print(
+        f"[{rank}/{size}]: DONE.\n"
+        f"Local train samples and targets have shapes {train_samples_local.shape} and {train_targets_local.shape}.\n"
+        f"Global test samples and targets have shapes {test_samples.shape} and {test_targets.shape}.\n"
+        f"Labels are {train_targets_local}"
+    )
+
+    # -------------- Set up random forest. --------------
+    with MPITimer(comm, name="forest creation"):
+        distributed_random_forest = DistributedRandomForest(
+            n_trees_global=args.num_trees,
+            comm=comm,
+            random_state=seed,
+        )
+
+    # -------------- Train random forest. --------------
+    with MPITimer(comm, name="training"):
+        distributed_random_forest.train(
+            train_samples_local,
+            train_targets_local,
+        )
+
+    # -------------- Evaluate random forest. --------------
+    print(f"[{rank}/{size}]: Evaluate random forest.")
+    with MPITimer(comm, name="test"):  # Test trained model on test data.
+        distributed_random_forest.test(
+            test_samples,
+            test_targets,
+            n_classes,
+        )

3_ensembles/solutions/py/serial_forest.py

+"""Serial random forest in sklearn."""
+import argparse
+import pathlib
+from typing import Union
+import time
+
+import numpy as np
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.model_selection import train_test_split
+
+
+def load_data_csv(
+    path_to_data: Union[str, pathlib.Path],
+    train_split: float = 0.75,
+    seed: int = 42,
+):
+    """
+    Load data from CSV file via root process.
+
+    Params
+    ------
+    path_to_data : str
+        The path to .csv file.
+    train_split : float
+        The train-test split fraction.
+    seed : int
+        The seed used for train-test split.
+
+    Returns
+    -------
+    np.ndarray
+        The train samples.
+    np.ndarray
+        The train targets.
+    np.ndarray
+        The test samples.
+    nup.ndarray
+        The test targets.
+    """
+    # Load data into numpy array.
+    data = np.loadtxt(path_to_data, dtype=float, delimiter=",")
+    # Divide data into samples and targets.
+    samples = data[:, 1:]
+    targets = data[:, 0]
+    # Perform train-test split.
+    samples_train, samples_test, targets_train, targets_test = train_test_split(
+        samples, targets, test_size=1 - train_split, random_state=seed
+    )
+    return samples_train, targets_train, samples_test, targets_test
+
+
+if __name__ == "__main__":
+    print(
+        "########################################\n"
+        "# Serial Random Forest in Scikit-Learn #\n"
+        "########################################\n"
+    )
+    parser = argparse.ArgumentParser(prog="Serial Random Forest")
+
+    parser.add_argument(
+        "--num_trees",
+        type=int,
+        default=100,
+        help="The global number of trees.",
+    )
+    parser.add_argument(
+        "--train_split",
+        type=float,
+        default=0.75,
+        help="The fraction of the dataset used for training.",
+    )
+    parser.add_argument(
+        "--seed",
+        type=int,
+        default=42,
+        help="The random seed.",
+    )
+    args = parser.parse_args()
+    path_to_data = "/pfs/work7/workspace/scratch/ku4408-VL-ScalableAI/data/SUSY.csv"
+    print("Loading data...")
+    start_load = time.perf_counter()
+    (
+        train_samples,
+        train_targets,
+        test_samples,
+        test_targets,
+    ) = load_data_csv(path_to_data, args.train_split, args.seed)
+    elapsed_load = time.perf_counter() - start_load
+    print(
+        "DONE.\n"
+        f"Train samples and targets have shapes {train_samples.shape} and {train_targets.shape}.\n"
+        f"First ten elements are: {train_samples[:10]} and {train_targets[:10]}\n"
+        f"Test samples and targets have shapes {test_samples.shape} and {test_targets.shape}.\n"
+        f"First ten elements are: {test_samples[:10]} and {test_targets[:10]}\n"
+        f"Time for data loading is {elapsed_load} s.\n"
+        "Set up classifier."
+    )
+    classifier = RandomForestClassifier(
+        n_estimators=args.num_trees, random_state=args.seed
+    )
+    start_train = time.perf_counter()
+    print("Train.")
+    _ = classifier.fit(train_samples, train_targets)
+    acc = classifier.score(test_samples, test_targets)
+    elapsed_train = time.perf_counter() - start_train
+    print(f"Time for training is {elapsed_train} s.\n Accuracy is {acc}.")

3_ensembles/solutions/py/test_dataloaders.py

+"""Test dataloaders."""
+import argparse
+
+from mpi4py import MPI
+
+from dataloaders import load_data_csv_parallel, load_data_csv_root
+from distributed_forest import MPITimer
+
+if __name__ == "__main__":
+    # Set up communicator.
+    comm = MPI.COMM_WORLD
+    rank, size = comm.rank, comm.size
+
+    # Data file specifications
+    header_lines = 0
+    sep = ","
+    encoding = "utf-8"
+    path_to_data = "/pfs/work7/workspace/scratch/ku4408-VL-ScalableAI/data/SUSY.csv"
+
+    parser = argparse.ArgumentParser(prog="Distributed Random Forest")
+    parser.add_argument(
+        "--train_split",
+        type=float,
+        default=0.75,
+        help="The fraction of the dataset used for training.",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="The verbosity.",
+    )
+    args = parser.parse_args()
+
+    seed = size
+
+    print(f"[{rank}/{size}]: Loading data using truly parallel dataloader...")
+    with MPITimer(comm, name="truly parallel data loading"):
+        (
+            train_samples_local_par,
+            train_targets_local_par,
+            test_samples_par,
+            test_targets_par,
+        ) = load_data_csv_parallel(
+            path_to_data, header_lines, comm, args.train_split, args.verbose
+        )
+
+    print(f"[{rank}/{size}]: Loading data using root-based dataloader...")
+    with MPITimer(comm, name="root-based data loading"):
+        (
+            train_samples_local_root,
+            train_targets_local_root,
+            test_samples_root,
+            test_targets_root,
+        ) = load_data_csv_root(
+            path_to_data, header_lines, comm, seed, args.train_split, args.verbose
+        )
+
+    print(
+        f"[{rank}/{size}]: DONE.\n"
+        f"Parallel: Local train samples / targets have shapes "
+        f"{train_samples_local_par.shape} / {train_targets_local_par.shape}.\n"
+        f"Parallel: Global test samples / targets have shapes {test_samples_par.shape} / {test_targets_par.shape}.\n"
+        f"Root: Local train samples / targets have shapes "
+        f"{train_samples_local_root.shape} / {train_targets_local_root.shape}.\n"
+        f"Root: Global test samples / targets have shapes {test_samples_root.shape} / {test_targets_root.shape}."
+    )

3_ensembles/solutions/sh/slurm_dataloaders.out

+[1/2]: Loading data using truly parallel dataloader...
+[0/2]: Loading data using truly parallel dataloader...
+File size is 2390277560 bytes.
+After Allgatherv: All line starts: [         0        479        957 ... 2390276125 2390276602 2390277082]
+[0/2]: Construct array with line starts and lengths in bytes.
+[1/2]: Construct array with line starts and lengths in bytes.
+[0/2]: Make global train-test split.
+[1/2]: Make global train-test split.
+[0/2]: Decode 1250000 test samples from file.
+[1/2]: Decode 1250000 test samples from file.
+[0/2]: Draw local 1875000 train indices.
+[0/2]: Decode train lines from file.
+[1/2]: Draw local 1875000 train indices.
+[1/2]: Decode train lines from file.
+Elapsed time truly parallel data loading: global average 2.5e+02s, local 2.5e+02s
+[0/2]: Loading data using root-based dataloader...
+[1/2]: Loading data using root-based dataloader...
+There are 3750000 train and 1250000 test samples.
+Local train samples: [1875000 1875000]
+train_indices have shape (3750000,).
+Elapsed time root-based data loading: global average 36s, local 36s
+[0/2]: DONE.
+Parallel: Local train samples / targets have shapes (1875000, 18) / (1875000,).
+Parallel: Global test samples / targets have shapes (1250000, 18) / (1250000,).
+Root: Local train samples / targets have shapes (1875000, 18) / (1875000,).
+Root: Global test samples / targets have shapes (1250000, 18) / (1250000,).
+[1/2]: DONE.
+Parallel: Local train samples / targets have shapes (1875000, 18) / (1875000,).
+Parallel: Global test samples / targets have shapes (1250000, 18) / (1250000,).
+Root: Local train samples / targets have shapes (1875000, 18) / (1875000,).
+Root: Global test samples / targets have shapes (1250000, 18) / (1250000,).
+
+============================= JOB FEEDBACK =============================
+
+NodeName=uc2n[113,116]
+Job ID: 22915408
+Cluster: uc2
+User/Group: ku4408/scc
+State: COMPLETED (exit code 0)
+Nodes: 2
+Cores per node: 80
+CPU Utilized: 00:09:44
+CPU Efficiency: 1.12% of 14:26:40 core-walltime
+Job Wall-clock time: 00:05:25
+Memory Utilized: 2.58 GB
+Memory Efficiency: 1.47% of 175.78 GB

3_ensembles/solutions/sh/slurm_parallel.out

+[1/2]: Loading data...
+######################################################
+# Distributed Random Forest in Scikit-Learn with MPI #
+######################################################
+
+[0/2]: Loading data...
+Using truly parallel dataloader...
+File size is 2390277560 bytes.
+[1/2]: Construct array with line starts and lengths in bytes.
+After Allgatherv: All line starts: [         0        479        957 ... 2390276125 2390276602 2390277082]
+[0/2]: Construct array with line starts and lengths in bytes.
+[1/2]: Make global train-test split.
+[0/2]: Make global train-test split.
+[1/2]: Decode 1250000 test samples from file.
+[0/2]: Decode 1250000 test samples from file.
+[1/2]: Draw local 1875000 train indices.
+[1/2]: Decode train lines from file.
+[0/2]: Draw local 1875000 train indices.
+[0/2]: Decode train lines from file.
+Elapsed time data loading: global average 2.5e+02s, local 2.5e+02s
+[0/2]: DONE.
+Local train samples and targets have shapes (1875000, 18) and (1875000,).
+Global test samples and targets have shapes (1250000, 18) and (1250000,).
+Labels are [0. 0. 0. ... 0. 0. 1.]
+Elapsed time forest creation: global average 2.8e-05s, local 3.8e-05s
+[0/2]: Set up and train local random forest with 50 trees and random state 2.
+[1/2]: DONE.
+Local train samples and targets have shapes (1875000, 18) and (1875000,).
+Global test samples and targets have shapes (1250000, 18) and (1250000,).
+Labels are [1. 0. 0. ... 0. 0. 1.]
+[1/2]: Set up and train local random forest with 50 trees and random state 3.
+Elapsed time training: global average 8.8e+02s, local 8.8e+02s
+[0/2]: Evaluate random forest.
+[0/2]: Get predictions of individual sub estimators.
+[1/2]: Evaluate random forest.
+[1/2]: Get predictions of individual sub estimators.
+[1/2]: Calculate majority vote via histograms.
+[0/2]: Calculate majority vote via histograms.
+[1/2]: Local accuracy is 0.7971136, global accuracy is 0.7998568.
+[0/2]: Local accuracy is 0.79728, global accuracy is 0.7998568.
+Elapsed time test: global average 56s, local 56s
+[1/2]: Loading data...
+######################################################
+# Distributed Random Forest in Scikit-Learn with MPI #
+######################################################
+
+[0/2]: Loading data...
+Using root-based dataloader with Scatterv...
+There are 3750000 train and 1250000 test samples.
+Local train samples: [1875000 1875000]
+train_indices have shape (3750000,).
+Elapsed time data loading: global average 35s, local 35s
+[0/2]: DONE.
+Local train samples and targets have shapes (1875000, 18) and (1875000,).
+Global test samples and targets have shapes (1250000, 18) and (1250000,).
+Labels are [0. 0. 0. ... 1. 0. 0.]
+Elapsed time forest creation: global average 2e-05s, local 1.9e-05s
+[0/2]: Set up and train local random forest with 50 trees and random state 2.
+[1/2]: DONE.
+Local train samples and targets have shapes (1875000, 18) and (1875000,).
+Global test samples and targets have shapes (1250000, 18) and (1250000,).
+Labels are [0. 1. 1. ... 0. 1. 0.]
+[1/2]: Set up and train local random forest with 50 trees and random state 3.
+Elapsed time training: global average 8.8e+02s, local 8.8e+02s
+[0/2]: Evaluate random forest.
+[0/2]: Get predictions of individual sub estimators.
+[1/2]: Evaluate random forest.
+[1/2]: Get predictions of individual sub estimators.
+[0/2]: Calculate majority vote via histograms.
+[1/2]: Calculate majority vote via histograms.
+[0/2]: Local accuracy is 0.7977704, global accuracy is 0.800064.
+[1/2]: Local accuracy is 0.7974024, global accuracy is 0.800064.
+Elapsed time test: global average 57s, local 57s
+
+============================= JOB FEEDBACK =============================
+
+NodeName=uc2n[222,237]
+Job ID: 22915390
+Cluster: uc2
+User/Group: ku4408/scc
+State: COMPLETED (exit code 0)
+Nodes: 2
+Cores per node: 80
+CPU Utilized: 01:11:56
+CPU Efficiency: 1.23% of 4-01:36:00 core-walltime
+Job Wall-clock time: 00:36:36
+Memory Utilized: 3.12 GB
+Memory Efficiency: 1.77% of 175.78 GB

3_ensembles/solutions/sh/slurm_serial.out

+########################################
+# Serial Random Forest in Scikit-Learn #
+########################################
+
+Loading data...
+DONE.
+Train samples and targets have shapes (3750000, 18) and (3750000,).
+First ten elements are: [[ 1.43997777e+00  1.63248479e+00 -9.67991173e-01  5.68317890e-01
+   1.46224272e+00  1.36972353e-01  8.19436967e-01  9.11546707e-01
+   1.21039116e+00 -5.06843209e-01  9.31933105e-01  1.27073431e+00
+   1.21000016e+00  1.47810447e+00  8.52468789e-01  1.11379075e+00
+   2.47141235e-02  6.13238990e-01]
+ [ 3.43844175e-01 -7.04108357e-01 -1.51597571e+00  5.51238716e-01
+   5.92378020e-01  1.29997504e+00  6.11836970e-01  2.91105419e-01
+   9.18442786e-01 -1.56051174e-01  4.62793738e-01  6.52638137e-01
+   1.25141346e+00  1.32741857e+00  4.72825408e-01  1.00110984e+00
+   9.48828042e-01  1.48614004e-01]
+ [ 5.07223248e-01  3.59210372e-01  4.22794193e-01  7.39285111e-01
+   8.21336746e-01 -2.84344971e-01  6.44737303e-01 -1.26930571e+00
+   8.19674611e-01 -1.98983908e-01  5.15161872e-01  7.95682371e-01
+   1.37060583e+00  1.57789564e+00  4.93557423e-01  1.17942798e+00
+   7.52048492e-01  2.89388001e-01]
+ [ 5.56809664e-01 -1.57331979e+00  1.35683000e+00  7.13156343e-01
+   7.61644185e-01  1.53109705e+00  7.97575951e-01 -2.99961656e-01
+   1.19725823e+00 -4.84998196e-01  1.09257638e+00  9.92057800e-01
+   8.05751562e-01  2.16137958e+00  1.08722878e+00  1.62078691e+00
+   4.51993123e-02  3.25273015e-02]
+ [ 3.59211493e+00  2.13520462e-03 -6.44264281e-01  1.90623689e+00
+   3.20609003e-01  3.08485538e-01  3.18263960e+00 -2.74604857e-01
+   2.91751528e+00  3.33075738e+00  2.50526094e+00  1.63983834e+00
+   5.80852270e-01  0.00000000e+00  1.69048905e+00  5.32282293e-01
+   7.83510923e-01  1.63441002e-01]
+ [ 7.58597434e-01 -1.89242971e+00 -1.67973864e+00  6.62953973e-01
+  -1.17093217e+00 -1.02506316e+00  1.43694293e+00  3.01254123e-01
+   2.15702868e+00 -1.12557161e+00  6.50241256e-01  1.39164054e+00
+   1.89918971e+00  2.74402332e+00  7.75830925e-01  2.08014536e+00
+   1.54032695e+00  4.80755001e-01]
+ [ 9.87198830e-01  1.21461833e+00  1.33140802e-01  1.17224276e+00
+  -9.93977427e-01 -9.41504121e-01  6.46450996e-01  1.56007898e+00
+   9.70402718e-01 -3.25230628e-01  1.75528979e+00  1.04119718e+00
+   5.26380122e-01  1.83662927e+00  1.74301934e+00  1.45607185e+00
+   2.37897053e-01  4.61494997e-02]
+ [ 1.70994639e+00 -8.22762012e-01 -1.13427246e+00  9.53481674e-01
+  -1.90318573e+00  4.29294139e-01  1.42728555e+00  3.82643938e-01
+   2.05661044e-01 -9.03856218e-01  1.40107632e+00  1.72245789e+00
+   1.09095061e+00  0.00000000e+00  1.44269097e+00  1.23851955e+00
+   1.05762923e+00  4.25258994e-01]
+ [ 9.43593442e-01  1.99409112e-01  8.14792871e-01  7.85433173e-01
+  -4.54714209e-01 -1.14372945e+00  5.78886461e+00  6.90133393e-01
+   2.04307199e+00  5.17880249e+00  7.89597631e-01  1.86898780e+00
+   2.10046601e+00  1.05997956e+00  1.11436117e+00  1.30555189e+00
+   1.50189817e+00  7.06036985e-01]
+ [ 7.64213026e-01  1.81087300e-01 -1.32228279e+00  6.47381306e-01
+  -7.69020736e-01 -2.73123175e-01  1.21376109e+00  7.54100859e-01
+   1.82200515e+00 -8.98288131e-01  6.92662835e-01  1.20266879e+00
+   1.54078197e+00  2.04965854e+00  7.89022744e-01  1.62554586e+00
+   1.52046657e+00  3.61533999e-01]] and [0. 0. 1. 0. 1. 0. 0. 1. 1. 0.]
+Test samples and targets have shapes (1250000, 18) and (1250000,).
+First ten elements are: [[ 6.91279709e-01  1.76920056e+00 -1.71181107e+00  4.76238310e-01
+   1.54056251e+00 -2.67858446e-01  8.72983932e-01  2.78744638e-01
+   1.08713210e+00 -5.51878333e-01  5.12543857e-01  8.91626298e-01
+   1.54371738e+00  8.60364318e-01  5.83059669e-01  6.91559553e-01
+   1.50008941e+00  5.59077024e-01]
+ [ 4.17667389e-01 -1.88861191e+00 -2.25399002e-01  5.72387338e-01
+  -1.19409788e+00  7.04542026e-02  6.95429265e-01  1.66337061e+00
+   1.04392505e+00 -3.82097363e-01  4.27827954e-01  8.09540808e-01
+   1.67913723e+00  1.76087189e+00  4.39944327e-01  1.31074250e+00
+   1.36796749e+00  1.91391006e-01]
+ [ 2.74744093e-01 -1.38396299e+00 -1.03303587e+00  4.39079940e-01
+   1.67037070e+00 -6.75462365e-01  7.48247623e-01  9.07953739e-01
+   1.12321198e+00 -4.34616148e-01  9.00804639e-01  7.02887774e-01
+   6.92424834e-01  1.52986670e+00  9.32529151e-01  1.16853487e+00
+   1.54546940e+00  2.30877008e-03]
+ [ 1.16500580e+00  1.00420511e+00 -1.22337127e+00  1.62781668e+00
+  -3.76755238e-01  8.39448214e-01  2.00318694e+00  1.60608697e+00
+   2.76874542e+00  2.29274392e+00  1.51524031e+00  1.37557471e+00
+   8.05598021e-01  0.00000000e+00  1.52915168e+00  1.25636113e+00
+   1.52380025e+00  4.66813985e-03]
+ [ 1.40449703e+00 -1.45557001e-01  5.21603674e-02  6.55963719e-01
+   1.24605799e+00  1.40364683e+00  6.71340346e-01 -1.68536985e+00
+   6.29944921e-01 -3.38737279e-01  1.19020391e+00  1.10914612e+00
+   8.26955855e-01  7.59544134e-01  1.14556456e+00  1.11023533e+00
+   2.60205507e-01  3.26079011e-01]
+ [ 2.63595748e+00  6.21905744e-01  1.27720702e+00  2.45305467e+00
+   1.10806942e+00 -5.16619682e-01  7.82781899e-01  1.04785419e+00
+   4.86200094e-01  9.83858526e-01  2.23186541e+00  1.23496139e+00
+   4.91022170e-01  6.09043658e-01  2.15673685e+00  4.74038422e-01
+   5.26653826e-01  4.24163006e-02]
+ [ 8.90232325e-01 -9.47344065e-01  6.88447535e-01  7.76006341e-01
+  -1.44117546e+00 -4.05649900e-01  4.55961943e-01  3.85926753e-01
+   3.63408327e-01  7.74472058e-01  7.38129973e-01  3.90631944e-01
+   4.69624609e-01  0.00000000e+00  6.06057763e-01  2.01749176e-01
+   2.38361638e-02  6.27153963e-02]
+ [ 8.09924126e-01  1.76636660e+00  1.50027168e+00  8.24404478e-01
+   9.41239297e-01 -9.60501552e-01  1.00532985e+00 -1.67402878e-01
+   1.49302173e+00 -5.72425246e-01  7.54992843e-01  1.13767517e+00
+   1.33718264e+00  1.82880926e+00  7.92772830e-01  1.42058229e+00
+   1.07851815e+00  3.78154993e-01]
+ [ 2.71482444e+00 -1.04809391e+00 -9.96524235e-04  2.07914996e+00
+  -7.31364310e-01 -1.69832718e+00  1.54993641e+00  1.68483472e+00
+   3.84886861e-01 -1.23952270e+00  2.09488463e+00  2.30906534e+00
+   9.78124380e-01  3.79600048e-01  2.12716794e+00  5.43927491e-01
+   1.45123398e+00  3.90201986e-01]
+ [ 6.09283328e-01 -8.08349609e-01  2.94039518e-01  9.17876959e-01
+  -3.87946656e-03  1.57775986e+00  9.13081244e-02  1.04072893e+00
+   1.14134960e-01  4.03893739e-01  6.74048424e-01  2.03140706e-01
+   2.67436147e-01  0.00000000e+00  6.38434172e-01  3.05932641e-01
+   7.21062347e-02  2.93387994e-02]] and [0. 0. 1. 1. 1. 1. 1. 1. 1. 0.]
+Time for data loading is 39.535260654985905 s.
+Set up classifier.
+Train.
+Time for training is 4739.061137255281 s.
+ Accuracy is 0.8003664.
+
+============================= JOB FEEDBACK =============================
+
+NodeName=uc2n378
+Job ID: 22915357
+Cluster: uc2
+User/Group: ku4408/scc
+State: COMPLETED (exit code 0)
+Nodes: 1
+Cores per node: 40
+CPU Utilized: 01:19:24
+CPU Efficiency: 2.47% of 2-05:34:40 core-walltime
+Job Wall-clock time: 01:20:22
+Memory Utilized: 7.58 GB
+Memory Efficiency: 17.25% of 43.95 GB

3_ensembles/solutions/sh/submit_parallel.sh

+#!/bin/bash
+
+#SBATCH --job-name=RF2                     # Job name
+#SBATCH --partition=multiple               # Queue for the resource allocation
+#SBATCH --nodes=2                          # Number of nodes
+#SBATCH --time=70:00                       # Wall-clock time limit
+#SBATCH --ntasks-per-node=1                # maximum count of tasks per node
+#SBATCH --cpus-per-task=40                 # Number of CPUs per task
+#SBATCH --mail-type=ALL                    # Notify user by email when certain event types occur.
+
+export OMP_NUM_THREADS=${SLURM_CPUS_PER_TASK}
+export VENVDIR=<path/to/your/venv>         # Export path to your virtual environment.
+export PYDIR=<path/to/your/python/script>  # Export path to directory containing Python script.
+
+# Set up modules.
+module purge                               # Unload all currently loaded modules.
+module load compiler/gnu/13.3              # Load required modules.
+module load mpi/openmpi/4.1
+module load devel/cuda/12.4
+module load lib/hdf5/1.14.4-gnu-13.3-openmpi-4.1
+
+source ${VENVDIR}/bin/activate             # Activate your virtual environment.
+
+mpirun python ${PYDIR}/distributed_forest.py --dataloader parallel # Use truly parallel dataloader.
+mpirun python ${PYDIR}/distributed_forest.py --dataloader root  # Use root-based dataloader.

3_ensembles/solutions/sh/submit_serial.sh

+#!/bin/bash
+
+#SBATCH --job-name=RF1                     # Job name
+#SBATCH --partition=single                 # Queue for the resource allocation
+#SBATCH --time=24:00:00                    # Wall-clock time limit
+#SBATCH --cpus-per-task=40                 # Number of CPUs per task
+#SBATCH --mail-type=ALL                    # Notify user by email when certain event types occur.
+
+export OMP_NUM_THREADS=${SLURM_CPUS_PER_TASK}
+export VENVDIR=<path/to/your/venv>         # Export path to your virtual environment.
+export PYDIR=<path/to/your/python/script>  # Export path to directory containing Python script.
+
+# Set up modules.
+module purge                               # Unload all currently loaded modules.
+module load compiler/gnu/13.3              # Load required modules.
+module load mpi/openmpi/4.1
+module load devel/cuda/12.4
+module load lib/hdf5/1.14.4-gnu-13.3-openmpi-4.1
+
+source ${VENVDIR}/bin/activate             # Activate your virtual environment.
+
+python -u ${PYDIR}/serial_forest.py        # Run your Python script.

3_ensembles/solutions/sh/submit_test_dl_2.sh

+#!/bin/bash
+
+#SBATCH --job-name=dataloader_test         # Job name
+#SBATCH --partition=dev_multiple           # Queue for the resource allocation
+#SBATCH --nodes=2                          # Number of nodes
+#SBATCH --time=30:00                       # Wall-clock time limit  
+#SBATCH --ntasks-per-node=1                # Maximum count of tasks per node
+#SBATCH --cpus-per-task=40                 # CPUs per task
+#SBATCH --mail-type=ALL                    # Notify user by email when certain event types occur.
+
+export OMP_NUM_THREADS=${SLURM_CPUS_PER_TASK}
+export VENVDIR=<path/to/your/venv>         # Export path to your virtual environment.
+export PYDIR=<path/to/your/python/script>  # Export path to directory containing Python script.
+
+# Set up modules.
+module purge                               # Unload all currently loaded modules.
+module load compiler/gnu/13.3              # Load required modules.
+module load mpi/openmpi/4.1
+module load devel/cuda/12.4
+module load lib/hdf5/1.14.4-gnu-13.3-openmpi-4.1
+
+source ${VENVDIR}/bin/activate # Activate your virtual environment.
+
+mpirun python ${PYDIR}/test_dataloaders.py