08-big-data-application-pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9988a010-b1d5-44e4-9eea-bf6e4466b4a7",
   "metadata": {},
   "source": [
    "<img src=\"images/panel_logo.png\" width=\"10%\" align=\"right\">\n",
    "\n",
    "# Big Data Application\n",
    "\n",
    "In this notebook we are going to put our visualizations together into an application using Panel Pipelines.\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "73bc78b7-5878-4d16-be17-5c552409059c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import param\n",
    "import panel as pn\n",
    "\n",
    "pn.extension()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d1de4c21-2d5f-4047-a0c7-2808de8f92b8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask_gateway\n",
    "import dask.dataframe as dd\n",
    "\n",
    "import hvplot.dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "129da6c7-c6df-4fa5-9d62-668f8b9d2a8d",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "gateway = dask_gateway.Gateway()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "abfc8e92-fac4-426d-8bd4-7b1e333d5dce",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "class LaunchDaskCluster(param.Parameterized):\n",
    "\n",
    "    workers = pn.widgets.IntRangeSlider(name='Number of Workers', start=1, end=10, value=(1, 5), step=1)\n",
    "    launch_dask = pn.widgets.Button(name='Launch Dask')\n",
    "    dask_status = pn.widgets.StaticText(name='Dask Status', value='Not Connected')\n",
    "    years = pn.widgets.IntRangeSlider(name='Choose Date Range', start=2003, end=2022, value=(2018, 2022), step=1)\n",
    "    cluster = None\n",
    "    cluster_name = param.String()\n",
    "    client = None\n",
    "    ready = param.Boolean(default=False, precedence=-1)\n",
    "    \n",
    "    def __init__(self, **params):\n",
    "        super().__init__(**params)\n",
    "        self.launch_dask.on_click(self.dask_launcher)\n",
    "    \n",
    "    @param.output(('cluster_name', param.String), ('start_year', param.Number), ('end_year', param.Number))\n",
    "    def output(self):\n",
    "        return self.cluster_name, self.years.value[0], self.years.value[1]\n",
    "\n",
    "    def dask_launcher(self, event):\n",
    "        if len(running_clusters := gateway.list_clusters())>0:\n",
    "            self.dask_status.value = \"Found existing dask cluster, connecting and rescaling\"\n",
    "            self.cluster = gateway.connect(running_clusters[0].name)\n",
    "            self.cluster.adapt(self.workers.value[0], self.workers.value[1])\n",
    "        else:\n",
    "            self.dask_status.value = \"Launching new dask cluster\"\n",
    "            self.cluster = gateway.new_cluster(conda_environment=\"global/global-data-of-unusual-size\", profile=\"Medium Worker\")\n",
    "            self.cluster.adapt(self.workers.value[0], self.workers.value[1])\n",
    "        \n",
    "        self.cluster_name = self.cluster.name\n",
    "        self.client = self.cluster.get_client()\n",
    "        self.dask_status.value = \"Waiting for at least 1 worker\"\n",
    "        self.client.wait_for_workers(1)\n",
    "        self.dask_status.value = f\"Cluster Ready - {self.client.dashboard_link}\"\n",
    "        self.ready = True\n",
    "\n",
    "    def panel(self):\n",
    "        return pn.Column(\n",
    "            self.workers,\n",
    "            self.years,\n",
    "            self.launch_dask,\n",
    "            self.dask_status,\n",
    "        )\n",
    "\n",
    "stage1 = LaunchDaskCluster()\n",
    "# stage1.panel()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f3eae75-0829-4bca-bebc-6d49c8ba0050",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "class Dashboard(param.Parameterized):\n",
    "    method = pn.widgets.RadioButtonGroup(name='Method', options=['min', 'mean', 'max'])\n",
    "    field = pn.widgets.RadioButtonGroup(name='Field', options=['DEP_DELAY', 'ARR_DELAY'])\n",
    "    groupby = pn.widgets.RadioButtonGroup(name='GroupBy', options=['YEAR', 'MONTH', 'DAY_OF_MONTH', 'OP_CARRIER'], value='MONTH')\n",
    "    cluster_name = param.String()\n",
    "    start_year = param.Number()\n",
    "    end_year = param.Number()\n",
    "    \n",
    "    def __init__(self, **params):\n",
    "        super().__init__(**params)\n",
    "        url = \"gcs://quansight-datasets/airline-ontime-performance/sorted/full_dataset.parquet\"\n",
    "        columns = [\n",
    "            'YEAR', 'MONTH', 'DAY_OF_MONTH', 'DAY_OF_WEEK', 'FL_DATE', 'OP_CARRIER', \n",
    "            'TAIL_NUM', 'OP_CARRIER_FL_NUM', 'ORIGIN', 'DEST', 'CRS_DEP_TIME', \n",
    "            'DEP_TIME', 'DEP_DELAY', 'ARR_TIME', 'ARR_DELAY', 'CANCELLED', \n",
    "            'CANCELLATION_CODE', 'DIVERTED', 'AIR_TIME', 'FLIGHTS', 'DISTANCE',\n",
    "            'CARRIER_DELAY', 'WEATHER_DELAY', 'NAS_DELAY', 'SECURITY_DELAY', \n",
    "            'LATE_AIRCRAFT_DELAY', 'DIV_ARR_DELAY'\n",
    "        ]\n",
    "        self.flights = dd.read_parquet(\n",
    "                f\"gcs://quansight-datasets/airline-ontime-performance/sorted/parquet_by_year\", \n",
    "                filters=[\n",
    "                    ('YEAR', '>=', self.start_year),\n",
    "                    ('YEAR', '<=', self.end_year)\n",
    "                ],\n",
    "                columns=columns,\n",
    "        ).interactive()\n",
    "        \n",
    "    @param.depends('cluster_name')\n",
    "    def dask_dashboard(self):\n",
    "        self.cluster = gateway.connect(self.cluster_name)\n",
    "        self.client = self.cluster.get_client()\n",
    "        return pn.pane.HTML(f\"\"\"\n",
    "        <iframe width=\"800\" height=\"800\" src=\"{self.client.dashboard_link}\"\n",
    "        frameborder=\"0\" scrolling=\"no\" marginheight=\"0\" marginwidth=\"0\"></iframe>\n",
    "        \"\"\")\n",
    "    \n",
    "    @param.depends('groupby', 'method', 'field')        \n",
    "    def plot_data(self):\n",
    "        return (\n",
    "            self.flights\n",
    "                .groupby(self.groupby)[self.field]\n",
    "                .agg(how=self.method)\n",
    "                .hvplot()\n",
    "        ).panel()\n",
    "    \n",
    "    @param.output('cluster_name', param.String)\n",
    "    def output(self):\n",
    "        return self.cluster_name\n",
    "        \n",
    "        \n",
    "    def panel(self):\n",
    "        return pn.Row(\n",
    "            pn.Column(\n",
    "                self.groupby,\n",
    "                self.field,\n",
    "                self.method,\n",
    "                pn.layout.Divider(),\n",
    "                self.plot_data,\n",
    "            ),\n",
    "            self.dask_dashboard,\n",
    "        )\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02c3c59d-c0a1-4027-85b9-831969de8b46",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "stage2 = Dashboard(\n",
    "            cluster_name=stage1.output()[0],\n",
    "            start_year=stage1.output()[1],\n",
    "            end_year=stage1.output()[2]\n",
    ")\n",
    "# stage2.panel()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a33f2f8-21a6-49fa-a465-f6645f77d75f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "class StopDaskCluster(param.Parameterized):\n",
    "    cluster_name = param.String()\n",
    "    dask_status = pn.widgets.StaticText(name='Dask Status', value='')\n",
    "    stop_dask = pn.widgets.Button(name='Stop Dask Cluster')\n",
    "    \n",
    "    def __init__(self, **params):\n",
    "        super().__init__(**params)\n",
    "        self.stop_dask.on_click(self.shutdown_cluster)\n",
    "        \n",
    "    @param.depends('cluster_name')\n",
    "    def dask_dashboard(self):\n",
    "        self.dask_status.value = \"Checking Status\"\n",
    "        self.cluster = gateway.connect(self.cluster_name)\n",
    "        self.dask_status.value = self.cluster.status\n",
    "    \n",
    "    def shutdown_cluster(self, event):\n",
    "        self.dask_status.value = \"Shutting Cluster Down\"\n",
    "        self.cluster.shutdown()\n",
    "        self.dask_status.value = self.cluster.status\n",
    "        \n",
    "    def panel(self):\n",
    "        return pn.Column(\n",
    "            self.stop_dask,\n",
    "            self.dask_status,\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eb2aa07e-74cc-48f2-81b0-4e69698c9469",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "stage3 = StopDaskCluster(cluster_name=stage2.output())\n",
    "# stage3.panel()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5b32269-5ed8-4a51-98be-c948f5395275",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline = pn.pipeline.Pipeline()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "34292833-48d3-4d96-abb4-4abb66602b80",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline.add_stage('Launch Dask', stage1)\n",
    "pipeline.add_stage('Visualization', stage2)\n",
    "pipeline.add_stage('Stop Cluster', stage3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f9bae548-2a2a-4110-87b2-19f8af410e66",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline.servable()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9738004d-2170-4ba8-860d-13894b3f861a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "gateway.list_clusters()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c57eefab-6fdd-4945-9806-d8fc5c688064",
   "metadata": {},
   "outputs": [],
   "source": [
    "# panel serve 07b-big-data-dashboard-pipeline.ipynb --allow-websocket-origin=nebari.quansight.dev\n",
    "# https://nebari.quansight.dev/user/dharhas@quansight.com/proxy/5006/07b-big-data-dashboard-pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e04da66-c0ba-415b-a453-257242b215fd",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## Next →\n",
    "\n",
    "We'll discuss some ways to [collaborate on data science projects](./09-collaborative-data-science.ipynb)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "36ee3771-1930-4003-9f40-28e28dba7365",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "global-global-data-of-unusual-size",
   "language": "python",
   "name": "conda-env-global-global-data-of-unusual-size-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}