dlt_csdb.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "GYREioraz1m6"
   },
   "source": [
    "# **[Demo] Create a `dlt` pipeline in 4 steps** 🚀\n",
    "\n",
    "---\n",
    "\n",
    "**TL;DR: create and execute a pipeline that loads data from the [PokeAPI](https://pokeapi.co/) into a [DuckDB](https://duckdb.org/) database using the `dlt` library**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "U33sg53V4xlR"
   },
   "source": [
    "## 1 **Install `dlt`**"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "GSfjew3L5Efg",
    "ExecuteTime": {
     "end_time": "2024-12-21T14:52:13.446080Z",
     "start_time": "2024-12-21T14:52:12.419571Z"
    }
   },
   "source": [
    "%%capture\n",
    "!pip install \"dlt[duckdb]\" # Install dlt with all the necessary DuckDB dependencies"
   ],
   "outputs": [],
   "execution_count": 4
  },
  {
   "cell_type": "code",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "t5FVypQXvr2t",
    "outputId": "26442baa-6a43-4882-9b97-29575d5c7122",
    "ExecuteTime": {
     "end_time": "2024-12-21T16:11:21.575459Z",
     "start_time": "2024-12-21T16:11:20.786282Z"
    }
   },
   "source": [
    "!dlt --version"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "dlt 0.3.23\r\n"
     ]
    }
   ],
   "execution_count": 1
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1wf1R0yQh7pv"
   },
   "source": [
    "## 2 **Import `dlt` and initialize the pipeline**"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "A3NRS0y38alk",
    "ExecuteTime": {
     "end_time": "2024-12-21T16:11:26.374240Z",
     "start_time": "2024-12-21T16:11:26.099536Z"
    }
   },
   "source": [
    "import dlt\n",
    "\n",
    "pipeline = dlt.pipeline(pipeline_name=\"csd_pipeline\",\n",
    "                        destination=\"duckdb\",\n",
    "                        dataset_name=\"csdb_releases\")"
   ],
   "outputs": [],
   "execution_count": 2
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "JgbaVBwZaYuF"
   },
   "source": [
    "## 3 **Get data from the source**\n",
    "\n",
    "We load the list of pokemons data using url."
   ]
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "M5CWr_eZGfps",
    "ExecuteTime": {
     "end_time": "2024-12-21T16:17:50.036758Z",
     "start_time": "2024-12-21T16:17:49.874118Z"
    }
   },
   "source": [
    "from dlt.sources.helpers import requests\n",
    "import xml.etree.ElementTree as ET\n",
    "\n",
    "POKEMON_URL = \"https://csdb.dk/webservice/?type=release&id=1&depth=1\"\n",
    "\n",
    "response = requests.get(POKEMON_URL)\n",
    "if response.status_code == 200:\n",
    "    root = ET.fromstring(response.content)\n",
    "\n",
    "    # Extract and transform 'results' data into list of dictionaries\n",
    "    results = root.findall(\".//results\")  # Use correct path based on XML\n",
    "    data = []\n",
    "    for result in results:\n",
    "        data.append({\n",
    "            \"name\": result.find(\"name\").text if result.find(\"name\") is not None else None,\n",
    "            \"type\": result.find(\"type\").text if result.find(\"type\") is not None else None,\n",
    "            # Add all relevant fields based on the XML tags\n",
    "        })\n",
    "\n",
    "    print(\"Formatted Data for Pipeline:\", data)\n",
    "else:\n",
    "    print(f\"Failed to fetch data. HTTP Status Code: {response.status_code}\")\n",
    "    data = []  # Ensure data is at least an empty list to avoid further errors\n"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Formatted Data for Pipeline: []\n"
     ]
    }
   ],
   "execution_count": 18
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "D-3RP79kugz3"
   },
   "source": [
    "## 4 **Run the pipeline**"
   ]
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-12-21T16:18:03.800177Z",
     "start_time": "2024-12-21T16:18:03.107294Z"
    }
   },
   "cell_type": "code",
   "source": [
    "# Check data before running the pipeline\n",
    "print(\"Data Preview:\", data[:5])  # Show a sample of the first 5 records\n",
    "\n",
    "# Run the pipeline with normalized data\n",
    "load_info = pipeline.run(data, table_name='pokemon')\n",
    "print(load_info)"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data Preview: []\n",
      "Pipeline csd_pipeline completed in 0.69 seconds\n",
      "1 load package(s) were loaded to destination duckdb and into dataset csdb_releases\n",
      "The duckdb destination used duckdb:////Users/rolf/Documents/src/GitHub/dlt-test/csd_pipeline.duckdb location to store data\n",
      "Load package 1734797883.271083 is LOADED and contains no failed jobs\n"
     ]
    }
   ],
   "execution_count": 19
  },
  {
   "cell_type": "code",
   "metadata": {
    "colab": {
     "background_save": true
    },
    "id": "CoVzXSF5sInX",
    "outputId": "14ba6db2-1a51-4e50-cb39-1927d3024d7b",
    "ExecuteTime": {
     "end_time": "2024-12-21T16:18:23.203908Z",
     "start_time": "2024-12-21T16:18:23.174910Z"
    }
   },
   "source": [
    "# Normalize and load the data onto the locally created duckdb database 'pokemon_pipeline.duckdb'\n",
    "load_info = pipeline.run(data, table_name='pokemon')\n",
    "print(load_info)"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Pipeline csd_pipeline completed in 0.03 seconds\n",
      "0 load package(s) were loaded to destination duckdb and into dataset None\n",
      "The duckdb destination used duckdb:////Users/rolf/Documents/src/GitHub/dlt-test/csd_pipeline.duckdb location to store data\n"
     ]
    }
   ],
   "execution_count": 20
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4QBtnFC_3o6F"
   },
   "source": [
    "## 🎉 **Finish!** 🎉"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "DT2DTGZiyJ8g"
   },
   "source": [
    "## **Bonus: Query the loaded data** 🦆\n",
    "\n",
    "To access the loaded data, connect to the DuckDB database using the Python DuckDB connector"
   ]
  },
  {
   "metadata": {},
   "cell_type": "code",
   "outputs": [],
   "execution_count": null,
   "source": ""
  },
  {
   "cell_type": "code",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 173
    },
    "id": "ki1Fjk6_sTd5",
    "outputId": "e4a92697-0e32-4380-8901-1ef6b3f105a5",
    "ExecuteTime": {
     "end_time": "2024-12-21T16:18:27.434005Z",
     "start_time": "2024-12-21T16:18:27.302345Z"
    }
   },
   "source": [
    "import duckdb\n",
    "import polars as pl  # Import Polars for DataFrame manipulations\n",
    "\n",
    "# Connect to the DuckDB database\n",
    "conn = duckdb.connect(f\"{pipeline.pipeline_name}.duckdb\")\n",
    "\n",
    "# Query and convert result to a Polars DataFrame\n",
    "describe_result = pl.DataFrame(\n",
    "    conn.sql(\"DESCRIBE\").fetchall(),\n",
    "    schema=conn.sql(\"DESCRIBE\").columns,\n",
    "    orient=\"row\"  # Specify row orientation explicitly\n",
    ")\n",
    "\n",
    "# Display the Polars DataFrame\n",
    "print(describe_result)\n"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "shape: (3, 6)\n",
      "┌──────────────┬───────────────┬────────────────────┬───────────────────┬──────────────┬───────────┐\n",
      "│ database     ┆ schema        ┆ name               ┆ column_names      ┆ column_types ┆ temporary │\n",
      "│ ---          ┆ ---           ┆ ---                ┆ ---               ┆ ---          ┆ ---       │\n",
      "│ str          ┆ str           ┆ str                ┆ list[str]         ┆ list[str]    ┆ bool      │\n",
      "╞══════════════╪═══════════════╪════════════════════╪═══════════════════╪══════════════╪═══════════╡\n",
      "│ csd_pipeline ┆ csdb_releases ┆ _dlt_loads         ┆ [\"load_id\",       ┆ [\"VARCHAR\",  ┆ false     │\n",
      "│              ┆               ┆                    ┆ \"schema_name\", …  ┆ \"VARCHAR\", … ┆           │\n",
      "│              ┆               ┆                    ┆ \"…                ┆ \"VARC…       ┆           │\n",
      "│ csd_pipeline ┆ csdb_releases ┆ _dlt_pipeline_stat ┆ [\"version\",       ┆ [\"BIGINT\",   ┆ false     │\n",
      "│              ┆               ┆ e                  ┆ \"engine_version\", ┆ \"BIGINT\", …  ┆           │\n",
      "│              ┆               ┆                    ┆ …                 ┆ \"VARCHA…     ┆           │\n",
      "│ csd_pipeline ┆ csdb_releases ┆ _dlt_version       ┆ [\"version\",       ┆ [\"BIGINT\",   ┆ false     │\n",
      "│              ┆               ┆                    ┆ \"engine_version\", ┆ \"BIGINT\", …  ┆           │\n",
      "│              ┆               ┆                    ┆ …                 ┆ \"VARCHA…     ┆           │\n",
      "└──────────────┴───────────────┴────────────────────┴───────────────────┴──────────────┴───────────┘\n"
     ]
    }
   ],
   "execution_count": 21
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2024-12-21T16:18:35.444801Z",
     "start_time": "2024-12-21T16:18:35.431811Z"
    }
   },
   "cell_type": "code",
   "source": [
    "import duckdb\n",
    "import polars as pl\n",
    "\n",
    "# Example in-memory data\n",
    "pokemon_data = [(1, \"bulbasaur\"), (2, \"ivysaur\"), (3, \"venusaur\")]\n",
    "\n",
    "# Connect to DuckDB (in-memory)\n",
    "conn = duckdb.connect()\n",
    "\n",
    "# Step 1: Create the table schema\n",
    "conn.execute(\"CREATE TABLE pokemon_table (id INT, name VARCHAR)\")\n",
    "\n",
    "# Step 2: Insert data using prepared statements\n",
    "for record in pokemon_data:\n",
    "    conn.execute(\"INSERT INTO pokemon_table VALUES (?, ?)\", record)\n",
    "\n",
    "# Step 3: Query data directly and load into a Polars DataFrame\n",
    "result = pl.DataFrame(\n",
    "    conn.execute(\"SELECT * FROM pokemon_table\").fetchall(),\n",
    "    schema=[\"id\", \"name\"],\n",
    "    orient=\"row\",  # Explicitly specify row orientation\n",
    ")\n",
    "\n",
    "print(result)\n"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "shape: (3, 2)\n",
      "┌─────┬───────────┐\n",
      "│ id  ┆ name      │\n",
      "│ --- ┆ ---       │\n",
      "│ i64 ┆ str       │\n",
      "╞═════╪═══════════╡\n",
      "│ 1   ┆ bulbasaur │\n",
      "│ 2   ┆ ivysaur   │\n",
      "│ 3   ┆ venusaur  │\n",
      "└─────┴───────────┘\n"
     ]
    }
   ],
   "execution_count": 22
  },
  {
   "cell_type": "code",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 664
    },
    "id": "zRM6PrbY7y9m",
    "outputId": "e9ebbd5a-1e39-4ac7-dabf-179eca021e77",
    "ExecuteTime": {
     "end_time": "2024-12-21T15:24:29.041274Z",
     "start_time": "2024-12-21T15:24:29.026340Z"
    }
   },
   "source": [
    "# this lets us query data without adding schema prefix to table names\n",
    "conn.sql(f\"SET search_path = '{pipeline.dataset_name}'\")\n",
    "\n",
    "# Retrieve SQL query results and convert them directly into a Polars DataFrame\n",
    "stats_table = conn.sql(\"SELECT * FROM pokemon\").pl()  # .pl() assumes Polars compatibility here\n",
    "\n",
    "# Display the Polars DataFrame\n",
    "print(stats_table)\n"
   ],
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "shape: (20, 4)\n",
      "┌────────────┬─────────────────────────────────┬───────────────────┬────────────────┐\n",
      "│ name       ┆ url                             ┆ _dlt_load_id      ┆ _dlt_id        │\n",
      "│ ---        ┆ ---                             ┆ ---               ┆ ---            │\n",
      "│ str        ┆ str                             ┆ str               ┆ str            │\n",
      "╞════════════╪═════════════════════════════════╪═══════════════════╪════════════════╡\n",
      "│ bulbasaur  ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ bAvKtRkBARAxwg │\n",
      "│ ivysaur    ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ CvzTJVbcInbpuA │\n",
      "│ venusaur   ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ rwJZT1m/2u19/g │\n",
      "│ charmander ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ ei8/E+h2co+Wkg │\n",
      "│ charmeleon ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ Bbfe9TBXvs0HAA │\n",
      "│ …          ┆ …                               ┆ …                 ┆ …              │\n",
      "│ pidgey     ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ 3BbZfT1Bg5eyww │\n",
      "│ pidgeotto  ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ XLRn3hHCihmwEw │\n",
      "│ pidgeot    ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ hw6MMP2S59+Tiw │\n",
      "│ rattata    ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ fP6qwbE3Xn4Uwg │\n",
      "│ raticate   ┆ https://pokeapi.co/api/v2/poke… ┆ 1734793981.588002 ┆ NcqA4+s87bUZrA │\n",
      "└────────────┴─────────────────────────────────┴───────────────────┴────────────────┘\n"
     ]
    }
   ],
   "execution_count": 24
  },
  {
   "cell_type": "code",
   "metadata": {
    "id": "4oxlpJJC8vCc",
    "ExecuteTime": {
     "end_time": "2024-12-21T15:26:04.398412Z",
     "start_time": "2024-12-21T15:26:04.383644Z"
    }
   },
   "source": "conn.close()\n",
   "outputs": [],
   "execution_count": 25
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ol13bwL3yENc"
   },
   "source": [
    "# **Next Steps** 👀"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "nNIRk_NM1Mf6"
   },
   "source": [
    "## **[Give `dlt` a ⭐ on GitHub](https://github.com/dlt-hub/dlt)**\n",
    "## **[Join the `dlt` community on Slack](https://join.slack.com/t/dlthub-community/shared_invite/zt-1slox199h-HAE7EQoXmstkP_bTqal65g)**\n",
    "## **[Read the `dlt` docs](https://dlthub.com/docs/)**"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "name": "python3",
   "language": "python"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}