\n",
"Client\n", "
| \n",
"\n",
"Cluster\n", "
| \n",
"
![\"PI](\"https://upload.wikimedia.org/wikipedia/commons/8/84/Pi_30K.gif\")
](https://en.wikipedia.org/wiki/Pi#Monte_Carlo_methods)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"\n", "Let's define a function to compute $\\pi$ and another function to print out some info during the compute." ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "import dask.array as da\n", "import numpy as np\n", "\n", "def calc_pi_mc(size_in_bytes, chunksize_in_bytes=200e6): \n", " \"\"\"Calculate PI using a Monte Carlo estimate.\"\"\"\n", " \n", " size = int(size_in_bytes / 8) # size= no. of random numbers to generate (x & y vals), divide 8 bcz numpy float64's generated by random.uniform are 8 bytes\n", " chunksize = int(chunksize_in_bytes / 8)\n", " \n", " xy = da.random.uniform(0, 1, # this generates a set of x and y value pairs on the interval [0,1) of type float64\n", " size=(size / 2, 2), # divide 2 because we are generating an equal number of x and y values (to get our points)\n", " chunks=(chunksize / 2, 2)) \n", " \n", " in_circle = ((xy ** 2).sum(axis=-1) < 1) # a boolean array, True for points that fall inside the unit circle (x**2 + y**2 < 1)\n", " pi = 4 * in_circle.mean() # mean= sum the number of True elements, divide by the total number of elements in the array\n", "\n", " return pi\n", "\n", "def print_pi_stats(size, pi, time_delta, num_workers): \n", " \"\"\"Print pi, calculate offset from true value, and print some stats.\"\"\"\n", " print(f\"{size / 1e9} GB\\n\"\n", " f\"\\tMC pi: {pi : 13.11f}\"\n", " f\"\\tErr: {abs(pi - np.pi) : 10.3e}\\n\"\n", " f\"\\tWorkers: {num_workers}\"\n", " f\"\\t\\tTime: {time_delta : 7.3f}s\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## The Actual Calculations\n", "\n", "We loop over different volumes (1GB, 10GB, and 100GB) of double-precision random numbers (float64, 8 bytes each) and estimate $\\pi$ as described above. Note, we call the function with the .compute() method to start the computations. To see the dask workers computing, execute the code block below and then quickly click over to your dask workers tab." ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "1.0 GB\n", "\tMC pi: 3.14163468800\tErr: 4.203e-05\n", "\tWorkers: 3\t\tTime: 1.156s\n", "10.0 GB\n", "\tMC pi: 3.14153445760\tErr: 5.820e-05\n", "\tWorkers: 3\t\tTime: 1.139s\n", "100.0 GB\n", "\tMC pi: 3.14159745088\tErr: 4.797e-06\n", "\tWorkers: 3\t\tTime: 8.593s\n" ] } ], "source": [ "for size in (1e9 * n for n in (1, 10, 100)):\n", " \n", " start = time()\n", " pi = calc_pi_mc(size).compute()\n", " elaps = time() - start\n", "\n", " print_pi_stats(size, pi, time_delta=elaps,\n", " num_workers=len(cluster.scheduler.workers))\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Scale the Cluster to Twice its Size and Re-run the Same Calculations\n", "\n", "We increase the number of workers times 2 and the re-run the experiments. You could also double the size of the cluster by doubling the number of jobs, cores, or memory." ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Scaling from 3 to 6 workers.\n" ] }, { "data": { "text/html": [ "
\n",
"Client\n", "
| \n",
"\n",
"Cluster\n", "
| \n",
"
\n",
"Client\n", "
| \n",
"\n",
"Cluster\n", "
| \n",
"
\n",
"Client\n", "
| \n",
"\n",
"Cluster\n", "
| \n",
"