3_Data and ML pipelines with Kedro.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f44b92b1-640b-4ee4-8dd8-2041f9438a8e",
     "showTitle": false,
     "title": ""
    },
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "source": [
    "# 3. Data and ML pipelines with Kedro"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "40a586fc-ce16-4c59-9460-e922ecaf8354",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## The `DataCatalog`\n",
    "\n",
    "Kedro’s [Data Catalog](https://docs.kedro.org/en/latest/data/) is a registry of all data sources available for use by the project. It offers a separate place to declare details of the datasets your projects use. Kedro provides built-in datasets for different file types and file systems so you don’t have to write any of the logic for reading or writing data.\n",
    "\n",
    "Kedro offers a range of datasets, including CSV, Excel, Parquet, Feather, HDF5, JSON, Pickle, SQL Tables, SQL Queries, Spark DataFrames, and more. They are supported with the APIs of pandas, spark, networkx, matplotlib, yaml, and beyond. It relies on fsspec to read and save data from a variety of data stores including local file systems, network file systems, cloud object stores, and Hadoop. You can pass arguments in to load and save operations, and use versioning and credentials for data access.\n",
    "\n",
    "To start using the Data Catalog, create an instance of the `DataCatalog` class with a dictionary configuration as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "fd304399-077e-4fe8-88c1-70983bbc57ce",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from kedro.io import DataCatalog"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d61c1cf4-f385-4247-83c2-55e13a5d2a48",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Using the Parquet file for now for simplicity\n",
    "catalog = DataCatalog.from_config(\n",
    "    {\n",
    "        \"reddit_submissions_raw\": {\n",
    "            \"type\": \"polars.EagerPolarsDataset\",\n",
    "            \"file_format\": \"parquet\",\n",
    "            \"filepath\": \"submissions.pq\",\n",
    "        }\n",
    "    },\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "241709ef-8951-491e-9365-a561126b3095",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Each entry in the dictionary represents a **dataset**, and each dataset has a **type** as well as some extra properties. Datasets are Python classes that take care of all the I/O needs in Kedro. In this case, we're using `kedro_datasets.ibis.TableDataset`, you can read [its full documentation](https://docs.kedro.org/projects/kedro-datasets/en/kedro-datasets-3.0.1/api/kedro_datasets.ibis.TableDataset.html) online.\n",
    "\n",
    "After the catalog is created, `catalog.list()` will yield a list of the available dataset names, which you can load using the `catalog.load(<dataset_name>)` method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4571dc9f-1e89-4fb0-9488-48c730526612",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['reddit_submissions_raw']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "catalog.list()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "7dce890f-60be-4c01-bee7-b4268f562b9e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "df = catalog.load(\"reddit_submissions_raw\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c83a22ca-0de8-45cd-b62b-b75375ef1a83",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Notice that the resulting object is the exact same Polars `DataFrame` we were using previously!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9a5b9561-c64f-4419-91a5-c731d78667a8",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "polars.dataframe.frame.DataFrame"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5a501aff-2227-4f19-b914-2d1272bf95aa",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (5, 12)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>title</th><th>author_name</th><th>creation_datetime</th><th>subreddit_name</th><th>num_comments</th><th>sfw</th><th>score</th><th>upvote_ratio</th><th>is_self</th><th>permalink</th><th>selftext</th><th>flair_text</th></tr><tr><td>str</td><td>str</td><td>datetime[μs, UTC]</td><td>str</td><td>i64</td><td>bool</td><td>i64</td><td>f64</td><td>bool</td><td>str</td><td>str</td><td>str</td></tr></thead><tbody><tr><td>&quot;WIBTA if I got back at my bfs …</td><td>&quot;Gioyxp3ez3z11&quot;</td><td>2024-07-08 05:16:00 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;WIBTA\n",
       " So my Bf 20s was in a b…</td><td>null</td></tr><tr><td>&quot;AITA for getting snarky with m…</td><td>&quot;nointroductionssosa&quot;</td><td>2024-07-08 05:15:57 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Hi all. It’s a long one. I app…</td><td>null</td></tr><tr><td>&quot;AITA for going on a holiday wh…</td><td>&quot;Many-Yogurt2479&quot;</td><td>2024-07-08 05:14:38 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Me and a group of three friend…</td><td>null</td></tr><tr><td>&quot;AITAH For arguing with family …</td><td>&quot;PaleontologistJust39&quot;</td><td>2024-07-08 05:14:05 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Hello, I&#x27;m 19M and my mother h…</td><td>null</td></tr><tr><td>&quot;AITA for trying to convince my…</td><td>&quot;Throw-whyevenstart&quot;</td><td>2024-07-08 05:11:09 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>4</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;My (38f) husband (40M) recentl…</td><td>null</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (5, 12)\n",
       "┌────────────┬───────────┬───────────┬───────────┬───┬─────────┬───────────┬───────────┬───────────┐\n",
       "│ title      ┆ author_na ┆ creation_ ┆ subreddit ┆ … ┆ is_self ┆ permalink ┆ selftext  ┆ flair_tex │\n",
       "│ ---        ┆ me        ┆ datetime  ┆ _name     ┆   ┆ ---     ┆ ---       ┆ ---       ┆ t         │\n",
       "│ str        ┆ ---       ┆ ---       ┆ ---       ┆   ┆ bool    ┆ str       ┆ str       ┆ ---       │\n",
       "│            ┆ str       ┆ datetime[ ┆ str       ┆   ┆         ┆           ┆           ┆ str       │\n",
       "│            ┆           ┆ μs, UTC]  ┆           ┆   ┆         ┆           ┆           ┆           │\n",
       "╞════════════╪═══════════╪═══════════╪═══════════╪═══╪═════════╪═══════════╪═══════════╪═══════════╡\n",
       "│ WIBTA if I ┆ Gioyxp3ez ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ WIBTA     ┆ null      │\n",
       "│ got back   ┆ 3z11      ┆ 8         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ So my Bf  ┆           │\n",
       "│ at my bfs  ┆           ┆ 05:16:00  ┆           ┆   ┆         ┆ omments/1 ┆ 20s was   ┆           │\n",
       "│ …          ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ in a b…   ┆           │\n",
       "│ AITA for   ┆ nointrodu ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ Hi all.   ┆ null      │\n",
       "│ getting    ┆ ctionssos ┆ 8         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ It’s a    ┆           │\n",
       "│ snarky     ┆ a         ┆ 05:15:57  ┆           ┆   ┆         ┆ omments/1 ┆ long one. ┆           │\n",
       "│ with m…    ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ I app…    ┆           │\n",
       "│ AITA for   ┆ Many-Yogu ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ Me and a  ┆ null      │\n",
       "│ going on a ┆ rt2479    ┆ 8         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ group of  ┆           │\n",
       "│ holiday    ┆           ┆ 05:14:38  ┆           ┆   ┆         ┆ omments/1 ┆ three     ┆           │\n",
       "│ wh…        ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ friend…   ┆           │\n",
       "│ AITAH For  ┆ Paleontol ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ Hello,    ┆ null      │\n",
       "│ arguing    ┆ ogistJust ┆ 8         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ I'm 19M   ┆           │\n",
       "│ with       ┆ 39        ┆ 05:14:05  ┆           ┆   ┆         ┆ omments/1 ┆ and my    ┆           │\n",
       "│ family …   ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ mother h… ┆           │\n",
       "│ AITA for   ┆ Throw-why ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ My (38f)  ┆ null      │\n",
       "│ trying to  ┆ evenstart ┆ 8         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ husband   ┆           │\n",
       "│ convince   ┆           ┆ 05:11:09  ┆           ┆   ┆         ┆ omments/1 ┆ (40M)     ┆           │\n",
       "│ my…        ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ recentl…  ┆           │\n",
       "└────────────┴───────────┴───────────┴───────────┴───┴─────────┴───────────┴───────────┴───────────┘"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dd0a517e-50cc-42f4-8278-8a8c31f8d2da",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## The `OmegaConfigLoader`\n",
    "\n",
    "Instead of creating the Data Catalog by hand like this, Kedro usually stores configuration in YAML files. To load them, Kedro offers a [configuration loader](https://docs.kedro.org/en/latest/configuration/configuration_basics.html) based on the [Omegaconf](https://omegaconf.readthedocs.io/) library called the `OmegaConfigLoader`. This adds several interesting features, such as\n",
    "\n",
    "- Consolidating different configuration files into one\n",
    "- Substitution, templating\n",
    "- [Resolvers](https://omegaconf.readthedocs.io/en/2.3_branch/custom_resolvers.html)\n",
    "- And [much more](https://docs.kedro.org/en/latest/configuration/advanced_configuration.html)\n",
    "\n",
    "To start using it, first dump the catalog configuration to a `catalog.yml` file, and then use `OmegaConfigLoader` as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "771df6d0-a259-44c1-ab0a-6f8aaa0cb631",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing catalog.yml\n"
     ]
    }
   ],
   "source": [
    "%%writefile catalog.yml\n",
    "reddit_submissions_raw:\n",
    "  type: polars.EagerPolarsDataset\n",
    "  file_format: parquet\n",
    "  filepath: submissions.pq"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dcb2acde-31de-4886-ac60-3bb5f0546fb7",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from kedro.config import OmegaConfigLoader\n",
    "\n",
    "config_loader = OmegaConfigLoader(\n",
    "    conf_source=\".\",  # Directory where configuration files are located\n",
    "    config_patterns={\"catalog\": [\"catalog.yml\"]},  # For simplicity for this notebook\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "433dca34-8072-47bb-956b-1b12353ddb33",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'reddit_submissions_raw': {'type': 'polars.EagerPolarsDataset',\n",
       "  'file_format': 'parquet',\n",
       "  'filepath': 'submissions.pq'}}"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "catalog_config = config_loader.get(\"catalog\")\n",
    "catalog_config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c43fa6bf-5d27-4083-b673-5d29ea5a8df9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "catalog = DataCatalog.from_config(catalog_config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9ca78e97-ca59-479d-96a4-7377cb090a4a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (1, 12)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>title</th><th>author_name</th><th>creation_datetime</th><th>subreddit_name</th><th>num_comments</th><th>sfw</th><th>score</th><th>upvote_ratio</th><th>is_self</th><th>permalink</th><th>selftext</th><th>flair_text</th></tr><tr><td>str</td><td>str</td><td>datetime[μs, UTC]</td><td>str</td><td>i64</td><td>bool</td><td>i64</td><td>f64</td><td>bool</td><td>str</td><td>str</td><td>str</td></tr></thead><tbody><tr><td>&quot;WIBTA if I got back at my bfs …</td><td>&quot;Gioyxp3ez3z11&quot;</td><td>2024-07-08 05:16:00 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;WIBTA\n",
       " So my Bf 20s was in a b…</td><td>null</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (1, 12)\n",
       "┌────────────┬───────────┬───────────┬───────────┬───┬─────────┬───────────┬───────────┬───────────┐\n",
       "│ title      ┆ author_na ┆ creation_ ┆ subreddit ┆ … ┆ is_self ┆ permalink ┆ selftext  ┆ flair_tex │\n",
       "│ ---        ┆ me        ┆ datetime  ┆ _name     ┆   ┆ ---     ┆ ---       ┆ ---       ┆ t         │\n",
       "│ str        ┆ ---       ┆ ---       ┆ ---       ┆   ┆ bool    ┆ str       ┆ str       ┆ ---       │\n",
       "│            ┆ str       ┆ datetime[ ┆ str       ┆   ┆         ┆           ┆           ┆ str       │\n",
       "│            ┆           ┆ μs, UTC]  ┆           ┆   ┆         ┆           ┆           ┆           │\n",
       "╞════════════╪═══════════╪═══════════╪═══════════╪═══╪═════════╪═══════════╪═══════════╪═══════════╡\n",
       "│ WIBTA if I ┆ Gioyxp3ez ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ WIBTA     ┆ null      │\n",
       "│ got back   ┆ 3z11      ┆ 8         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ So my Bf  ┆           │\n",
       "│ at my bfs  ┆           ┆ 05:16:00  ┆           ┆   ┆         ┆ omments/1 ┆ 20s was   ┆           │\n",
       "│ …          ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ in a b…   ┆           │\n",
       "└────────────┴───────────┴───────────┴───────────┴───┴─────────┴───────────┴───────────┴───────────┘"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "catalog.load(\"reddit_submissions_raw\").head(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0f1786d5-0d36-4e8f-8d34-7ee14d3fd747",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Nodes and pipelines\n",
    "\n",
    "Now comes the interesting part. Kedro structures the computation on Directed Acyclic Graphs (DAGs), which are created by instantiating `Pipeline` objects with a list of `Node`s. By linking the inputs and outpus of each node, Kedro is then able to perform a topological sort and produce a graph.\n",
    "\n",
    "Let's start creating a trivial pipeline with 2 nodes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "daa6af0f-a4b7-4a0c-929c-b8150bad90a1",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "import polars as pl\n",
    "\n",
    "\n",
    "def enrich_submissions(df: pl.DataFrame) -> pl.DataFrame:\n",
    "    # Two types of posts: AITA and WIBTA https://www.reddit.com/r/AmItheAsshole/wiki/howtopost/\n",
    "    enriched_df = (\n",
    "        df.with_columns(\n",
    "            pl.col(\"title\").str.extract(r\"^(AITA|WIBTA)\", 1).alias(\"post_type\"),\n",
    "            pl.col(\"selftext\").str.len_chars().alias(\"text_length\"),\n",
    "        )\n",
    "    )\n",
    "    return enriched_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from nltk.sentiment.vader import SentimentIntensityAnalyzer\n",
    "\n",
    "\n",
    "def sentiment_analysis_by_sentences(df: pl.DataFrame) -> pl.DataFrame:\n",
    "    sia = SentimentIntensityAnalyzer()\n",
    "    sentences = (\n",
    "        df.with_columns(\n",
    "            pl.col(\"selftext\").str.split(\".\").list.eval(pl.element().str.strip_chars()).alias(\"sentences\")\n",
    "        )\n",
    "        .select(pl.col(\"permalink\", \"sentences\"))\n",
    "        .explode(\"sentences\")\n",
    "        .with_columns(\n",
    "            pl.col(\"sentences\").map_elements(\n",
    "                lambda s: sia.polarity_scores(s),\n",
    "                return_dtype=pl.Struct({\"neg\": pl.Float64, \"neu\": pl.Float64, \"pos\": pl.Float64, \"compound\": pl.Float64}),\n",
    "            ).alias(\"sentiment_scores\"),\n",
    "        )\n",
    "    )\n",
    "    return sentences"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def create_model_input_table(df: pl.DataFrame, sentences: pl.DataFrame) -> pl.DataFrame:\n",
    "    return df.join(\n",
    "        (\n",
    "            sentences\n",
    "            .group_by(\"permalink\")\n",
    "            .agg(\n",
    "                pl.col(\"sentiment_scores\").struct.field(\"compound\").mean().alias(\"compound_sentiment\"),\n",
    "            )\n",
    "        ),\n",
    "        on=\"permalink\",\n",
    "        how=\"left\",\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1fab6879-716b-42dd-bc5a-d0bfed9ff5c5",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (1, 14)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>title</th><th>author_name</th><th>creation_datetime</th><th>subreddit_name</th><th>num_comments</th><th>sfw</th><th>score</th><th>upvote_ratio</th><th>is_self</th><th>permalink</th><th>selftext</th><th>flair_text</th><th>post_type</th><th>text_length</th></tr><tr><td>str</td><td>str</td><td>datetime[μs, UTC]</td><td>str</td><td>i64</td><td>bool</td><td>i64</td><td>f64</td><td>bool</td><td>str</td><td>str</td><td>str</td><td>str</td><td>u32</td></tr></thead><tbody><tr><td>&quot;WIBTA if I got back at my bfs …</td><td>&quot;Gioyxp3ez3z11&quot;</td><td>2024-07-08 05:16:00 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;WIBTA\n",
       " So my Bf 20s was in a b…</td><td>null</td><td>&quot;WIBTA&quot;</td><td>1373</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (1, 14)\n",
       "┌───────────┬───────────┬───────────┬───────────┬───┬───────────┬───────────┬───────────┬──────────┐\n",
       "│ title     ┆ author_na ┆ creation_ ┆ subreddit ┆ … ┆ selftext  ┆ flair_tex ┆ post_type ┆ text_len │\n",
       "│ ---       ┆ me        ┆ datetime  ┆ _name     ┆   ┆ ---       ┆ t         ┆ ---       ┆ gth      │\n",
       "│ str       ┆ ---       ┆ ---       ┆ ---       ┆   ┆ str       ┆ ---       ┆ str       ┆ ---      │\n",
       "│           ┆ str       ┆ datetime[ ┆ str       ┆   ┆           ┆ str       ┆           ┆ u32      │\n",
       "│           ┆           ┆ μs, UTC]  ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "╞═══════════╪═══════════╪═══════════╪═══════════╪═══╪═══════════╪═══════════╪═══════════╪══════════╡\n",
       "│ WIBTA if  ┆ Gioyxp3ez ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ WIBTA     ┆ null      ┆ WIBTA     ┆ 1373     │\n",
       "│ I got     ┆ 3z11      ┆ 8         ┆ sshole    ┆   ┆ So my Bf  ┆           ┆           ┆          │\n",
       "│ back at   ┆           ┆ 05:16:00  ┆           ┆   ┆ 20s was   ┆           ┆           ┆          │\n",
       "│ my bfs …  ┆           ┆ UTC       ┆           ┆   ┆ in a b…   ┆           ┆           ┆          │\n",
       "└───────────┴───────────┴───────────┴───────────┴───┴───────────┴───────────┴───────────┴──────────┘"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "enrich_submissions(df).head(n=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (1, 3)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>permalink</th><th>sentences</th><th>sentiment_scores</th></tr><tr><td>str</td><td>str</td><td>struct[4]</td></tr></thead><tbody><tr><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;WIBTA\n",
       " So my Bf 20s was in a b…</td><td>{0.0,1.0,0.0,0.0}</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (1, 3)\n",
       "┌─────────────────────────────────┬───────────────────────────┬───────────────────┐\n",
       "│ permalink                       ┆ sentences                 ┆ sentiment_scores  │\n",
       "│ ---                             ┆ ---                       ┆ ---               │\n",
       "│ str                             ┆ str                       ┆ struct[4]         │\n",
       "╞═════════════════════════════════╪═══════════════════════════╪═══════════════════╡\n",
       "│ /r/AmItheAsshole/comments/1dy0… ┆ WIBTA                     ┆ {0.0,1.0,0.0,0.0} │\n",
       "│                                 ┆  So my Bf 20s was in a b… ┆                   │\n",
       "└─────────────────────────────────┴───────────────────────────┴───────────────────┘"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sentiment_analysis_by_sentences(df).head(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c2b3fd81-9071-4ba6-b9ff-c83960ec534a",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Notice that these are plain Python functions, receiving Polars DataFrames and returning more Polars DataFrames.\n",
    "\n",
    "Now, let's wrap them using the `node` convenience function from Kedro:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "98b82f40-e81f-4f87-8baf-119cd7e754fc",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Node(enrich_submissions, 'reddit_submissions_raw', 'reddit_submissions_enriched', None)"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from kedro.pipeline import node\n",
    "\n",
    "node(\n",
    "    func=enrich_submissions,\n",
    "    inputs=\"reddit_submissions_raw\",\n",
    "    outputs=\"reddit_submissions_enriched\",\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4cbb449b-67a5-4ff6-bbe7-d4abf3ad44a5",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "Conceptually, a `Node` is a wrapper around a Python function that defines a single step in a pipeline. It has inputs and outputs, which are the names of the Data Catalog datasets that the function will receive and return, respectively. Therefore, you could execute it as follows:\n",
    "\n",
    "```python\n",
    "n0.func(\n",
    "    *[catalog.load(input_dataset) for input_dataset in n0.inputs],\n",
    ")\n",
    "```\n",
    "\n",
    "Let's not do that though; Kedro will take care of it.\n",
    "\n",
    "The next step is to assemble the pipeline:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "0cc02764-dd8e-497f-b28b-4605808fc69e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Pipeline([\n",
       "Node(enrich_submissions, 'reddit_submissions_raw', 'reddit_submissions_enriched', None),\n",
       "Node(sentiment_analysis_by_sentences, 'reddit_submissions_raw', 'reddit_sentiment_by_sentences', None),\n",
       "Node(create_model_input_table, ['reddit_submissions_enriched', 'reddit_sentiment_by_sentences'], 'reddit_model_input', None)\n",
       "])"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from kedro.pipeline import pipeline\n",
    "\n",
    "\n",
    "pipe = pipeline([\n",
    "    node(\n",
    "        func=enrich_submissions,\n",
    "        inputs=\"reddit_submissions_raw\",\n",
    "        outputs=\"reddit_submissions_enriched\",\n",
    "    ),\n",
    "    node(\n",
    "        func=sentiment_analysis_by_sentences,\n",
    "        inputs=\"reddit_submissions_raw\",\n",
    "        outputs=\"reddit_sentiment_by_sentences\",\n",
    "    ),\n",
    "    node(\n",
    "        func=create_model_input_table,\n",
    "        inputs=[\"reddit_submissions_enriched\", \"reddit_sentiment_by_sentences\"],\n",
    "        outputs=\"reddit_model_input\",\n",
    "    ),\n",
    "])\n",
    "pipe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9b67d597-593f-4732-b615-7994ca568d69",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "And finally, you can now execute the pipeline. For the purposes of this tutorial, you can use Kedro's `SequentialRunner` directly:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "\n",
    "logging.basicConfig(level=logging.INFO)\n",
    "\n",
    "# Workaround: Restore logging handlers, see https://github.com/kedro-org/kedro/issues/3985\n",
    "_old_handlers = logging.getLogger().handlers.copy()\n",
    "\n",
    "import kedro.runner\n",
    "\n",
    "logging.getLogger().handlers = _old_handlers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e1f9648e-f921-4971-9109-c795f32a4f36",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:kedro.runner.sequential_runner:Using synchronous mode for loading and saving data. Use the --async flag for potential performance gains. https://docs.kedro.org/en/stable/nodes_and_pipelines/run_a_pipeline.html#load-and-save-asynchronously\n",
      "INFO:kedro.io.data_catalog:Loading data from [dark_orange]reddit_submissions_raw[/dark_orange] (EagerPolarsDataset)...\n",
      "INFO:kedro.pipeline.node:Running node: enrich_submissions([reddit_submissions_raw]) -> [reddit_submissions_enriched]\n",
      "INFO:kedro.io.data_catalog:Saving data to [dark_orange]reddit_submissions_enriched[/dark_orange] (MemoryDataset)...\n",
      "INFO:kedro.runner.sequential_runner:Completed 1 out of 3 tasks\n",
      "INFO:kedro.io.data_catalog:Loading data from [dark_orange]reddit_submissions_raw[/dark_orange] (EagerPolarsDataset)...\n",
      "INFO:kedro.pipeline.node:Running node: sentiment_analysis_by_sentences([reddit_submissions_raw]) -> [reddit_sentiment_by_sentences]\n",
      "INFO:kedro.io.data_catalog:Saving data to [dark_orange]reddit_sentiment_by_sentences[/dark_orange] (MemoryDataset)...\n",
      "INFO:kedro.runner.sequential_runner:Completed 2 out of 3 tasks\n",
      "INFO:kedro.io.data_catalog:Loading data from [dark_orange]reddit_submissions_enriched[/dark_orange] (MemoryDataset)...\n",
      "INFO:kedro.io.data_catalog:Loading data from [dark_orange]reddit_sentiment_by_sentences[/dark_orange] (MemoryDataset)...\n",
      "INFO:kedro.pipeline.node:Running node: create_model_input_table([reddit_submissions_enriched;reddit_sentiment_by_sentences]) -> [reddit_model_input]\n",
      "INFO:kedro.io.data_catalog:Saving data to [dark_orange]reddit_model_input[/dark_orange] (MemoryDataset)...\n",
      "INFO:kedro.runner.sequential_runner:Completed 3 out of 3 tasks\n",
      "INFO:kedro.runner.sequential_runner:Pipeline execution completed successfully.\n",
      "INFO:kedro.io.data_catalog:Loading data from [dark_orange]reddit_model_input[/dark_orange] (MemoryDataset)...\n"
     ]
    }
   ],
   "source": [
    "from kedro.runner import SequentialRunner\n",
    "\n",
    "outputs = SequentialRunner().run(pipe, catalog=catalog)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a75082ec-44e5-4690-bec8-cb66159619e8",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "The output of the `.run(...)` method will be \"Any node outputs that cannot be processed by the `DataCatalog`\". Since `reddit_model_input` is not declared in the Data Catalog, it's right there in the dictionary:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "72581dd5-09d9-43d6-bdb4-8726d3aa0955",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"></pre>\n"
      ],
      "text/plain": []
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "\u001b[1;35mdict_keys\u001b[0m\u001b[1m(\u001b[0m\u001b[1m[\u001b[0m\u001b[32m'reddit_model_input'\u001b[0m\u001b[1m]\u001b[0m\u001b[1m)\u001b[0m"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "outputs.keys()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a5986e58-b2e9-4cd2-91ba-2045c955ca93",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"></pre>\n"
      ],
      "text/plain": []
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (5, 15)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>title</th><th>author_name</th><th>creation_datetime</th><th>subreddit_name</th><th>num_comments</th><th>sfw</th><th>score</th><th>upvote_ratio</th><th>is_self</th><th>permalink</th><th>selftext</th><th>flair_text</th><th>post_type</th><th>text_length</th><th>compound_sentiment</th></tr><tr><td>str</td><td>str</td><td>datetime[μs, UTC]</td><td>str</td><td>i64</td><td>bool</td><td>i64</td><td>f64</td><td>bool</td><td>str</td><td>str</td><td>str</td><td>str</td><td>u32</td><td>f64</td></tr></thead><tbody><tr><td>&quot;WIBTA if I got back at my bfs …</td><td>&quot;Gioyxp3ez3z11&quot;</td><td>2024-07-08 05:16:00 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;WIBTA\n",
       " So my Bf 20s was in a b…</td><td>null</td><td>&quot;WIBTA&quot;</td><td>1373</td><td>-0.124925</td></tr><tr><td>&quot;AITA for getting snarky with m…</td><td>&quot;nointroductionssosa&quot;</td><td>2024-07-08 05:15:57 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Hi all. It’s a long one. I app…</td><td>null</td><td>&quot;AITA&quot;</td><td>2717</td><td>0.130606</td></tr><tr><td>&quot;AITA for going on a holiday wh…</td><td>&quot;Many-Yogurt2479&quot;</td><td>2024-07-08 05:14:38 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Me and a group of three friend…</td><td>null</td><td>&quot;AITA&quot;</td><td>1772</td><td>0.117583</td></tr><tr><td>&quot;AITAH For arguing with family …</td><td>&quot;PaleontologistJust39&quot;</td><td>2024-07-08 05:14:05 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Hello, I&#x27;m 19M and my mother h…</td><td>null</td><td>&quot;AITA&quot;</td><td>736</td><td>-0.100257</td></tr><tr><td>&quot;AITA for trying to convince my…</td><td>&quot;Throw-whyevenstart&quot;</td><td>2024-07-08 05:11:09 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>4</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;My (38f) husband (40M) recentl…</td><td>null</td><td>&quot;AITA&quot;</td><td>1461</td><td>-0.016927</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "\n",
       "shape: \u001b[1m(\u001b[0m\u001b[1;36m5\u001b[0m, \u001b[1;36m15\u001b[0m\u001b[1m)\u001b[0m\n",
       "┌───────────┬───────────┬───────────┬───────────┬───┬───────────┬───────────┬───────────┬──────────┐\n",
       "│ title     ┆ author_na ┆ creation_ ┆ subreddit ┆ … ┆ flair_tex ┆ post_type ┆ text_leng ┆ compound │\n",
       "│ ---       ┆ me        ┆ datetime  ┆ _name     ┆   ┆ t         ┆ ---       ┆ th        ┆ _sentime │\n",
       "│ str       ┆ ---       ┆ ---       ┆ ---       ┆   ┆ ---       ┆ str       ┆ ---       ┆ nt       │\n",
       "│           ┆ str       ┆ datetime\u001b[1m[\u001b[0m ┆ str       ┆   ┆ str       ┆           ┆ u32       ┆ ---      │\n",
       "│           ┆           ┆ μs, UTC\u001b[1m]\u001b[0m  ┆           ┆   ┆           ┆           ┆           ┆ f64      │\n",
       "╞═══════════╪═══════════╪═══════════╪═══════════╪═══╪═══════════╪═══════════╪═══════════╪══════════╡\n",
       "│ WIBTA if  ┆ Gioyxp3ez ┆ \u001b[1;36m2024\u001b[0m-\u001b[1;36m07\u001b[0m-\u001b[1;36m0\u001b[0m ┆ r/AmItheA ┆ … ┆ null      ┆ WIBTA     ┆ \u001b[1;36m1373\u001b[0m      ┆ \u001b[1;36m-0.12492\u001b[0m │\n",
       "│ I got     ┆ 3z11      ┆ \u001b[1;36m8\u001b[0m         ┆ sshole    ┆   ┆           ┆           ┆           ┆ \u001b[1;36m5\u001b[0m        │\n",
       "│ back at   ┆           ┆ \u001b[1;92m05:16:00\u001b[0m  ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ my bfs …  ┆           ┆ UTC       ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ AITA for  ┆ nointrodu ┆ \u001b[1;36m2024\u001b[0m-\u001b[1;36m07\u001b[0m-\u001b[1;36m0\u001b[0m ┆ r/AmItheA ┆ … ┆ null      ┆ AITA      ┆ \u001b[1;36m2717\u001b[0m      ┆ \u001b[1;36m0.130606\u001b[0m │\n",
       "│ getting   ┆ ctionssos ┆ \u001b[1;36m8\u001b[0m         ┆ sshole    ┆   ┆           ┆           ┆           ┆          │\n",
       "│ snarky    ┆ a         ┆ \u001b[1;92m05:15:57\u001b[0m  ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ with m…   ┆           ┆ UTC       ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ AITA for  ┆ Many-Yogu ┆ \u001b[1;36m2024\u001b[0m-\u001b[1;36m07\u001b[0m-\u001b[1;36m0\u001b[0m ┆ r/AmItheA ┆ … ┆ null      ┆ AITA      ┆ \u001b[1;36m1772\u001b[0m      ┆ \u001b[1;36m0.117583\u001b[0m │\n",
       "│ going on  ┆ rt2479    ┆ \u001b[1;36m8\u001b[0m         ┆ sshole    ┆   ┆           ┆           ┆           ┆          │\n",
       "│ a holiday ┆           ┆ \u001b[1;92m05:14:38\u001b[0m  ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ wh…       ┆           ┆ UTC       ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ AITAH For ┆ Paleontol ┆ \u001b[1;36m2024\u001b[0m-\u001b[1;36m07\u001b[0m-\u001b[1;36m0\u001b[0m ┆ r/AmItheA ┆ … ┆ null      ┆ AITA      ┆ \u001b[1;36m736\u001b[0m       ┆ \u001b[1;36m-0.10025\u001b[0m │\n",
       "│ arguing   ┆ ogistJust ┆ \u001b[1;36m8\u001b[0m         ┆ sshole    ┆   ┆           ┆           ┆           ┆ \u001b[1;36m7\u001b[0m        │\n",
       "│ with      ┆ \u001b[1;36m39\u001b[0m        ┆ \u001b[1;92m05:14:05\u001b[0m  ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ family …  ┆           ┆ UTC       ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ AITA for  ┆ Throw-why ┆ \u001b[1;36m2024\u001b[0m-\u001b[1;36m07\u001b[0m-\u001b[1;36m0\u001b[0m ┆ r/AmItheA ┆ … ┆ null      ┆ AITA      ┆ \u001b[1;36m1461\u001b[0m      ┆ \u001b[1;36m-0.01692\u001b[0m │\n",
       "│ trying to ┆ evenstart ┆ \u001b[1;36m8\u001b[0m         ┆ sshole    ┆   ┆           ┆           ┆           ┆ \u001b[1;36m7\u001b[0m        │\n",
       "│ convince  ┆           ┆ \u001b[1;92m05:11:09\u001b[0m  ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "│ my…       ┆           ┆ UTC       ┆           ┆   ┆           ┆           ┆           ┆          │\n",
       "└───────────┴───────────┴───────────┴───────────┴───┴───────────┴───────────┴───────────┴──────────┘"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "outputs[\"reddit_model_input\"].head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise 1\n",
    "\n",
    "1. Create a `exclude_social_features` function that receives the raw data as a Polars `DataFrame` and returns another `Da only the `\"score\", \"num_comments\", \"upvote_ratio\"` columns.\n",
    "2. Rewrite the pipeline so that this function is in the first node. _Hint: You will have to introduce another intermediate dataset_."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/ex01_kedro_pipeline.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Appendix: Custom datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sometimes there is not an appropriate dataset in `kedro_datasets` that suits our needs. In that cases we need to write our own.\n",
    "\n",
    "Have a look at `reddit_a_predictor.datasets.PolarsDeltaDataset` to see how a simple, custom Kedro dataset works."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext dotenv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "%dotenv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "minio_endpoint_url = os.environ[\"MINIO_ENDPOINT_URL\"]\n",
    "minio_access_key = os.environ[\"MINIO_KEY\"]\n",
    "minio_secret_id = os.environ[\"MINIO_SECRET\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:kedro.io.data_catalog:Loading data from [dark_orange]reddit_submissions_raw[/dark_orange] (PolarsDeltaDataset)...\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"></pre>\n"
      ],
      "text/plain": []
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (1, 12)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>title</th><th>author_name</th><th>creation_datetime</th><th>subreddit_name</th><th>num_comments</th><th>sfw</th><th>score</th><th>upvote_ratio</th><th>is_self</th><th>permalink</th><th>selftext</th><th>flair_text</th></tr><tr><td>str</td><td>str</td><td>datetime[μs, UTC]</td><td>str</td><td>i64</td><td>bool</td><td>i64</td><td>f64</td><td>bool</td><td>str</td><td>str</td><td>str</td></tr></thead><tbody><tr><td>&quot;AITAH For arguing with family …</td><td>&quot;PaleontologistJust39&quot;</td><td>2024-07-08 05:14:05 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dy0…</td><td>&quot;Hello, I&#x27;m 19M and my mother h…</td><td>null</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "\n",
       "shape: \u001b[1m(\u001b[0m\u001b[1;36m1\u001b[0m, \u001b[1;36m12\u001b[0m\u001b[1m)\u001b[0m\n",
       "┌────────────┬───────────┬───────────┬───────────┬───┬─────────┬───────────┬───────────┬───────────┐\n",
       "│ title      ┆ author_na ┆ creation_ ┆ subreddit ┆ … ┆ is_self ┆ permalink ┆ selftext  ┆ flair_tex │\n",
       "│ ---        ┆ me        ┆ datetime  ┆ _name     ┆   ┆ ---     ┆ ---       ┆ ---       ┆ t         │\n",
       "│ str        ┆ ---       ┆ ---       ┆ ---       ┆   ┆ bool    ┆ str       ┆ str       ┆ ---       │\n",
       "│            ┆ str       ┆ datetime\u001b[1m[\u001b[0m ┆ str       ┆   ┆         ┆           ┆           ┆ str       │\n",
       "│            ┆           ┆ μs, UTC\u001b[1m]\u001b[0m  ┆           ┆   ┆         ┆           ┆           ┆           │\n",
       "╞════════════╪═══════════╪═══════════╪═══════════╪═══╪═════════╪═══════════╪═══════════╪═══════════╡\n",
       "│ AITAH For  ┆ Paleontol ┆ \u001b[1;36m2024\u001b[0m-\u001b[1;36m07\u001b[0m-\u001b[1;36m0\u001b[0m ┆ r/AmItheA ┆ … ┆ true    ┆ \u001b[35m/r/\u001b[0m\u001b[95mAmIthe\u001b[0m ┆ Hello,    ┆ null      │\n",
       "│ arguing    ┆ ogistJust ┆ \u001b[1;36m8\u001b[0m         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ I'm 19M   ┆           │\n",
       "│ with       ┆ \u001b[1;36m39\u001b[0m        ┆ \u001b[1;92m05:14:05\u001b[0m  ┆           ┆   ┆         ┆ omments/\u001b[1;36m1\u001b[0m ┆ and my    ┆           │\n",
       "│ family …   ┆           ┆ UTC       ┆           ┆   ┆         ┆ dy0…      ┆ mother h… ┆           │\n",
       "└────────────┴───────────┴───────────┴───────────┴───┴─────────┴───────────┴───────────┴───────────┘"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "catalog = DataCatalog.from_config(\n",
    "    {\n",
    "        \"reddit_submissions_raw\": {\n",
    "            \"type\": \"reddit_a_predictor.datasets.PolarsDeltaDataset\",\n",
    "            \"filepath\": \"s3://reddit-submissions/submissions-raw\",\n",
    "            \"credentials\": \"minio_object_store\",\n",
    "            \"storage_options\": {\n",
    "                \"AWS_ALLOW_HTTP\": \"true\",\n",
    "                \"AWS_S3_ALLOW_UNSAFE_RENAME\": \"true\",\n",
    "                \"AWS_EC2_METADATA_DISABLED\": \"true\",\n",
    "            }\n",
    "        }\n",
    "    },\n",
    "    # Credentials are treated separately\n",
    "    credentials={\n",
    "        \"minio_object_store\": {\n",
    "            \"AWS_ENDPOINT_URL\": minio_endpoint_url,\n",
    "            \"AWS_ACCESS_KEY_ID\": minio_access_key,\n",
    "            \"AWS_SECRET_ACCESS_KEY\": minio_secret_id,\n",
    "        }\n",
    "    }\n",
    ")\n",
    "catalog.load(\"reddit_submissions_raw\").head(1)"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "environmentMetadata": null,
   "language": "python",
   "notebookMetadata": {
    "mostRecentlyExecutedCommandWithImplicitDF": {
     "commandId": 4172678518454607,
     "dataframes": [
      "_sqldf"
     ]
    },
    "pythonIndentUnit": 4
   },
   "notebookName": "First Steps with Kedro on Databricks",
   "widgets": {}
  },
  "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.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}