{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Distributed Computing with Dask\n",
        "\n",
        "<!--\n",
        "Project: Dask Parallel Computing\n",
        "Author: Molla Samser\n",
        "Designer & Tester: Rima Khatun\n",
        "Website: https://rskworld.in\n",
        "Email: help@rskworld.in, support@rskworld.in\n",
        "Phone: +91 93305 39277\n",
        "-->\n",
        "\n",
        "This notebook demonstrates distributed computing with Dask across multiple workers.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from dask.distributed import Client, LocalCluster\n",
        "import dask.array as da\n",
        "import time\n",
        "\n",
        "print(\"Distributed Computing Demo\")\n",
        "print(\"=\" * 50)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Setting Up Local Cluster\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Create a local cluster (for demonstration)\n",
        "# In production, you would connect to a remote cluster\n",
        "cluster = LocalCluster(n_workers=4, threads_per_worker=2)\n",
        "client = Client(cluster)\n",
        "\n",
        "print(\"Cluster Dashboard:\", client.dashboard_link)\n",
        "print(f\"Number of workers: {len(client.scheduler_info()['workers'])}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Distributed Array Operations\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Create a large array distributed across workers\n",
        "large_array = da.random.random((50000, 50000), chunks=(5000, 5000))\n",
        "\n",
        "print(f\"Array shape: {large_array.shape}\")\n",
        "print(f\"Number of chunks: {large_array.numblocks}\")\n",
        "\n",
        "# Perform computation\n",
        "print(\"\\nPerforming distributed computation...\")\n",
        "start_time = time.time()\n",
        "result = (large_array ** 2).sum().compute()\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Result: {result}\")\n",
        "print(f\"Computation time: {end_time - start_time:.2f} seconds\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Monitoring Workers\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Check worker status\n",
        "info = client.scheduler_info()\n",
        "print(\"Worker Information:\")\n",
        "for worker_id, worker_info in info['workers'].items():\n",
        "    print(f\"\\nWorker {worker_id}:\")\n",
        "    print(f\"  Status: {worker_info.get('status', 'unknown')}\")\n",
        "    print(f\"  Memory: {worker_info.get('memory', {}).get('limit', 'N/A')}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Distributed Task Processing\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from dask import delayed\n",
        "\n",
        "@delayed\n",
        "def process_chunk(data_chunk):\n",
        "    \"\"\"Process a chunk of data\"\"\"\n",
        "    time.sleep(0.1)\n",
        "    return data_chunk.sum()\n",
        "\n",
        "# Create multiple tasks\n",
        "chunks = [da.random.random((1000, 1000), chunks=(1000, 1000)) for _ in range(20)]\n",
        "tasks = [process_chunk(chunk.sum()) for chunk in chunks]\n",
        "\n",
        "print(\"Processing 20 tasks across workers...\")\n",
        "start_time = time.time()\n",
        "results = client.compute(tasks)\n",
        "final_results = client.gather(results)\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Results: {len(final_results)} tasks completed\")\n",
        "print(f\"Total time: {end_time - start_time:.2f} seconds\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Cleanup\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Close the client and cluster\n",
        "client.close()\n",
        "cluster.close()\n",
        "print(\"Cluster closed\")\n"
      ]
    }
  ],
  "metadata": {
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}