ETL_titanic_dataset.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qhSf9K5iGLrt"
      },
      "source": [
        "**Story:**\n",
        "\n",
        "Let us simulate the following case:\n",
        "I am a Data (ETL) Engineer and my colleague from Data Science department asks me to work with some data, they provide me with the data source, transformation rules and asks to load processed data into another database for further processing.\n",
        "Then my colleague creates a ticket in Jira and assigns it to me.\n",
        "I am starting...\n",
        "\n",
        "*Another variation of the above case*: Data Scientist can perform functions of Data Engineer as well.\n",
        "\n",
        "**Use Case:**\n",
        "\n",
        "*Data (ETL) Engineer:*\n",
        "1.  As source in this particular case uses sklearn.datasets library in Python;\n",
        "2.  Works with Titanic dataset. This training set is widely used in education purposes;\n",
        "3.  Performs transformation of data according to rules provided by Data Scientist;\n",
        "4.  Prepares transformed data for the pipeline;\n",
        "5.  Creates a pipeline;\n",
        "6.  Runs it;\n",
        "7.  Checks DuckDB as a destination: load is successful;\n",
        "7.  Shares streamlit link with Data Scientist for further exploration;\n",
        "8.  Monitors Pipeline's health\n",
        "\n",
        "*Data Scientist:*\n",
        "1. Defines source and destination;\n",
        "2. Creates a set of transformation rules: convert categoricals to strings, remove certain columns, removes records with 0 values, change types of columns\n",
        "2. Works with the loaded data via shared Streamlit link or directly in DuckDB\n",
        "\n",
        "For the case I use dlt library\n",
        "\n",
        "\n",
        "Now let us move to practice\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7YlSmVrbL6la"
      },
      "source": [
        "Let us install the following libraries:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "R5onp75jMzhX",
        "outputId": "59f153b4-9af9-4f1d-99a8-7f0b9ca781ea"
      },
      "outputs": [],
      "source": [
        "!pip install dlt\n",
        "!pip install streamlit\n",
        "!pip install duckdb\n",
        "!pip install scikit-learn\n",
        "!pip install numpy pandas\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AwIl6EGnTNTJ"
      },
      "source": [
        "**In case one needs to UNINSTALL duckdb run the cell below. Otherwise skip!**\n",
        "\n",
        "Just in case how to  uninstall the package, it might be needed during the below path:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "1yaJoDCKTI1U"
      },
      "outputs": [],
      "source": [
        "!pip uninstall -y duckdb"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_AXcCPYTNKsg"
      },
      "source": [
        "Let us import all the libraries we need:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "c7WOpgh0L-fu"
      },
      "outputs": [],
      "source": [
        "import pandas as pd # we use it for data pre-processing\n",
        "from sklearn.datasets import fetch_openml # using fetch_openml we download titanic data set from OpenML repository\n",
        "import dlt # data load tool\n",
        "import duckdb # destination, where we load our pre-processed data\n",
        "import streamlit # this is UI framework to have an access to results in DuckDB"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "G3MpkgzzQMwt"
      },
      "source": [
        "Fetch dataset and make accessible for further operations:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "BRehLxz9QPO9",
        "outputId": "52882697-5eef-410c-b270-8372949cfe7a"
      },
      "outputs": [],
      "source": [
        "titanic = fetch_openml('titanic', version=1, as_frame=True) # let us fix the version of dataset. So the data is constant all the way down.\n",
        "titanic_data = titanic.frame # now it is pandas dataframe"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "J00kvVXwRsKs"
      },
      "source": [
        "Let us implement pre-processing with means of pandas library. With this step we prepare data according to request of Data Scientist.\n",
        "\n",
        "At this stage also Feature Engineering could be applied.\n",
        "\n",
        "Let us declare the source, for the case we use decorators and yield operator:\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "qjXWydMgTHHR"
      },
      "outputs": [],
      "source": [
        "@dlt.resource #declaring resource, as I understand this decorator wraps up output as an input for the pipeline, at least it works like this without exceptions\n",
        "def transform_data_with_pandas(titanic_data):\n",
        "\n",
        "    # Here we do a few tricks on data: converting categorical columns to strings\n",
        "    titanic_data['survived'] = titanic_data['survived'].astype(str)\n",
        "    titanic_data['sex'] = titanic_data['sex'].astype(str)\n",
        "\n",
        "    # Below we define columns to removed from the dataframe, we consider them non meaningful, just because I remember this data, but it also can be proved with scietific means\n",
        "    columns_to_drop = ['embarked', 'parch', 'sibsp', 'ticket', 'boat', 'body', 'home.dest', 'name', 'cabin']\n",
        "    titanic_data.drop(columns=columns_to_drop, inplace=True)\n",
        "\n",
        "    # Here we convert age and fare into int64 for accuracy purpose, before it had double datatype and looked not friendly to human\n",
        "    titanic_data['age'] = titanic_data['age'].fillna(0).astype('int64')\n",
        "    titanic_data['fare'] = titanic_data['fare'].fillna(0).astype('int64')\n",
        "\n",
        "    #This construction is required for resource declare, to feed it into pipeline it has to be a list of dictionaries.\n",
        "    for record in titanic_data.to_dict(orient='records'):\n",
        "        yield record\n",
        "\n",
        "    #Initially I had this code instead yield with no return and it also worked:\n",
        "    #titanic_list = titanic_data.to_dict(orient='records')\n",
        "    #return titanic_list\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "uXM49XLrtWXY"
      },
      "source": [
        "Create and run the pipeline. At this moment schema qiuck_start and table passengers in duckDB are created, data is loaded:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "VdyXTtxZtaBM",
        "outputId": "9cb197b3-b515-44b7-e042-427d3c0b2042"
      },
      "outputs": [],
      "source": [
        "# Create. I used name from example in docs as well as the code itself is taken from docs\n",
        "pipeline = dlt.pipeline(\n",
        "    pipeline_name=\"quick_start\", destination=\"duckdb\", dataset_name=\"titanic\"\n",
        ")\n",
        "#load happens here. I use 'replace' which re-writes the whole dataset. Dataset is small compare to real-life, so it is ok.\n",
        "load_info = pipeline.run(transform_data_with_pandas(titanic_data), table_name=\"passengers\", write_disposition='replace')\n",
        "print(load_info)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gAyXCcLPuB27"
      },
      "source": [
        "Present result in streamlit framework. It is a cool feature!"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Vf9hea6wuMlf",
        "outputId": "befc4d01-3c7e-4e2e-af92-65bceeb145af"
      },
      "outputs": [],
      "source": [
        "# Show the pipeline status. Here you can interact with duckDB via SQL queries, check he content of loaded data and other load info.\n",
        "!dlt pipeline quick_start show"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "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.5"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}