{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Machine Learning 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 machine learning with Dask, including parallel model training, hyperparameter tuning, and large-scale data preprocessing.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "import dask.array as da\n",
        "import dask.dataframe as dd\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "from sklearn.datasets import make_classification\n",
        "from sklearn.model_selection import train_test_split\n",
        "import time\n",
        "\n",
        "# Note: dask-ml is optional, using sklearn with Dask arrays for demonstration\n",
        "print(\"Machine Learning with Dask\")\n",
        "print(\"=\" * 50)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Creating Large ML Dataset\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Generate large classification dataset\n",
        "print(\"Generating large dataset...\")\n",
        "X, y = make_classification(\n",
        "    n_samples=100000,\n",
        "    n_features=50,\n",
        "    n_informative=30,\n",
        "    n_redundant=10,\n",
        "    n_classes=3,\n",
        "    random_state=42\n",
        ")\n",
        "\n",
        "# Convert to Dask arrays\n",
        "X_dask = da.from_array(X, chunks=(10000, 50))\n",
        "y_dask = da.from_array(y, chunks=10000)\n",
        "\n",
        "print(f\"Dataset shape: {X_dask.shape}\")\n",
        "print(f\"Number of chunks: {X_dask.numblocks}\")\n",
        "print(f\"Classes: {np.unique(y)}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Parallel Data Preprocessing\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from sklearn.preprocessing import StandardScaler\n",
        "from dask import delayed\n",
        "\n",
        "# Standardize features in parallel\n",
        "print(\"Standardizing features...\")\n",
        "start_time = time.time()\n",
        "\n",
        "# Compute mean and std in parallel\n",
        "mean = X_dask.mean(axis=0).compute()\n",
        "std = X_dask.std(axis=0).compute()\n",
        "\n",
        "# Normalize\n",
        "X_normalized = (X_dask - mean) / (std + 1e-8)\n",
        "\n",
        "# Compute a sample to verify\n",
        "sample = X_normalized[:1000, :].compute()\n",
        "\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Normalization completed in {end_time - start_time:.2f} seconds\")\n",
        "print(f\"Sample mean: {sample.mean():.6f}\")\n",
        "print(f\"Sample std: {sample.std():.6f}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Parallel Model Training\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from sklearn.ensemble import RandomForestClassifier\n",
        "from sklearn.linear_model import LogisticRegression\n",
        "from dask import delayed, compute\n",
        "\n",
        "# Split data\n",
        "X_train, X_test, y_train, y_test = train_test_split(\n",
        "    X, y, test_size=0.2, random_state=42\n",
        ")\n",
        "\n",
        "# Convert to Dask arrays\n",
        "X_train_dask = da.from_array(X_train, chunks=(5000, 50))\n",
        "X_test_dask = da.from_array(X_test, chunks=(5000, 50))\n",
        "\n",
        "# Train multiple models in parallel\n",
        "@delayed\n",
        "def train_model(model_class, X, y):\n",
        "    model = model_class()\n",
        "    model.fit(X, y)\n",
        "    return model\n",
        "\n",
        "@delayed\n",
        "def evaluate_model(model, X, y):\n",
        "    return model.score(X, y)\n",
        "\n",
        "# Train different models in parallel\n",
        "print(\"Training models in parallel...\")\n",
        "start_time = time.time()\n",
        "\n",
        "rf_model = train_model(RandomForestClassifier, X_train, y_train)\n",
        "lr_model = train_model(LogisticRegression, X_train, y_train)\n",
        "\n",
        "# Evaluate in parallel\n",
        "rf_score = evaluate_model(rf_model, X_test, y_test)\n",
        "lr_score = evaluate_model(lr_model, X_test, y_test)\n",
        "\n",
        "rf_score_val, lr_score_val = compute(rf_score, lr_score)\n",
        "\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Training completed in {end_time - start_time:.2f} seconds\")\n",
        "print(f\"Random Forest Score: {rf_score_val:.4f}\")\n",
        "print(f\"Logistic Regression Score: {lr_score_val:.4f}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Hyperparameter Tuning\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from sklearn.model_selection import GridSearchCV\n",
        "from dask.distributed import Client, LocalCluster\n",
        "\n",
        "# Set up cluster for parallel hyperparameter tuning\n",
        "cluster = LocalCluster(n_workers=4, threads_per_worker=1)\n",
        "client = Client(cluster)\n",
        "\n",
        "print(f\"Cluster Dashboard: {client.dashboard_link}\")\n",
        "\n",
        "# Define parameter grid\n",
        "param_grid = {\n",
        "    'n_estimators': [50, 100, 200],\n",
        "    'max_depth': [10, 20, None]\n",
        "}\n",
        "\n",
        "# Use smaller dataset for demonstration\n",
        "X_small, y_small = X[:10000], y[:10000]\n",
        "\n",
        "# Grid search with parallel execution\n",
        "print(\"\\nPerforming grid search...\")\n",
        "start_time = time.time()\n",
        "\n",
        "grid_search = GridSearchCV(\n",
        "    RandomForestClassifier(random_state=42),\n",
        "    param_grid,\n",
        "    cv=3,\n",
        "    n_jobs=-1,\n",
        "    verbose=1\n",
        ")\n",
        "\n",
        "grid_search.fit(X_small, y_small)\n",
        "\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"\\nGrid search completed in {end_time - start_time:.2f} seconds\")\n",
        "print(f\"Best parameters: {grid_search.best_params_}\")\n",
        "print(f\"Best score: {grid_search.best_score_:.4f}\")\n",
        "\n",
        "client.close()\n",
        "cluster.close()\n"
      ]
    }
  ],
  "metadata": {
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Delayed Computations - Lazy Evaluation 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 delayed computations and lazy evaluation in Dask.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from dask import delayed, compute\n",
        "import time\n",
        "import numpy as np\n",
        "\n",
        "print(\"Delayed Computations Demo\")\n",
        "print(\"=\" * 50)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Basic Delayed Functions\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Define a function that takes time to execute\n",
        "@delayed\n",
        "def process_data(x):\n",
        "    \"\"\"Simulate data processing\"\"\"\n",
        "    time.sleep(0.1)  # Simulate work\n",
        "    return x * 2\n",
        "\n",
        "# Create delayed computations\n",
        "results = [process_data(i) for i in range(10)]\n",
        "\n",
        "print(\"Created delayed computations (not executed yet)\")\n",
        "print(f\"Type: {type(results[0])}\")\n",
        "\n",
        "# Now compute all results in parallel\n",
        "print(\"\\nComputing results in parallel...\")\n",
        "start_time = time.time()\n",
        "final_results = compute(*results)\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Results: {final_results}\")\n",
        "print(f\"Total time: {end_time - start_time:.2f} seconds\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Complex Workflow\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "@delayed\n",
        "def load_data(filename):\n",
        "    \"\"\"Simulate loading data\"\"\"\n",
        "    time.sleep(0.2)\n",
        "    return np.random.rand(100)\n",
        "\n",
        "@delayed\n",
        "def process_data(data):\n",
        "    \"\"\"Simulate processing data\"\"\"\n",
        "    time.sleep(0.3)\n",
        "    return data.mean()\n",
        "\n",
        "@delayed\n",
        "def combine_results(*results):\n",
        "    \"\"\"Combine multiple results\"\"\"\n",
        "    time.sleep(0.1)\n",
        "    return sum(results)\n",
        "\n",
        "# Create workflow\n",
        "data1 = load_data('file1.csv')\n",
        "data2 = load_data('file2.csv')\n",
        "data3 = load_data('file3.csv')\n",
        "\n",
        "processed1 = process_data(data1)\n",
        "processed2 = process_data(data2)\n",
        "processed3 = process_data(data3)\n",
        "\n",
        "final_result = combine_results(processed1, processed2, processed3)\n",
        "\n",
        "print(\"Workflow created. Executing...\")\n",
        "start_time = time.time()\n",
        "result = final_result.compute()\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Final result: {result}\")\n",
        "print(f\"Total time: {end_time - start_time:.2f} seconds\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Visualizing Task Graph\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Visualize the computation graph\n",
        "try:\n",
        "    final_result.visualize(filename='task_graph.png')\n",
        "    print(\"Task graph saved to task_graph.png\")\n",
        "except Exception as e:\n",
        "    print(f\"Visualization requires graphviz: {e}\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Parallel Processing Example\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "@delayed\n",
        "def expensive_computation(n):\n",
        "    \"\"\"Simulate expensive computation\"\"\"\n",
        "    time.sleep(0.5)\n",
        "    return sum(range(n))\n",
        "\n",
        "# Process multiple tasks in parallel\n",
        "tasks = [expensive_computation(i * 1000) for i in range(1, 11)]\n",
        "\n",
        "print(\"Processing 10 tasks in parallel...\")\n",
        "start_time = time.time()\n",
        "results = compute(*tasks)\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Results: {results}\")\n",
        "print(f\"Total time: {end_time - start_time:.2f} seconds\")\n",
        "print(f\"\\nIf sequential, would take: {0.5 * 10:.2f} seconds\")\n"
      ]
    }
  ],
  "metadata": {
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}