2_Local lakehouse with MinIO and Delta.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "21b0d8a6",
   "metadata": {},
   "source": [
    "# 2. Local lakehouse with MinIO and Delta"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "70c053be-de24-4735-856f-ce01ff25e7d6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "import polars as pl"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ac5ca095",
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext dotenv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "a1333255",
   "metadata": {},
   "outputs": [],
   "source": [
    "%dotenv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "735ec6c3-6f69-441d-99de-9846347935f6",
   "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": 5,
   "id": "6042227e",
   "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: (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;AITA for kicking guests out of…</td><td>&quot;TheHylind&quot;</td><td>2024-07-06 05:36:33 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>7</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwi…</td><td>&quot;For context, I work at a local…</td><td>&quot;Not enough info&quot;</td></tr><tr><td>&quot;AITA for reporting coworker&#x27;s …</td><td>&quot;Allethiia&quot;</td><td>2024-07-06 05:21:10 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>3</td><td>0.6</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwi…</td><td>&quot;I (28) have been working at my…</td><td>&quot;TL;DR&quot;</td></tr><tr><td>&quot;AITA for cancelling my birthda…</td><td>&quot;Lis_wj&quot;</td><td>2024-07-06 05:14:00 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>6</td><td>true</td><td>1</td><td>0.6</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwi…</td><td>&quot;I (26F) have been really stres…</td><td>&quot;Not the A-hole&quot;</td></tr><tr><td>&quot;AITA: I told my sister she has…</td><td>&quot;dswizzle2&quot;</td><td>2024-07-06 05:09:53 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>14</td><td>true</td><td>18</td><td>0.8</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwh…</td><td>&quot;For context, I(27F) and my sis…</td><td>&quot;Not the A-hole&quot;</td></tr><tr><td>&quot;WIBTA for calling out my frien…</td><td>&quot;gremlinoverlord_420&quot;</td><td>2024-07-06 04:57:21 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>9</td><td>true</td><td>0</td><td>0.33</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwh…</td><td>&quot;I (f) have gotten fed up with …</td><td>&quot;Not the A-hole&quot;</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",
       "│ AITA for   ┆ TheHylind ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ For       ┆ Not       │\n",
       "│ kicking    ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ context,  ┆ enough    │\n",
       "│ guests out ┆           ┆ 05:36:33  ┆           ┆   ┆         ┆ omments/1 ┆ I work at ┆ info      │\n",
       "│ of…        ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ a local…  ┆           │\n",
       "│ AITA for   ┆ Allethiia ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ I (28)    ┆ TL;DR     │\n",
       "│ reporting  ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ have been ┆           │\n",
       "│ coworker's ┆           ┆ 05:21:10  ┆           ┆   ┆         ┆ omments/1 ┆ working   ┆           │\n",
       "│ …          ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ at my…    ┆           │\n",
       "│ AITA for   ┆ Lis_wj    ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ I (26F)   ┆ Not the   │\n",
       "│ cancelling ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ have been ┆ A-hole    │\n",
       "│ my         ┆           ┆ 05:14:00  ┆           ┆   ┆         ┆ omments/1 ┆ really    ┆           │\n",
       "│ birthda…   ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ stres…    ┆           │\n",
       "│ AITA: I    ┆ dswizzle2 ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ For       ┆ Not the   │\n",
       "│ told my    ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ context,  ┆ A-hole    │\n",
       "│ sister she ┆           ┆ 05:09:53  ┆           ┆   ┆         ┆ omments/1 ┆ I(27F)    ┆           │\n",
       "│ has…       ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwh…      ┆ and my    ┆           │\n",
       "│            ┆           ┆           ┆           ┆   ┆         ┆           ┆ sis…      ┆           │\n",
       "│ WIBTA for  ┆ gremlinov ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ I (f)     ┆ Not the   │\n",
       "│ calling    ┆ erlord_42 ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ have      ┆ A-hole    │\n",
       "│ out my     ┆ 0         ┆ 04:57:21  ┆           ┆   ┆         ┆ omments/1 ┆ gotten    ┆           │\n",
       "│ frien…     ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwh…      ┆ fed up    ┆           │\n",
       "│            ┆           ┆           ┆           ┆   ┆         ┆           ┆ with …    ┆           │\n",
       "└────────────┴───────────┴───────────┴───────────┴───┴─────────┴───────────┴───────────┴───────────┘"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = pl.read_parquet(\"submissions.pq\").sort(\"creation_datetime\", descending=True)\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ab558297",
   "metadata": {},
   "source": [
    "Writing to a local Delta table is okay, but doing so to an object-storage is even better! Enter MinIO:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "5916f76e",
   "metadata": {},
   "outputs": [],
   "source": [
    "minio_storage_options = {\n",
    "    \"AWS_ENDPOINT_URL\": minio_endpoint_url,\n",
    "    \"AWS_ACCESS_KEY_ID\": minio_access_key,\n",
    "    \"AWS_SECRET_ACCESS_KEY\": minio_secret_id,\n",
    "    \"AWS_ALLOW_HTTP\": \"true\",\n",
    "    \"AWS_S3_ALLOW_UNSAFE_RENAME\": \"true\",\n",
    "    \"AWS_EC2_METADATA_DISABLED\": \"true\",\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2417d642",
   "metadata": {},
   "outputs": [],
   "source": [
    "df.write_delta(\n",
    "    \"s3://reddit-submissions/submissions-raw\",\n",
    "    mode=\"error\",  # To avoid accidentally overwriting the data\n",
    "    storage_options=minio_storage_options,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fd6efb8a",
   "metadata": {},
   "source": [
    "Better than just overwriting the data, Delta allows you to do `UPSERT`-like operations (UPdate and inSERT):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "4dd201d6",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'num_source_rows': 668,\n",
       " 'num_target_rows_inserted': 507,\n",
       " 'num_target_rows_updated': 161,\n",
       " 'num_target_rows_deleted': 0,\n",
       " 'num_target_rows_copied': 17,\n",
       " 'num_output_rows': 685,\n",
       " 'num_target_files_added': 2,\n",
       " 'num_target_files_removed': 1,\n",
       " 'execution_time_ms': 243,\n",
       " 'scan_time_ms': 0,\n",
       " 'rewrite_time_ms': 163}"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(\n",
    "    df.write_delta(\n",
    "        \"s3://reddit-submissions/submissions-raw\",\n",
    "        mode=\"merge\",\n",
    "        storage_options=minio_storage_options,\n",
    "        delta_merge_options={\n",
    "            \"predicate\": \"s.permalink = t.permalink\",\n",
    "            \"source_alias\": \"s\",\n",
    "            \"target_alias\": \"t\",\n",
    "        },\n",
    "    )\n",
    "    .when_matched_update_all()\n",
    "    .when_not_matched_insert_all()\n",
    "    .execute()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f59c34ac",
   "metadata": {},
   "source": [
    "And now you can read your data as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "d8e427f0",
   "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: (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;AITA for telling my sister if …</td><td>&quot;aitaeatinghabits&quot;</td><td>2024-07-06 05:39:18 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/1dwi…</td><td>&quot;I (20f) just learned I am auti…</td><td>null</td></tr><tr><td>&quot;AITA for kicking guests out of…</td><td>&quot;TheHylind&quot;</td><td>2024-07-06 05:36:33 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>7</td><td>true</td><td>1</td><td>1.0</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwi…</td><td>&quot;For context, I work at a local…</td><td>&quot;Not enough info&quot;</td></tr><tr><td>&quot;AITA for telling my bff to lea…</td><td>&quot;Direct-Television426&quot;</td><td>2024-07-06 05:34:17 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/1dwi…</td><td>&quot;So, my partner and I are both …</td><td>null</td></tr><tr><td>&quot;AITA for reporting coworker&#x27;s …</td><td>&quot;Allethiia&quot;</td><td>2024-07-06 05:21:10 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>2</td><td>true</td><td>3</td><td>0.6</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwi…</td><td>&quot;I (28) have been working at my…</td><td>&quot;TL;DR&quot;</td></tr><tr><td>&quot;AITA for cancelling my birthda…</td><td>&quot;Lis_wj&quot;</td><td>2024-07-06 05:14:00 UTC</td><td>&quot;r/AmItheAsshole&quot;</td><td>6</td><td>true</td><td>1</td><td>0.6</td><td>true</td><td>&quot;/r/AmItheAsshole/comments/1dwi…</td><td>&quot;I (26F) have been really stres…</td><td>&quot;Not the A-hole&quot;</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",
       "│ AITA for   ┆ aitaeatin ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ I (20f)   ┆ null      │\n",
       "│ telling my ┆ ghabits   ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ just      ┆           │\n",
       "│ sister if  ┆           ┆ 05:39:18  ┆           ┆   ┆         ┆ omments/1 ┆ learned I ┆           │\n",
       "│ …          ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ am auti…  ┆           │\n",
       "│ AITA for   ┆ TheHylind ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ For       ┆ Not       │\n",
       "│ kicking    ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ context,  ┆ enough    │\n",
       "│ guests out ┆           ┆ 05:36:33  ┆           ┆   ┆         ┆ omments/1 ┆ I work at ┆ info      │\n",
       "│ of…        ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ a local…  ┆           │\n",
       "│ AITA for   ┆ Direct-Te ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ So, my    ┆ null      │\n",
       "│ telling my ┆ levision4 ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ partner   ┆           │\n",
       "│ bff to     ┆ 26        ┆ 05:34:17  ┆           ┆   ┆         ┆ omments/1 ┆ and I are ┆           │\n",
       "│ lea…       ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ both …    ┆           │\n",
       "│ AITA for   ┆ Allethiia ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ I (28)    ┆ TL;DR     │\n",
       "│ reporting  ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ have been ┆           │\n",
       "│ coworker's ┆           ┆ 05:21:10  ┆           ┆   ┆         ┆ omments/1 ┆ working   ┆           │\n",
       "│ …          ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ at my…    ┆           │\n",
       "│ AITA for   ┆ Lis_wj    ┆ 2024-07-0 ┆ r/AmItheA ┆ … ┆ true    ┆ /r/AmIthe ┆ I (26F)   ┆ Not the   │\n",
       "│ cancelling ┆           ┆ 6         ┆ sshole    ┆   ┆         ┆ Asshole/c ┆ have been ┆ A-hole    │\n",
       "│ my         ┆           ┆ 05:14:00  ┆           ┆   ┆         ┆ omments/1 ┆ really    ┆           │\n",
       "│ birthda…   ┆           ┆ UTC       ┆           ┆   ┆         ┆ dwi…      ┆ stres…    ┆           │\n",
       "└────────────┴───────────┴───────────┴───────────┴───┴─────────┴───────────┴───────────┴───────────┘"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = (\n",
    "    pl.read_delta(\n",
    "        \"s3://reddit-submissions/submissions-raw\", storage_options=minio_storage_options\n",
    "    ).sort(\"creation_datetime\", descending=True)\n",
    ")\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "59638cc2",
   "metadata": {},
   "source": [
    "Delta tables allow for _time travel_:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9ec51992",
   "metadata": {},
   "outputs": [],
   "source": [
    "df0 = (\n",
    "    pl.read_delta(\n",
    "        \"s3://reddit-submissions/submissions-raw\",\n",
    "        version=0,  # The first version of the data, YMMV\n",
    "        storage_options=minio_storage_options\n",
    "    ).sort(\"creation_datetime\", descending=True)\n",
    ")\n",
    "df0.head()"
   ]
  }
 ],
 "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.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}