<jupyter_start><jupyter_text>Ray Crash Course - Python Multiprocessing with Ray© 2019-2021, Anyscale. All Rights Reserved![Anyscale Academy](../images/AnyscaleAcademyLogo.png)This lesson explores how to replace two popular multiprocessing libraries with Ray replacements to break the one-machine boundary:* [`multiprocessing.Pool`](https://docs.python.org/3/library/multiprocessing.htmlmodule-multiprocessing.pool) for general management of process pools.* [`joblib`](https://joblib.readthedocs.io/en/latest/), the underpinnings of [scikit-learn](https://scikit-learn.org/stable/), which Ray can scale to a cluster.We also examine how Ray can work with Python's [`asyncio`](https://docs.python.org/3/library/asyncio.html).> **Tip:** For more about Ray, see [ray.io](https://ray.io) or the [Ray documentation](https://docs.ray.io/en/latest/).<jupyter_code>import ray, time, sys, os
import numpy as np
ray.init(ignore_reinit_error=True)<jupyter_output><empty_output><jupyter_text>The Ray Dashboard, if you are running this notebook on a local machine:<jupyter_code>print(f'Dashboard URL: http://{ray.get_dashboard_url()}')<jupyter_output><empty_output><jupyter_text>Drop-in Replacements for Popular Single-node, Multiprocessing LibrariesThe Python community has three popular libraries for breaking out of Python's _global interpreter lock_ to enable better multiprocessing and concurrency. Ray now offers drop-in replacements for two of them, [`multiprocessing.Pool`](https://docs.python.org/3/library/multiprocessing.htmlmodule-multiprocessing.pool) and [`joblib`](https://joblib.readthedocs.io/en/latest/), and integration with the third, Python's [`asyncio`](https://docs.python.org/3/library/asyncio.html).This section explores the `multiprocessing.Pool` and `joblib` replacements.| Library | Library Docs | Ray Docs | Description || :------ | :----------- | :------- | :---------- || `multiprocessing.Pool` | [docs](https://docs.python.org/3/library/multiprocessing.htmlmodule-multiprocessing.pool) | [Ray](https://docs.ray.io/en/latest/multiprocessing.html) | Create a pool of processes for running work. The Ray replacement allows scaling to a cluster. || `joblib` | [docs](https://joblib.readthedocs.io/en/latest/) | [Ray](https://docs.ray.io/en/latest/joblib.html) | Ray supports running distributed [scikit-learn](https://scikit-learn.org/stable/) programs by implementing a Ray backend for `joblib` using Ray Actors instead of local processes. This makes it easy to scale existing applications that use scikit-learn from a single node to a cluster. |  Multiprocessing.PoolIf your application already uses `multiprocessing.Pool`, then scaling beyond a single node just requires replacing your import statements from this:```pythonfrom multiprocessing.pool import Pool```To this:```pythonfrom ray.util.multiprocessing.pool import Pool```A local Ray cluster will be started the first time you create a Pool and your tasks will be distributed across it. See [Run on a Cluster](https://docs.ray.io/en/latest/multiprocessing.htmlrun-on-a-cluster) in the Ray documentation for details on how to use a multi-node Ray cluster instead.Here is an example:<jupyter_code>from ray.util.multiprocessing import Pool

def f(index):
    return index

def run_with_pool(n=100):
    pool = Pool()
    for result in pool.map(f, range(n)):
        print(f'{result}|', end='')

run_with_pool()<jupyter_output><empty_output><jupyter_text>We used a function `run_with_pool()` to wrap a scope around the `pool` construction. That way, it goes out of scope when we're finished and Ray can reclaim the resources.The full `multiprocessing.Pool` API is supported. Please see Python's [multiprocessing documentation](https://docs.python.org/3/library/multiprocessing.htmlmodule-multiprocessing.pool) for details.  JoblibRay supports running distributed [scikit-learn](https://scikit-learn.org/) programs by implementing a Ray backend for [joblib](https://joblib.readthedocs.io/) using Ray Actors instead of local processes. This makes it easy to scale existing applications that use scikit-learn from a single node to a cluster.> **Note:** This API is new and may be revised in the future. Please [report any issues](https://github.com/ray-project/ray/issues) you encounter. To get started, use `from ray.util.joblib import register_ray` and then run `register_ray()`. This will register Ray as a `joblib` backend for `scikit-learn` to use. Then run your original `scikit-learn` code inside `with joblib.parallel_backend('ray')`. This will start a local Ray cluster. See [Run on a Cluster](https://docs.ray.io/en/latest/joblib.htmlrun-on-a-cluster) in the Ray documentation for details on how to use a multi-node Ray cluster instead.Here is an example. First, we set up Ray with `joblib`:<jupyter_code>import joblib
from ray.util.joblib import register_ray
register_ray()<jupyter_output><empty_output><jupyter_text>Now let's use an example taken from the scikit-learn examples, [Restricted Boltzmann Machine features for digit classification](https://scikit-learn.org/stable/auto_examples/neural_networks/plot_rbm_logistic_classification.htmlsphx-glr-auto-examples-neural-networks-plot-rbm-logistic-classification-py).<jupyter_code># Authors: <NAME>, <NAME>, <NAME>
# License: BSD

import numpy as np
from scipy.ndimage import convolve
from sklearn import linear_model, datasets, metrics
from sklearn.model_selection import train_test_split
from sklearn.neural_network import BernoulliRBM
from sklearn.pipeline import Pipeline
from sklearn.base import clone
# #############################################################################
# Setting up

def nudge_dataset(X, Y):
    """
    This produces a dataset 5 times bigger than the original one,
    by moving the 8x8 images in X around by 1px to left, right, down, up
    """
    direction_vectors = [
        [[0, 1, 0],
         [0, 0, 0],
         [0, 0, 0]],

        [[0, 0, 0],
         [1, 0, 0],
         [0, 0, 0]],

        [[0, 0, 0],
         [0, 0, 1],
         [0, 0, 0]],

        [[0, 0, 0],
         [0, 0, 0],
         [0, 1, 0]]]

    def shift(x, w):
        return convolve(x.reshape((8, 8)), mode='constant', weights=w).ravel()

    X = np.concatenate([X] +
                       [np.apply_along_axis(shift, 1, X, vector)
                        for vector in direction_vectors])
    Y = np.concatenate([Y for _ in range(5)], axis=0)
    return X, Y


# Load Data
X, y = datasets.load_digits(return_X_y=True)
X = np.asarray(X, 'float32')
X, Y = nudge_dataset(X, y)
X = (X - np.min(X, 0)) / (np.max(X, 0) + 0.0001)  # 0-1 scaling

X_train, X_test, Y_train, Y_test = train_test_split(
    X, Y, test_size=0.2, random_state=0)

# Models we will use
logistic = linear_model.LogisticRegression(solver='newton-cg', tol=1)
rbm = BernoulliRBM(random_state=0, verbose=True)

rbm_features_classifier = Pipeline(
    steps=[('rbm', rbm), ('logistic', logistic)])
# #############################################################################
# Training

# Hyper-parameters. These were set by cross-validation,
# using a GridSearchCV. Here we are not performing cross-validation to
# save time.
rbm.learning_rate = 0.06
rbm.n_iter = 10
# More components tend to give better prediction performance, but larger
# fitting time
rbm.n_components = 100
logistic.C = 6000<jupyter_output><empty_output><jupyter_text>Now we actually use the Ray backend for `joblib`:<jupyter_code>with joblib.parallel_backend('ray'):
    # Training RBM-Logistic Pipeline
    rbm_features_classifier.fit(X_train, Y_train)

    # Training the Logistic regression classifier directly on the pixel
    raw_pixel_classifier = clone(logistic)
    raw_pixel_classifier.C = 100.
    raw_pixel_classifier.fit(X_train, Y_train)

    # #############################################################################
    # Evaluation

    Y_pred = rbm_features_classifier.predict(X_test)
    print("Logistic regression using RBM features:\n%s\n" % (
        metrics.classification_report(Y_test, Y_pred)))

    Y_pred = raw_pixel_classifier.predict(X_test)
    print("Logistic regression using raw pixel features:\n%s\n" % (
        metrics.classification_report(Y_test, Y_pred)))<jupyter_output><empty_output><jupyter_text>If you see warnings about the `The 'context' argument`, you can safely ignore them.   Using Ray with asyncioPython's [`asyncio`](https://docs.python.org/3/library/asyncio.html) can be used with Ray actors and tasks.> **Note:** The Async API support is experimental and work is ongoing to improve it. Please [report any issues](https://github.com/ray-project/ray/issues) you encounter.  ActorsHere is an actor example, adapted from the [Ray documentation](https://docs.ray.io/en/latest/async_api.html).Note the comment before `run_concurrent`. While normally actor methods are invoked synchronously, in this case there may be concurrent invocations!<jupyter_code>import asyncio

@ray.remote
class AsyncActor:
    # Multiple invocations of this method can be running in
    # the event loop at the same time.
    async def run_concurrent(self, index):
        print(f'started {index}')
        await asyncio.sleep(0.2)   # Concurrent workload here
        print(f'finished {index}')
        return index

actor = AsyncActor.remote()

refs = []
values = []
for i in range(10):
    # regular ray.get
    refs.append(actor.run_concurrent.remote(i))

    # async ray.get
    values.append(await actor.run_concurrent.remote(10+i))
print(ray.get(refs))
print(values)<jupyter_output><empty_output><jupyter_text>Note that using `await` with a method invocation implicitly invokes `ray.get()` on the returned object ref.Under the hood, Ray runs all of the methods inside a single python event loop.> **Note:** Running blocking `ray.get` and `ray.wait` inside async actor methods is not allowed, because `ray.get` will block the execution of the event loop.You can limit the number of concurrent task running at once using the `max_concurrency` flag. By default, 1000 tasks can be running concurrently. In the following cell, we set the `max_concurrency` to `3`, so the subsequent cell will run tasks three at a time. Since there are `12` total, we'll have four groups, each sleeping about `0.2` seconds, so it should take about `0.8` seconds to run.<jupyter_code>actor3 = AsyncActor.options(max_concurrency=3).remote()
%time ray.get([actor3.run_concurrent.remote(i) for i in range(12)])<jupyter_output><empty_output><jupyter_text>See the [03: Ray Internals](../advanced-ray/03-Ray-Internals.ipynb) lesson in the [Advanced Ray](../advanced-ray/00-Advanced-Ray-Overview.ipynb) tutorial for more details on _async actors_.  Async TasksFor Ray tasks, the object refs returned by them can be converted to `async.Future` instances.<jupyter_code>@ray.remote
def some_task():
    return 1

# The normal Ray way:
ref, _ = ray.wait([some_task.remote()])
ray.get(ref)<jupyter_output><empty_output><jupyter_text>The `asyncio` alternative way:<jupyter_code>await some_task.remote()
future = await asyncio.wait([some_task.remote()])
print(future)
# A tuple is returned:
for x in future:
    print(f'  {type(x)} => {x}')<jupyter_output><empty_output><jupyter_text>See the [asyncio docs](https://docs.python.org/3/library/asyncio-task.html) for more details on `asyncio` patterns, including timeouts and `asyncio.gather`.<jupyter_code>ray.shutdown()  # "Undo ray.init()". Terminate all the processes started in this notebook.<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13360677196963086, 'fraction_numerical': 0.010821188585391396, 'mean_word_length': 4.724275724275724, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 53, 'https://': 25, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '132816', 'license': 'Apache-2.0', 'n_tokens_mistral': 3614, 'n_tokens_neox': 3449, 'n_words': 1090, 'path': 'ray-crash-course/04-Ray-Multiprocessing.ipynb', 'repo_name': 'alexy/ray-qiskit-tutorial', 'size': '11484'}