Examples

Parallely count bigrams in a text

"""
Counts the most common two-letters sequences (bigrams) in the content of an URL.

Usage:

    $ git clone https://github.com/Citi/parfun && cd parfun
    $ python -m examples.count_bigrams.main
"""

import collections
import psutil
import ssl

from typing import Counter, Iterable, List
from urllib.request import urlopen

import parfun as pf


def sum_counters(counters: Iterable[Counter[str]]) -> Counter[str]:
    return sum(counters, start=collections.Counter())


@pf.parallel(
    split=pf.per_argument(
        lines=pf.py_list.by_chunk
    ),
    combine_with=sum_counters,
)
def count_bigrams(lines: List[str]) -> Counter:
    counter: Counter[str] = collections.Counter()

    for line in lines:
        for word in line.split():
            for first, second in zip(word, word[1:]):
                bigram = f"{first}{second}"
                counter[bigram] += 1

    return counter


if __name__ == "__main__":
    N_WORKERS = psutil.cpu_count(logical=False)
    URL = "https://www.gutenberg.org/ebooks/100.txt.utf-8"
    TOP_K = 10

    with urlopen(URL, context=ssl._create_unverified_context()) as response:
        content = response.read().decode("utf-8").splitlines()

    with pf.set_parallel_backend_context("local_multiprocessing", max_workers=N_WORKERS):
        counts = count_bigrams(content)

    print(f"Top {TOP_K} words:")
    for word, count in counts.most_common(TOP_K):
        print(f"\t{word:<10}:\t{count}")

Parallely train a random tree regressor

"""
Trains a random tree regressor on the California housing dataset from scikit-learn.

Measures the training time when splitting the learning dataset process using Parfun.

Usage:

    $ git clone https://github.com/Citi/parfun && cd parfun
    $ python -m examples.california_housing.main
"""

import psutil
import timeit

from typing import List

import numpy as np
import pandas as pd

from sklearn.datasets import fetch_california_housing
from sklearn.base import RegressorMixin
from sklearn.tree import DecisionTreeRegressor

import parfun as pf


class MeanRegressor(RegressorMixin):
    def __init__(self, regressors: List[RegressorMixin]) -> None:
        super().__init__()
        self._regressors = regressors

    def predict(self, X):
        return np.mean([regressor.predict(X) for regressor in self._regressors])


@pf.parallel(
    split=pf.per_argument(dataframe=pf.dataframe.by_row),
    combine_with=lambda regressors: MeanRegressor(list(regressors))
)
def train_regressor(dataframe: pd.DataFrame, feature_names: List[str], target_name: str) -> RegressorMixin:

    regressor = DecisionTreeRegressor()
    regressor.fit(dataframe[feature_names], dataframe[[target_name]])

    return regressor


if __name__ == "__main__":
    N_WORKERS = psutil.cpu_count(logical=False)

    dataset = fetch_california_housing(download_if_missing=True)

    feature_names = dataset["feature_names"]
    target_name = dataset["target_names"][0]

    dataframe = pd.DataFrame(dataset["data"], columns=feature_names)
    dataframe[target_name] = dataset["target"]

    N_MEASURES = 5

    with pf.set_parallel_backend_context("local_single_process"):
        regressor = train_regressor(dataframe, feature_names, target_name)

        duration = (
            timeit.timeit(lambda: train_regressor(dataframe, feature_names, target_name), number=N_MEASURES)
            / N_MEASURES
        )

        print("Sequential training duration:", duration)

    with pf.set_parallel_backend_context("local_multiprocessing", max_workers=N_WORKERS):
        regressor = train_regressor(dataframe, feature_names, target_name)

        duration = (
            timeit.timeit(lambda: train_regressor(dataframe, feature_names, target_name), number=N_MEASURES)
            / N_MEASURES
        )

        print("Parallel training duration:", duration)