dask_dataframe.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "5f350668-556d-495f-8f90-3bce64dc6595",
   "metadata": {},
   "source": [
    "# Introduction to Dask DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a6fa4249-ea9b-434d-8a58-c20f9e104f51",
   "metadata": {},
   "source": [
    "Dask DataFrames coordinate many pandas DataFrames/Series arranged along the index. A Dask DataFrame is partitioned row-wise, grouping rows by index value for efficiency."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7b0aadce-d8b6-40c2-ab77-5fdd2ee488c4",
   "metadata": {},
   "source": [
    "![Dask DataFrame](https://docs.dask.org/en/stable/_images/dask-dataframe.svg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44e112b4-0172-4970-a9d8-14eb5dbd15ef",
   "metadata": {},
   "source": [
    "Many of existing methods from pandas API are available in Dask DataFrame. Checkout [this section](https://docs.dask.org/en/stable/dataframe.html#scope) of the documentation to learn more about these. In general, computations that are parallelizable are implemented in Dask DataFrame. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4362f85f-eb93-44ac-a9fc-b1b5648fd57e",
   "metadata": {},
   "source": [
    "In this lecture, you will learn to use Dask DataFrame to analyze large tabular climate data.  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "867ad78f-66b5-4ee7-91e0-15ee71de1fc1",
   "metadata": {},
   "source": [
    "## Analyzing Multiple Large CSV files using Dask Data Frame"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c52fa4bd-3ab4-4114-96f4-4ed5d273a3cb",
   "metadata": {},
   "source": [
    "For this tutorial, we will use the NOAA Global Historical Climatology Network Daily (GHCN-D) data available on AWS S3. \n",
    "You can read more about the data on Registry of Open Data on AWS [here](https://registry.opendata.aws/noaa-ghcn/).\n",
    "\n",
    "More information about the dataset, including the metadata descriptions, is available on [NOAA's website](https://www.ncei.noaa.gov/products/land-based-station/global-historical-climatology-network-daily). \n",
    "\n",
    "GHCN-D contains **daily observations** over global land areas. It contains station-based measurements from land-based stations worldwide, about two thirds of which are for precipitation measurement only. Some data are more than *175 years* old."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6bc7caba-5228-41af-92d2-c98eeea7e0db",
   "metadata": {
    "tags": []
   },
   "source": [
    "This dataset is very large and to analyze it within Python you need to use Dask Dataframe."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d6650833-b6bf-471a-8954-c488c27b8211",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Download Data from AWS S3 bucket"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d7b677e4-9b22-406b-97d1-1424625bb275",
   "metadata": {},
   "source": [
    "You can download the dataset from AWS S3 bucket using the following commands. This dataset does not require an AWS account (hence the `UNSIGNED` value should be passed to the S3 client should be passed)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "8b3bc9c7-0072-40f5-b0ab-a7b9f48b4ea3",
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import os\n",
    "from botocore import UNSIGNED\n",
    "from botocore.client import Config\n",
    "\n",
    "def download_s3_objects_no_auth(bucket_name, download_path, prefix, substring, aws_region = None):\n",
    "    \"\"\"\n",
    "    Download all objects from a public S3 bucket (no authentication) that contain a specific prefix and substring in their keys.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    bucket_name : str\n",
    "        The name of the S3 bucket.\n",
    "    download_path : str\n",
    "        Local directory where the files will be downloaded.\n",
    "    prefix : str\n",
    "        Characters that are required to be at the begining of the S3 object keys.\n",
    "    substring : str\n",
    "        The substring to search for in the S3 object keys.\n",
    "    aws_region : str\n",
    "        AWS region where the S3 bucket is located (optional).\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    Donwloads all the files that match the search criteria into the download_path\n",
    "    \n",
    "    \"\"\"\n",
    "    # Initialize the S3 client with no request signing (public bucket)\n",
    "    s3_client = boto3.client('s3', config = Config(signature_version=UNSIGNED), region_name = aws_region)\n",
    "\n",
    "    # Ensure the download path exists\n",
    "    if not os.path.exists(download_path):\n",
    "        os.makedirs(download_path)\n",
    "\n",
    "    # List all objects in the bucket that contain the prefix\n",
    "    response = s3_client.list_objects_v2(Bucket = bucket_name, Prefix = prefix)\n",
    "    if 'Contents' not in response:\n",
    "        print(f\"No objects found in the bucket '{bucket_name}'.\")\n",
    "        return\n",
    "\n",
    "    # Loop through objects and download those that contain the substring\n",
    "    for obj in response['Contents']:\n",
    "        key = obj['Key']\n",
    "        if substring in key:\n",
    "            local_filename = os.path.join(download_path, key.split('/')[-1])\n",
    "            if not os.path.exists(local_filename):\n",
    "                print(f\"Downloading {key} to {local_filename}...\")\n",
    "                s3_client.download_file(bucket_name, key, local_filename)\n",
    "                print(f\"Downloaded: {local_filename}\")\n",
    "            else:\n",
    "                print(f\"Object {key} exists at {local_filename}. Download skipped.\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "48e07720-fb46-43f9-9307-bef460c110f0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Downloading csv/by_year/2020.csv to ./data/2020.csv...\n",
      "Downloaded: ./data/2020.csv\n",
      "Downloading csv/by_year/2021.csv to ./data/2021.csv...\n",
      "Downloaded: ./data/2021.csv\n",
      "Downloading csv/by_year/2022.csv to ./data/2022.csv...\n",
      "Downloaded: ./data/2022.csv\n",
      "Downloading csv/by_year/2023.csv to ./data/2023.csv...\n",
      "Downloaded: ./data/2023.csv\n",
      "Downloading csv/by_year/2024.csv to ./data/2024.csv...\n",
      "Downloaded: ./data/2024.csv\n"
     ]
    }
   ],
   "source": [
    "download_s3_objects_no_auth(bucket_name=\"noaa-ghcn-pds\", download_path=\"./data\", prefix=\"csv/by_year/\", substring=\"202\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a74dadf-5d3d-4224-895c-c34ad46ccd5e",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Import Packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "674b643b-48c1-4b48-992e-b865b5962ead",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask.dataframe as dd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "75332b54-8b3e-4e00-b405-9f8c75144ab5",
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "http://127.0.0.1:8787/status\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2024-10-21 15:47:38,592 - distributed.shuffle._scheduler_plugin - WARNING - Shuffle 8abbcac8bb534c5df63620fa29fbc810 initialized by task ('shuffle-transfer-8abbcac8bb534c5df63620fa29fbc810', 190) executed on worker tcp://127.0.0.1:39539\n",
      "2024-10-21 15:47:49,517 - distributed.shuffle._scheduler_plugin - WARNING - Shuffle 8abbcac8bb534c5df63620fa29fbc810 deactivated due to stimulus 'task-finished-1729525669.5162039'\n",
      "2024-10-21 15:48:01,550 - distributed.shuffle._scheduler_plugin - WARNING - Shuffle f8cfbf179fd7567be671c5bcff2f3225 initialized by task ('shuffle-transfer-f8cfbf179fd7567be671c5bcff2f3225', 249) executed on worker tcp://127.0.0.1:33909\n",
      "2024-10-21 15:48:13,706 - distributed.shuffle._scheduler_plugin - WARNING - Shuffle f8cfbf179fd7567be671c5bcff2f3225 deactivated due to stimulus 'task-finished-1729525693.7046325'\n"
     ]
    }
   ],
   "source": [
    "from dask.distributed import Client, LocalCluster\n",
    "cluster = LocalCluster()\n",
    "client = Client(cluster)\n",
    "print(client.dashboard_link)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ca621f69-fe73-45b6-9ae5-ff53a8186123",
   "metadata": {},
   "source": [
    "### Read One CSV file"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e2b2ca93-70c2-47f4-b75e-fbce0e7bc6c8",
   "metadata": {},
   "source": [
    "Let's first load one CSV file and see how Dask Dataframe works. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "452d022d-e62d-4329-9640-fea39170b93f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = dd.read_csv(\"./data/2023.csv\", dtype = {\"Q_FLAG\" : \"object\"})"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3829d0e-567f-4f8c-b5df-85a6766ae35c",
   "metadata": {},
   "source": [
    "You can check the number of partitions that Dask by defualt selects. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "cd497f80-9942-42ec-b36a-ad25742cccc5",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "21"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.npartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8752de7e-6e81-4298-836f-6d47c260ea32",
   "metadata": {},
   "source": [
    "To change the number of partitions you need to define the `blocksize` in the `read_csv` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "9cb10290-4be6-49c5-b580-61d3ae0f2838",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = dd.read_csv(\"./data/2023.csv\", dtype = {\"Q_FLAG\" : \"object\"}, blocksize=25e6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "2e64bcca-1475-4443-bcc4-725f8dac24c4",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "54"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.npartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e6af28c-78df-49a8-9afb-b2bc6b1d6600",
   "metadata": {},
   "source": [
    "The following line is an unnecessary step and you should not run it regularly. This will try to load all the data into memory which can easily go beyond your memory limit."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "050ddd5a-d16e-42b1-9cc6-4944150d4c4f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# DO NOT RUN\n",
    "# df.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "df7d8f49-15fd-4536-b6af-19fe2cb30749",
   "metadata": {},
   "source": [
    "But if you would like to load a small portion of the data, you can load a few lines of the dataframe as following:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d094fe96-55cb-4151-825a-b754a0e4ac14",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>ID</th>\n",
       "      <th>DATE</th>\n",
       "      <th>ELEMENT</th>\n",
       "      <th>DATA_VALUE</th>\n",
       "      <th>M_FLAG</th>\n",
       "      <th>Q_FLAG</th>\n",
       "      <th>S_FLAG</th>\n",
       "      <th>OBS_TIME</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>AGM00060430</td>\n",
       "      <td>20230101</td>\n",
       "      <td>TMIN</td>\n",
       "      <td>101</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>S</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>AGM00060430</td>\n",
       "      <td>20230101</td>\n",
       "      <td>PRCP</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>S</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>AGM00060430</td>\n",
       "      <td>20230101</td>\n",
       "      <td>TAVG</td>\n",
       "      <td>148</td>\n",
       "      <td>H</td>\n",
       "      <td>NaN</td>\n",
       "      <td>S</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>AGM00060437</td>\n",
       "      <td>20230101</td>\n",
       "      <td>TMIN</td>\n",
       "      <td>87</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>S</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>AGM00060437</td>\n",
       "      <td>20230101</td>\n",
       "      <td>PRCP</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>S</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>96</th>\n",
       "      <td>US1MDAA0063</td>\n",
       "      <td>20231224</td>\n",
       "      <td>SNWD</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>N</td>\n",
       "      <td>730.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>97</th>\n",
       "      <td>US1MDAA0065</td>\n",
       "      <td>20231224</td>\n",
       "      <td>PRCP</td>\n",
       "      <td>5</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>N</td>\n",
       "      <td>645.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>98</th>\n",
       "      <td>US1MDAA0065</td>\n",
       "      <td>20231224</td>\n",
       "      <td>SNOW</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>N</td>\n",
       "      <td>645.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>99</th>\n",
       "      <td>US1MDAA0065</td>\n",
       "      <td>20231224</td>\n",
       "      <td>SNWD</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>N</td>\n",
       "      <td>645.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>100</th>\n",
       "      <td>US1MDAA0065</td>\n",
       "      <td>20231224</td>\n",
       "      <td>WESD</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>N</td>\n",
       "      <td>645.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5454 rows × 8 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "              ID      DATE ELEMENT  DATA_VALUE M_FLAG Q_FLAG S_FLAG  OBS_TIME\n",
       "0    AGM00060430  20230101    TMIN         101    NaN    NaN      S       NaN\n",
       "1    AGM00060430  20230101    PRCP           0    NaN    NaN      S       NaN\n",
       "2    AGM00060430  20230101    TAVG         148      H    NaN      S       NaN\n",
       "3    AGM00060437  20230101    TMIN          87    NaN    NaN      S       NaN\n",
       "4    AGM00060437  20230101    PRCP           0    NaN    NaN      S       NaN\n",
       "..           ...       ...     ...         ...    ...    ...    ...       ...\n",
       "96   US1MDAA0063  20231224    SNWD           0    NaN    NaN      N     730.0\n",
       "97   US1MDAA0065  20231224    PRCP           5    NaN    NaN      N     645.0\n",
       "98   US1MDAA0065  20231224    SNOW           0    NaN    NaN      N     645.0\n",
       "99   US1MDAA0065  20231224    SNWD           0    NaN    NaN      N     645.0\n",
       "100  US1MDAA0065  20231224    WESD           0    NaN    NaN      N     645.0\n",
       "\n",
       "[5454 rows x 8 columns]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.loc[:100, :].compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "b3967cbd-4965-49b8-9c0e-7983f797fb66",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>ID</th>\n",
       "      <th>DATE</th>\n",
       "      <th>ELEMENT</th>\n",
       "      <th>DATA_VALUE</th>\n",
       "      <th>M_FLAG</th>\n",
       "      <th>Q_FLAG</th>\n",
       "      <th>S_FLAG</th>\n",
       "      <th>OBS_TIME</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=54</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>string</td>\n",
       "      <td>int64</td>\n",
       "      <td>string</td>\n",
       "      <td>int64</td>\n",
       "      <td>string</td>\n",
       "      <td>string</td>\n",
       "      <td>string</td>\n",
       "      <td>float64</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<div>Dask Name: read_csv, 1 expression</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                    ID   DATE ELEMENT DATA_VALUE  M_FLAG  Q_FLAG  S_FLAG OBS_TIME\n",
       "npartitions=54                                                                   \n",
       "                string  int64  string      int64  string  string  string  float64\n",
       "                   ...    ...     ...        ...     ...     ...     ...      ...\n",
       "...                ...    ...     ...        ...     ...     ...     ...      ...\n",
       "                   ...    ...     ...        ...     ...     ...     ...      ...\n",
       "                   ...    ...     ...        ...     ...     ...     ...      ...\n",
       "Dask Name: read_csv, 1 expression\n",
       "Expr=ReadCSV(119441c)"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5752139a-fe3f-46f0-a643-b723aa5fc54c",
   "metadata": {},
   "source": [
    "As you see, `df` is empty again. This is because Dask does not store the outputs of the `df.compute()` back in `df`. If you need to keep these values, you should instead run `df = df.compute()`. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2c8bda25-43f0-49f2-8caa-6f4e0d55f6a4",
   "metadata": {},
   "source": [
    "### Read Multiple CSVs"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ef506c2a-b101-4a1a-8eb5-e5c3be974fed",
   "metadata": {},
   "source": [
    "Here, we will define a new df and load multiple CSV files. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "c12bb7db-66ec-4185-b976-433b0bbd6bf7",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "large_df = dd.read_csv(\"./data/*.csv\", dtype = {\"Q_FLAG\" : \"object\"}, blocksize=25e6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "77ed7865-916d-477d-acf1-f206279acb48",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "252"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "large_df.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "2a829bca-5ccc-4f1d-8762-d2c63346ca23",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>ID</th>\n",
       "      <th>DATE</th>\n",
       "      <th>ELEMENT</th>\n",
       "      <th>DATA_VALUE</th>\n",
       "      <th>M_FLAG</th>\n",
       "      <th>Q_FLAG</th>\n",
       "      <th>S_FLAG</th>\n",
       "      <th>OBS_TIME</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=252</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>string</td>\n",
       "      <td>int64</td>\n",
       "      <td>string</td>\n",
       "      <td>int64</td>\n",
       "      <td>string</td>\n",
       "      <td>string</td>\n",
       "      <td>string</td>\n",
       "      <td>float64</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<div>Dask Name: read_csv, 1 expression</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                     ID   DATE ELEMENT DATA_VALUE  M_FLAG  Q_FLAG  S_FLAG OBS_TIME\n",
       "npartitions=252                                                                   \n",
       "                 string  int64  string      int64  string  string  string  float64\n",
       "                    ...    ...     ...        ...     ...     ...     ...      ...\n",
       "...                 ...    ...     ...        ...     ...     ...     ...      ...\n",
       "                    ...    ...     ...        ...     ...     ...     ...      ...\n",
       "                    ...    ...     ...        ...     ...     ...     ...      ...\n",
       "Dask Name: read_csv, 1 expression\n",
       "Expr=ReadCSV(ffb3976)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "large_df"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2dcedaea-419d-438d-ae0a-c9f120d066a3",
   "metadata": {},
   "source": [
    "Let's calculate the mean of each type of observation in whole dataset. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "e50d2761-3810-4020-b01d-d051918916b6",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "mean_values = large_df.groupby(\"ELEMENT\")[\"DATA_VALUE\"].mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "2507c91f-814b-4c80-a875-327a9da8dedb",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Dask Series Structure:\n",
       "npartitions=1\n",
       "    float64\n",
       "        ...\n",
       "Dask Name: getitem, 4 expressions\n",
       "Expr=((ReadCSV(ffb3976)[['ELEMENT', 'DATA_VALUE']]).mean(observed=False, chunk_kwargs={'numeric_only': False}, aggregate_kwargs={'numeric_only': False}, _slice='DATA_VALUE'))['DATA_VALUE']"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mean_values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "242dfa6f-6be8-4da0-8f5b-6c84ee62cf51",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ELEMENT\n",
       "SN56     65.149429\n",
       "TAVG    109.644308\n",
       "WDF5    200.575220\n",
       "SN31    117.909843\n",
       "SX53    101.728783\n",
       "           ...    \n",
       "PSUN     42.392204\n",
       "SN32    143.052831\n",
       "SN55     80.291063\n",
       "SX51    167.196816\n",
       "DASF      2.545455\n",
       "Name: DATA_VALUE, Length: 75, dtype: float64"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mean_values.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ac0694e-0280-4a34-9313-6c148c4816e0",
   "metadata": {},
   "source": [
    "Next, we will select a station in Worcester, MA and calculate the mean for each observation. \n",
    "You can see the list of all stations on NOAA's website [here](https://www.ncei.noaa.gov/pub/data/ghcn/daily/ghcnd-stations.txt). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "aad68ba6-9a52-4925-86fa-21fda28a0525",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "worcester_df = large_df[large_df[\"ID\"].isin([\"US1MAWR0097\"])]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "52eaec1c-b2b6-4c11-8edf-195b7a8e7f53",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>ID</th>\n",
       "      <th>DATE</th>\n",
       "      <th>ELEMENT</th>\n",
       "      <th>DATA_VALUE</th>\n",
       "      <th>M_FLAG</th>\n",
       "      <th>Q_FLAG</th>\n",
       "      <th>S_FLAG</th>\n",
       "      <th>OBS_TIME</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=252</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>string</td>\n",
       "      <td>int64</td>\n",
       "      <td>string</td>\n",
       "      <td>int64</td>\n",
       "      <td>string</td>\n",
       "      <td>string</td>\n",
       "      <td>string</td>\n",
       "      <td>float64</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th></th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<div>Dask Name: getitem, 5 expressions</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                     ID   DATE ELEMENT DATA_VALUE  M_FLAG  Q_FLAG  S_FLAG OBS_TIME\n",
       "npartitions=252                                                                   \n",
       "                 string  int64  string      int64  string  string  string  float64\n",
       "                    ...    ...     ...        ...     ...     ...     ...      ...\n",
       "...                 ...    ...     ...        ...     ...     ...     ...      ...\n",
       "                    ...    ...     ...        ...     ...     ...     ...      ...\n",
       "                    ...    ...     ...        ...     ...     ...     ...      ...\n",
       "Dask Name: getitem, 5 expressions\n",
       "Expr=Filter(frame=ReadCSV(ffb3976), predicate=Isin(frame=ReadCSV(ffb3976)['ID'], values=_DelayedExpr(Delayed('delayed-1e99729721b067d61cb17db688716cf6'))))"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "worcester_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "f100d120-6fc1-445f-9b30-669081f65200",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "worcester_mean = worcester_df.groupby(\"ELEMENT\")[\"DATA_VALUE\"].mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "93ec22a0-f675-4163-91e3-9f38fae52d43",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Dask Series Structure:\n",
       "npartitions=1\n",
       "    float64\n",
       "        ...\n",
       "Dask Name: getitem, 8 expressions\n",
       "Expr=(((Filter(frame=ReadCSV(ffb3976), predicate=Isin(frame=ReadCSV(ffb3976)['ID'], values=_DelayedExpr(Delayed('delayed-1e99729721b067d61cb17db688716cf6')))))[['ELEMENT', 'DATA_VALUE']]).mean(observed=False, chunk_kwargs={'numeric_only': False}, aggregate_kwargs={'numeric_only': False}, _slice='DATA_VALUE'))['DATA_VALUE']"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "worcester_mean"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ddb21958-9bf2-4b8f-9717-18dde7927c8e",
   "metadata": {},
   "source": [
    "Now, we want to calculate the mean but we are interested to keep these values in memory. So we will assign the output to a new variable `worcester_mean_values`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "4e38aadf-384c-4b1f-8c04-3982a32986e2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "worcester_mean_values = worcester_mean.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "1bd28f16-720f-4807-8b37-a0c2197cfab2",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ELEMENT\n",
       "SNWD    158.750000\n",
       "SNOW      2.433884\n",
       "PRCP     53.095327\n",
       "WESD      0.000000\n",
       "WESF      0.000000\n",
       "Name: DATA_VALUE, dtype: float64"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "worcester_mean_values"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3bcf63e0-df71-4043-8054-0a45cb89f84f",
   "metadata": {},
   "source": [
    "### Task: find the station with the highest number of snow days"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "37be823c-a71d-4c5a-a421-7c9fe0d6b505",
   "metadata": {},
   "source": [
    "Write a function that received the name of a observation (e.g. `PRCP`) from the NOAA GHCN-D dataset, and returns the ID of the station(s) with the highest number of days that the target observation has been more than 0. \n",
    "\n",
    "Use this function to find the station that has the highest number of snow days across years 2020-2024:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c9a8798-e119-4b25-bc6d-bee40f707b1a",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "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.12.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}