<jupyter_start><jupyter_text>Ray Crash Course - Tasks© 2019-2020, Anyscale. All Rights Reserved![Anyscale Academy](../images/AnyscaleAcademy_Logo_clearbanner_141x100.png)Let's quickly explore the Ray API using some examples that demonstrate how Ray enables horizontal scalability.> **Tip:** For more about Ray, see [ray.io](https://ray.io) or the [Ray documentation](https://docs.ray.io/en/latest/). Evaluate the next _three_ notebook cells (shift+return, if you're new to notebooks), down to and including `run_simulations(...)`, and watch what happens:<jupyter_code>from task_lesson_util import make_dmaps, run_simulations, stop_simulations
from pi_calc import str_large_n
Ns = [1000, 10000, 100000]

dmaps = make_dmaps(Ns)

dmaps[0] + dmaps[1] + dmaps[2]
run_simulations(dmaps)
# TIP: If you want to stop them, uncomment and run the next line:
# stop_simulations(dmaps)<jupyter_output><empty_output><jupyter_text>(If you can't see it, click [here](../images/Pi-estimates.png) for a screen grab.)What we just did was estimate $\pi$ (~3.14159) using a [_Monte Carlo_ technique](https://en.wikipedia.org/wiki/Monte_Carlo_method), where we randomly sampled a _uniform distribution_, one with equal probably of picking any point in a square.It works like this. Imagine each blue square is a piece of paper 2 meters by 2 meters you put on a wall. The circle inside each one has radius 1 meter.Now suppose you throw $N$ darts at each paper. We're seeing $N = {\sim}1000, {\sim}10000, {\sim}100000$ examples. (This will be hard on your wall, so don't try this at home...)Some darts will land inside the circle, call them $n$, and the rest will land outside, $N-n$. The area of a circle is ${\pi}r^{2}$ and the area of a square is $(2r)^{2} = 4r^{2}$. The ratio of $n/N$ _approximately_ equals the ratio of the circle area over the square area, ${\pi}r^{2}/4r^{2} = {\pi}/4$. (Does it make sense that this ratio is independent of the actual radius value?).In other words,$\pi/4 \approx n/N$$\pi \approx 4n/N$So, to approximate $\pi$, we can count the number of darts thrown and the number that land inside the circle. You probably noticed three things while the simulations were running or after they finished:1. The accuracy improved for larger $N$... well usually. Sometimes a lower $N$ simulation gets "lucky" and does as well as a higher $N$. In a real experiment, we would do many runs and then compute the average and standard deviation. (We'll do that below.)2. Because each $N$ is 10 times the $N$ to the left, it took roughly 10 times as long for the second to finish compared to the first, etc.3. The updates in the second and third simulations appeared to go faster as the neighbors to the left finished.What this means is that if we really want a good estimate of $\pi$, we have to do runs with large $N$, but then we wait longer. Ideally, to get fast _and_ accurate results, we would do as much work as possible in parallel, leveraging all the CPU cores available on our machine ... or our cluster. Let's use [_Ray_](https://ray.io) to achieve this.> **Note:** There is a lot of Python code used in this notebook, both for calculating Pi and for the graphs. We won't show most of it, but all the code can be found in this directory, `pi_calc.py` (calculating $\pi$) and `task_lesson_util.py` (graphics).   Parallelism with RayWe did the previous calculation without fully exploiting all available cores. In a cluster, the rest of the cores _on the rest of the machines_ would be idle, too.We can use Ray to parallelize a lot of this work. Let's see how.<jupyter_code>import numpy as np
import math, statistics, time
import ray<jupyter_output><empty_output><jupyter_text>Some constants we'll use (and explain) below.<jupyter_code>num_workers = 8
trials = 20<jupyter_output><empty_output><jupyter_text>Let's start Ray. Although we don't need it immediately, we'll use the Ray Dashboard to watch performance metrics like CPU utilization.These tutorials run a Ray cluster in your environment. The following bash shell script will determine if Ray is already running using the CLI command `ray stat`. If not, it will tell you how to start Ray using the CLI command `ray start --head`, which starts the "head" node of a Ray cluster.If you are working on a laptop, this will be the only node in your "cluster". If you are working through this material using the Anyscale platform, you'll have a cluster of a 1 or more nodes and Rays should already be running, so you'll just see a message like _Ray already running or successfully started_. You'll learn more about the `ray` CLI in lesson 4, [Running Ray Clusters](04-Running-Ray-Clusters.ipynb). See also [this documentation](https://docs.ray.io/en/latest/starting-ray.htmlusing-ray-on-a-cluster).<jupyter_code>!../tools/start-ray.sh  --check --verbose<jupyter_output><empty_output><jupyter_text>If you are working on your laptop, make sure to run `ray stop` in a terminal window when you have finished for the day. If you are running on the Anyscale platform, this step is not necessary. Now we'll initialize Ray in this notebook process and connect to the cluster. The two keyword arguments passed do the following:* `address='auto'`: Automatically detect the cluster head node address by connecting to the local Ray processes on this machine.* `ignore_reinit_error=True`: Tells Ray not to complain if we rerun the cell. Note that Ray won't reestablish the connection; it will just ignore the request to initialize.<jupyter_code>ray.init(address='auto', ignore_reinit_error=True)<jupyter_output><empty_output><jupyter_text>> **Tip:** Having trouble starting Ray? See the [Troubleshooting](../reference/Troubleshooting-Tips-Tricks.ipynb) tips. The next cell prints the URL for the Ray Dashboard. **This is only correct if you are running this tutorial on a laptop.** Click the link to open the dashboard.If you are running on the Anyscale platform, use the URL provided by your instructor to open the Dashboard.<jupyter_code>print(f'Dashboard URL: http://{ray.get_webui_url()}')<jupyter_output><empty_output><jupyter_text>Let's define a function to do the Pi calculation that simplifies the code we used above for graphing purposes. We won't do the "dart graphs" from now on, because they add a lot of overhead that would obscure the performanceThis function estimates $\pi$ for the number of samples requested. It uses [NumPy](https://docs.scipy.org/doc/numpy/index.html). If you're not familiar with it, the implementation details aren't essential to understand, but the comments try to explain them.<jupyter_code>def estimate_pi(num_samples):
    xs = np.random.uniform(low=-1.0, high=1.0, size=num_samples)   # Generate num_samples random samples for the x coordinate.
    ys = np.random.uniform(low=-1.0, high=1.0, size=num_samples)   # Generate num_samples random samples for the y coordinate.
    xys = np.stack((xs, ys), axis=-1)                              # Like Python's "zip(a,b)"; creates np.array([(x1,y1), (x2,y2), ...]).
    inside = xs*xs + ys*ys <= 1.0                                  # Creates a predicate over all the array elements.
    xys_inside = xys[inside]                                       # Selects only those "zipped" array elements inside the circle.
    in_circle = xys_inside.shape[0]                                # Return the number of elements inside the circle.
    approx_pi = 4.0*in_circle/num_samples                          # The Pi estimate.
    return approx_pi<jupyter_output><empty_output><jupyter_text>Let's try it:<jupyter_code>Ns = [10000, 50000, 100000, 500000, 1000000] #, 5000000, 10000000]  # Larger values take a long time on small VMs and machines!
maxN = Ns[-1]
maxN
fmt = '{:10.5f} seconds: pi ~ {:7.6f}, stddev = {:5.4f}, error = {:5.4f}%'
def try_it(n, trials):
    print('trials = {:3d}, N = {:s}: '.format(trials, str_large_n(n, padding=12)), end='')   # str_large_n imported above.
    start = time.time()
    pis = [estimate_pi(n) for _ in range(trials)]
    approx_pi = statistics.mean(pis)
    stdev = statistics.stdev(pis)
    duration = time.time() - start
    error = (100.0*abs(approx_pi-np.pi)/np.pi)
    print(fmt.format(duration, approx_pi, stdev, error))   # str_large_n imported above.
    return trials, n, duration, approx_pi, stdev, error<jupyter_output><empty_output><jupyter_text>The next cell will take many seconds to complete (depending on your setup). As soon as it starts, switch to the Ray Dashboard you opened above. Notice the total CPU and memory utilizations at the top while the cell runs. > **Tip:** If all the following trials finish in under a few seconds for the largest `n` value in `Ns` and the largest number of trials, consider changing `Ns` above to add larger values.<jupyter_code>data_ns = [try_it(n, trials) for n in Ns]
data_trials = [try_it(maxN, trials) for trials in range(5,20,2)]<jupyter_output><empty_output><jupyter_text>(We'll graph the data below.) The CPU utilization never gets close to 100%. On a four-core machine, for example, the number will be about 25%. (The Ray process meters will stay at or near zero until later in this notebook.)So, this runs on one core, while the other cores are idle. Now we'll try with Ray.  From Python Functions to Ray TasksYou create a Ray _task_ by decorating a normal Python function with `@ray.remote`. These tasks will be scheduled across your Ray cluster (or your laptop CPU cores).> **Tip:** The [Ray Package Reference](https://ray.readthedocs.io/en/latest/package-ref.html) in the [Ray Docs](https://ray.readthedocs.io/en/latest/) is useful for exploring the API features we'll learn.Here is a Ray task for `estimate_pi`. All we need is a wrapper around the original function.<jupyter_code>@ray.remote
def ray_estimate_pi(num_samples):
    return estimate_pi(num_samples)<jupyter_output><empty_output><jupyter_text>Let's try it. To invoke a task, you use `function.remote(args)`:<jupyter_code>ray_estimate_pi.remote(100)<jupyter_output><empty_output><jupyter_text>What is this `ObjectID`? A Ray task is an asynchronous computation. The `ObjectID` returned is a _future_ that we use to retrieve the resulting value from the task when it completes. We use `ray.get(id)` to get it:<jupyter_code>id = ray_estimate_pi.remote(100)
print(ray.get(id))<jupyter_output><empty_output><jupyter_text>We can also work with lists of ids:<jupyter_code>ids = [ray_estimate_pi.remote(n) for n in [100, 1000, 10000]]
print(ray.get(ids))<jupyter_output><empty_output><jupyter_text>Okay, let's try our test run again with our Ray task. We'll need a new "try it" function, because of the different task invocation logic. This function doesn't need to be a Ray task, however, so no `@ray.remote` decorator is required.<jupyter_code>def ray_try_it(n, trials):
    print('trials = {:3d}, N = {:s}: '.format(trials, str_large_n(n, padding=12)), end='')   # str_large_n imported above.
    start = time.time()
    ids = [ray_estimate_pi.remote(n) for _ in range(trials)]
    pis = ray.get(ids)
    approx_pi = statistics.mean(pis)
    stdev = statistics.stdev(pis)
    duration = time.time() - start
    error = (100.0*abs(approx_pi-np.pi)/np.pi)
    print(fmt.format(duration, approx_pi, stdev, error))   # str_large_n imported above.
    return trials, n, duration, approx_pi, stdev, error
ray_data_ns = [ray_try_it(n, trials) for n in Ns]
ray_data_trials = [ray_try_it(maxN, trials) for trials in range(5,20,2)]<jupyter_output><empty_output><jupyter_text>The durations should be shorter than the non-Ray numbers. Let's graph our results and see. It will be easier if we first convert the `*data_*` lists to NumPy arrays, so they are easier to slice.<jupyter_code>np_data_ns         = np.array(data_ns)
np_data_trials     = np.array(data_trials)
np_ray_data_ns     = np.array(ray_data_ns)
np_ray_data_trials = np.array(ray_data_trials)
from bokeh_util import two_lines_plot, means_stddevs_plot  # Some plotting utilities in `./bokeh_util.py`.
from bokeh.plotting import show, figure
from bokeh.layouts import gridplot<jupyter_output><empty_output><jupyter_text>First a linear plot of the results<jupyter_code>two_lines = two_lines_plot(
    "N vs. Execution Times (Smaller Is Better)", 'N', 'Time', 'No Ray', 'Ray', 
    np_data_ns[:,1], np_data_ns[:,2], np_ray_data_ns[:,1], np_ray_data_ns[:,2],
    x_axis_type='linear', y_axis_type='linear')
show(two_lines, plot_width=800, plot_height=400)<jupyter_output><empty_output><jupyter_text>(If you can't see it, click [here](../images/Pi-Ns-vs-times-linear.png). Note that this image is from a run that included the larger values for `N` that are commented out in the definition of `Ns` above.)For relatively small `N` values, the performance overhead of overhead of Ray is a larger percentage of the calculation, so the overall performance benefit is less. However, as `N` increases, the advantage of Ray increases. Both plots are roughly-linear, because we are CPU bound, but Ray's execution time/N is lower. On a full cluster, the times could be dramatically better for larger `N`. A log-log plot shows the lower-N behavior more clearly:<jupyter_code>two_lines = two_lines_plot(
    "N vs. Execution Times (Smaller Is Better)", 'N', 'Time', 'No Ray', 'Ray', 
    np_data_ns[:,1], np_data_ns[:,2], np_ray_data_ns[:,1], np_ray_data_ns[:,2])
show(two_lines, plot_width=800, plot_height=400)<jupyter_output><empty_output><jupyter_text>(If you can't see it, click [here](../images/Pi-Ns-vs-times.png).)What about execution times as a function of the number of trials, for a fixed `N`?<jupyter_code>two_lines = two_lines_plot(
    "Trials (N=10,000,000) vs. Execution Times (Smaller Is Better)", 'Trials', 'Time', 'No Ray', 'Ray', 
    np_data_trials[:,0], np_data_trials[:,2], np_ray_data_trials[:,0], np_ray_data_trials[:,2], 
    x_axis_type='linear', y_axis_type='linear')
show(two_lines, plot_width=800, plot_height=400)<jupyter_output><empty_output><jupyter_text>(If you can't see it, click [here](../images/Pi-trials-vs-times.png).) Let's plot the approximate mean values and the standard deviations over the `num_workers` trials for each `N`.<jupyter_code>pi_without_ray_plot = means_stddevs_plot(
  np_data_ns[:,1], np_data_ns[:,3], np_data_ns[:,4], title = 'π Results without Ray')
# Use a grid to make it layout better.
pi_without_ray_grid = gridplot([[pi_without_ray_plot]], plot_width=1000, plot_height=400)
show(pi_without_ray_grid)<jupyter_output><empty_output><jupyter_text>(If you can't see it, click [here](../images/Pi-Results-without-Ray.png).)You may have to use the "crossed-arrows" controls scroll horizontally (click and drag) to see all of the graph.As you might expect, for low `N` values, the error bars are large and the mean estimate is poor, but for higher `N`, the errors grow smaller and results converge to the correct value.With Ray, the plot will look similar, because we did the same calculation, just faster:<jupyter_code>pi_with_ray_plot = means_stddevs_plot(
  np_ray_data_ns[:,1], np_ray_data_ns[:,3], np_ray_data_ns[:,4], title = 'π Results with Ray')
# Use a grid to make it layout better.
pi_with_ray_grid = gridplot([[pi_with_ray_plot]], plot_width=1000, plot_height=400)
show(pi_with_ray_grid)<jupyter_output><empty_output><jupyter_text>(If you can't see it, click [here](../images/Pi-Results-with-Ray.png).)  ray.get() vs. ray.wait()Calling `ray.get(ids)` blocks until all the tasks have completed that correspond to the input `ids`. That has been fine for this tutorial so far, but what if you're waiting for a number of tasks, where some will finish more quickly than others? What if you would like to process the completed results as they become available, even while other tasks are still running? That's where `ray.wait()` is recommended. Here we'll provide a brief example. For more details, see the [Ray Core - A Deeper Dive](../ray-core/00-Overview.ipynb) tutorial.<jupyter_code>@ray.remote
def ray_estimate_pi2(n, trial):
    time.sleep(trial)
    return n, trial, estimate_pi(n)
def ray_try_it2(ns, trials):
    start = time.time()
    ids = [ray_estimate_pi2.remote(n, trial) for trial in trials for n in ns]
    still_running = list(ids)
    while len(still_running) > 0:
        finished, still_running = ray.wait(still_running)
        ns_trials_pis = ray.get(finished)   # won't block
        print(f'{ns_trials_pis}, elapsed time = {time.time() - start} secs')<jupyter_output><empty_output><jupyter_text>Observe what happens next:<jupyter_code>ray_try_it2([100000,1000000,1000000], [2,4,6])<jupyter_output><empty_output><jupyter_text>ExercisesTry one or more of the following exercises to practice improving scalable performance using Ray. In particular, think about the granularity of tasks where Ray is most beneficial.See the [solutions notebook](solutions/Ray-Crash-Course-Solutions.ipynb) for a discussion of solutions to these exercises.  Exercise 1As currently written, the memory footprint of `estimate_pi` scales linearly with `N`, because it allocates two NumPy arrays of size `N`. This limits the size of `N` we can evaluate (as I confirmed by locking up my laptop...). However, this isn't actually necessary. We could do the same calculation in "blocks, for example `m` blocks of size `N/m` and then combine the results. Furthermore, there's no dependencies between the calculations with those blocks, giving us further potential speed-up by parellelizing them with Ray.Adapt `ray_estimate_pi` to use this technique. Pick some `N` value above which the calculation is done in blocks. Compare the performance of the old vs. new implementation. As you do this exercise, you might ponder the fact that we often averaged multiple trials for a given `N` and then ask yourself, what's the difference between averaging `10` trials for `N = 1000` vs. `1` trial for `N = 10000`, for example?  Exercise 2What `N` value is needed to get a reliable estimate to five decimal places, `3.1415` (for some definition of "reliable")? If you have a powerful machine or a cluster, you could try a higher accuracy. You'll need to use the solution to Exercise 1 or you can make a guess based on the results we've already seen in this notebook.  Exercise 3For small computation problems, Ray adds enough overhead that its benefits are outweighed. You can see from the performance graphs above that smaller `N` or smaller trial values will likely cause the curves to cross. Try small values of `N` and small trial numbers. When do the lines cross? Try timing individual runs for small `N` around the crossing point. What can you infer from this "tipping point" about appropriate sizing of tasks, at least for your test environment?<jupyter_code>ray.shutdown()  # "Undo ray.init()". Terminate all the processes started in this notebook.<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09734371666311073, 'fraction_numerical': 0.016375080008534243, 'mean_word_length': 4.784942918852206, 'pattern_counts': {'":': 0, '<': 94, '<?xml version=': 0, '>': 99, 'https://': 8, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '224719', 'license': 'Apache-2.0', 'n_tokens_mistral': 5897, 'n_tokens_neox': 5593, 'n_words': 2320, 'path': 'ray-crash-course/01-Ray-Tasks.ipynb', 'repo_name': 'yunjieyao/academy', 'size': '18748'}