Tutorial_ETL_with_soft_dependencies.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "61b2ee94-6df7-48fb-ae5e-da6f8f604e69",
   "metadata": {},
   "source": [
    "# ETL with soft dependencies \n",
    "\n",
    "this notebook tries to replicate a real world application:\n",
    "\n",
    "1. We update the new data (i this case a \"new\") pokemon\n",
    "2. We run some queries in paralallel because they are not co-dependent\n",
    "3. We run a final query after we have all the sources refreshed\n",
    "\n",
    "## First we can create the funtions\n",
    "\n",
    "1. Update info from a pokemon id\n",
    "2. Check ids and update a new pokemon\n",
    "3. Some random rankings\n",
    "4. A final ranking that uses the results from (3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "3565c58b-306f-4e8c-8d7c-b7ceb56f7383",
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "import pandas as pd\n",
    "import os\n",
    "from google.cloud import bigquery\n",
    "\n",
    "client = bigquery.Client()\n",
    "\n",
    "\n",
    "i = 200\n",
    "\n",
    "def get_pokemon_data(i):\n",
    "    url = f'https://pokeapi.co/api/v2/pokemon-species/{i}'\n",
    "    x = requests.get(url)\n",
    "    aux = x.json()['flavor_text_entries']\n",
    "    \n",
    "    aux = x.json()['flavor_text_entries']\n",
    "    text = [x['flavor_text'].replace('\\n',' ') for x in aux]\n",
    "    lang = [x['language']['name'] for x in aux]\n",
    "    version = [x['version']['name'] for x in aux]\n",
    "    df = pd.DataFrame({'text':text,'lang':lang,'version':version}).query('lang==\"en\"')\n",
    "    df['name'] = x.json()['name']\n",
    "    df['pokemon_id'] = i \n",
    "    \n",
    "    pokedex = df[['pokemon_id','text','lang','version']]\n",
    "    names = df[['pokemon_id','name']].drop_duplicates()\n",
    "    \n",
    "    url = f'https://pokeapi.co/api/v2/pokemon/{i}'\n",
    "    x = requests.get(url)\n",
    "    \n",
    "    stats = pd.DataFrame([(x ['base_stat'], x['stat']['name'])for  x in x.json()['stats']])\n",
    "    stats= stats.set_index([1]).transpose()\n",
    "    stats['pokemon_id'] = i\n",
    "    stats.columns = ['hp','attack','defense','specialAttack','specialDefense','speed','pokemon_id']\n",
    "    \n",
    "    \n",
    "    sprites = pd.DataFrame((x.json()['sprites']['front_default'], i+1)).transpose()\n",
    "    \n",
    "    sprites.columns = ['sprite_link','pokemon_id']\n",
    "    \n",
    "    sprites = sprites[['pokemon_id','sprite_link']]\n",
    "\n",
    "    return names, pokedex, stats, sprites\n",
    "\n",
    "def get_games_order(game):\n",
    "    url = f'https://pokeapi.co/api/v2/version/{game}'\n",
    "    x = requests.get(url)\n",
    "    aux = x.json()\n",
    "    aux = (aux['id'], aux['name'])\n",
    "    game_order = pd.DataFrame(aux).transpose()\n",
    "    game_order.columns = ['id','version']\n",
    "    return game_order\n",
    "\n",
    "def table_upload_append(client,df,table):\n",
    "    job_config = bigquery.LoadJobConfig(\n",
    "            write_disposition=\"WRITE_APPEND\",\n",
    "        )\n",
    "\n",
    "\n",
    "    job = client.load_table_from_dataframe(\n",
    "        df , table, job_config=job_config\n",
    "    )  # Make an API request.\n",
    "    job.result()  # Wait for the job to complete.\n",
    "\n",
    "\n",
    "def refresh_one_pokemon():\n",
    "    table = 'examples.names'\n",
    "\n",
    "    try:\n",
    "        sql = f'''\n",
    "        SELECT max(pokemon_id) i FROM {table}\n",
    "        '''\n",
    "        #print(sql)\n",
    "        \n",
    "        i = client.query(sql).result().to_dataframe().i.values[0] +1\n",
    "    except:\n",
    "        #print('no table?')\n",
    "        i = 1\n",
    "    #print(i)\n",
    "\n",
    "    names, pokedex, stats, sprites = get_pokemon_data(i)\n",
    "\n",
    "    table_upload_append(client,names,'examples.names')\n",
    "    table_upload_append(client,stats,'examples.stats')\n",
    "    table_upload_append(client,sprites,'examples.sprites')\n",
    "    table_upload_append(client,pokedex,'examples.pokedex')\n",
    "\n",
    "    sql = f'''\n",
    "        SELECT distinct version  FROM examples.pokedex\n",
    "        '''\n",
    "        \n",
    "    versions = client.query(sql).result().to_dataframe().version\n",
    "    \n",
    "    games = pd.concat([get_games_order(x) for x in versions] )\n",
    "\n",
    "    job_config = bigquery.LoadJobConfig(\n",
    "            write_disposition=\"WRITE_TRUNCATE\",\n",
    "        )\n",
    "\n",
    "\n",
    "    job = client.load_table_from_dataframe(\n",
    "        games , 'examples.games', job_config=job_config\n",
    "    )  # Make an API request.\n",
    "    job.result()  # Wait for the job to complete.\n",
    "\n",
    "    return names\n",
    "\n",
    "\n",
    "\n",
    "def set_top_all():\n",
    "    job_config = bigquery.QueryJobConfig(\n",
    "            write_disposition=\"WRITE_TRUNCATE\",\n",
    "            destination = f'''{os.environ['GCP_PROJECT']}.examples.top_all'''\n",
    "    )\n",
    "   \n",
    "    sql = '''SELECT  pokemon_id, name, \n",
    "    row_number() over(order by attack + `specialAttack`+ defense + `specialDefense` + speed desc) ranking\n",
    "    FROM examples.stats \n",
    "    left join examples.names using(pokemon_id)\n",
    "    order by attack + `specialAttack`+ defense + `specialDefense` + speed desc\n",
    "    limit 50\n",
    "    '''\n",
    "\n",
    "    # Start the query, passing in the extra configuration.\n",
    "    query_job = client.query(\n",
    "        sql,\n",
    "        # Location must match that of the dataset(s) referenced in the query\n",
    "        # and of the destination table.\n",
    "        location=\"US\",\n",
    "        job_config=job_config,\n",
    "    )  # API request - starts the query\n",
    "\n",
    "    query_job.result()\n",
    "\n",
    "\n",
    "def set_top_offensive():\n",
    "    job_config = bigquery.QueryJobConfig(\n",
    "            write_disposition=\"WRITE_TRUNCATE\",\n",
    "            destination = f'''{os.environ['GCP_PROJECT']}.examples.top_offense'''\n",
    "    )\n",
    "\n",
    "    sql = '''SELECT  pokemon_id, name, row_number() over(order by attack + `specialAttack` desc) ranking\n",
    "    FROM examples.stats \n",
    "    left join examples.names using(pokemon_id)\n",
    "    order by defense + attack + `specialAttack` desc\n",
    "    limit 50\n",
    "    '''\n",
    "\n",
    "    # Start the query, passing in the extra configuration.\n",
    "    query_job = client.query(\n",
    "        sql,\n",
    "        # Location must match that of the dataset(s) referenced in the query\n",
    "        # and of the destination table.\n",
    "        location=\"US\",\n",
    "        job_config=job_config,\n",
    "    )  # API request - starts the query\n",
    "\n",
    "    query_job.result()\n",
    "\n",
    "\n",
    "def set_top_defensive():\n",
    "    job_config = bigquery.QueryJobConfig(\n",
    "            write_disposition=\"WRITE_TRUNCATE\",\n",
    "            destination = f'''{os.environ['GCP_PROJECT']}.examples.top_defensive'''\n",
    "    )\n",
    "\n",
    "    sql = '''SELECT  pokemon_id, name, row_number() over(order by defense + `specialDefense` desc) ranking\n",
    "    FROM examples.stats \n",
    "    left join examples.names using(pokemon_id)\n",
    "    order by defense + `specialDefense` desc\n",
    "    limit 50\n",
    "    '''\n",
    "\n",
    "    # Start the query, passing in the extra configuration.\n",
    "    query_job = client.query(\n",
    "        sql,\n",
    "        # Location must match that of the dataset(s) referenced in the query\n",
    "        # and of the destination table.\n",
    "        location=\"US\",\n",
    "        job_config=job_config,\n",
    "    )  # API request - starts the query\n",
    "\n",
    "    query_job.result()\n",
    "\n",
    "\n",
    "def set_top_fast():\n",
    "    job_config = bigquery.QueryJobConfig(\n",
    "            write_disposition=\"WRITE_TRUNCATE\",\n",
    "            destination = f'''{os.environ['GCP_PROJECT']}.examples.top_fast'''\n",
    "    )\n",
    "\n",
    "    sql = '''SELECT  pokemon_id, name, row_number() over(order by speed desc) ranking\n",
    "    FROM examples.stats \n",
    "    left join examples.names using(pokemon_id)\n",
    "    order by speed desc\n",
    "    limit 50\n",
    "    '''\n",
    "\n",
    "    # Start the query, passing in the extra configuration.\n",
    "    query_job = client.query(\n",
    "        sql,\n",
    "        # Location must match that of the dataset(s) referenced in the query\n",
    "        # and of the destination table.\n",
    "        location=\"US\",\n",
    "        job_config=job_config,\n",
    "    )  # API request - starts the query\n",
    "\n",
    "    query_job.result()\n",
    "\n",
    "\n",
    "def final_ranking():\n",
    "\n",
    "    sql = '''\n",
    "    SELECT pokemon_id, names.name, row_number() over(order by  t1.ranking + t2.ranking + t3.ranking + t4.ranking) ranking_agg\n",
    "    FROM `examples.top_all` t1\n",
    "    inner join `examples.top_defensive`t2 using(pokemon_id)\n",
    "    inner join `examples.top_offense`t3 using(pokemon_id)\n",
    "    inner join `examples.top_fast`t4 using(pokemon_id)\n",
    "    inner join `examples.names` names using(pokemon_id)\n",
    "    order by t1.ranking + t2.ranking + t3.ranking + t4.ranking    \n",
    "    '''\n",
    "\n",
    "    job_config = bigquery.QueryJobConfig(\n",
    "            write_disposition=\"WRITE_TRUNCATE\",\n",
    "            destination = f'''{os.environ['GCP_PROJECT']}.examples.top_final'''\n",
    "    )\n",
    "\n",
    "    # Start the query, passing in the extra configuration.\n",
    "    query_job = client.query(\n",
    "        sql,\n",
    "        # Location must match that of the dataset(s) referenced in the query\n",
    "        # and of the destination table.\n",
    "        location=\"US\",\n",
    "        job_config=job_config,\n",
    "    )  # API request - starts the query\n",
    "\n",
    "    query_job.result()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4fc8299c-cdaf-48cf-b39c-4bcc6546b6fd",
   "metadata": {},
   "source": [
    "The steps to build and run are the following:\n",
    "\n",
    "1. The common task before defining a dag is defining the function that we will run in each node (we did this in the previous chunk)\n",
    "2. Creates nodes (please check that we did not add the dependency directly in here in this example)\n",
    "3. Define dependencies using > (thats the Soft dependency operator)\n",
    "4. Create the nodes list using all the ExecNodes availables in the envioronment... if you do not want to do it with all the created nodes please create the list by yourself as usual\n",
    "5. Create the dag with the list of nodes\n",
    "6. Run the dag\n",
    "7. Check the html output with one iframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "5c4ca12e-874e-4630-8a91-68965d8e4945",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023-06-15 11:10:09.721 | INFO     | easydags.node:execute:146 - Start executing add at 2023-06-15, 11:10:09\n",
      "2023-06-15 11:10:41.848 | INFO     | easydags.node:execute:146 - Start executing offensive at 2023-06-15, 11:10:41\n",
      "2023-06-15 11:10:41.849 | INFO     | easydags.node:execute:146 - Start executing all at 2023-06-15, 11:10:41\n",
      "2023-06-15 11:10:41.851 | INFO     | easydags.node:execute:146 - Start executing fast at 2023-06-15, 11:10:41\n",
      "2023-06-15 11:10:44.617 | INFO     | easydags.node:execute:146 - Start executing defensive at 2023-06-15, 11:10:44\n",
      "2023-06-15 11:10:47.837 | INFO     | easydags.node:execute:146 - Start executing final at 2023-06-15, 11:10:47\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "drawing\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "\n",
       "        <iframe\n",
       "            width=\"100%\"\n",
       "            height=\"600\"\n",
       "            src=\"Real ETL example (using pokemon)_states_run.html\"\n",
       "            frameborder=\"0\"\n",
       "            allowfullscreen\n",
       "            \n",
       "        ></iframe>\n",
       "        "
      ],
      "text/plain": [
       "<IPython.lib.display.IFrame at 0x7f9898570e80>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from easydags import  ExecNode, DAG\n",
    "\n",
    "\n",
    "\n",
    "add_pokemon = ExecNode('add', exec_function = refresh_one_pokemon)\n",
    "\n",
    "\n",
    "ranking_all = ExecNode('all', exec_function = set_top_all)\n",
    "\n",
    "ranking_offensive = ExecNode('offensive', exec_function = set_top_offensive)\n",
    "\n",
    "ranking_defensive = ExecNode('defensive', exec_function = set_top_defensive)\n",
    "\n",
    "ranking_fast = ExecNode('fast', exec_function = set_top_fast)\n",
    "\n",
    "ranking_final = ExecNode('final', exec_function = final_ranking)\n",
    "\n",
    "\n",
    "\n",
    "add_pokemon > ranking_all > ranking_final\n",
    "\n",
    "add_pokemon > ranking_offensive > ranking_final\n",
    "add_pokemon > ranking_defensive > ranking_final\n",
    "add_pokemon > ranking_fast > ranking_final\n",
    "\n",
    "\n",
    "\n",
    "# Creating the list of nodes... you can also do it by yourself! \n",
    "#nodes = [node_prepro, node_base,node_under,node_over,node_smote,node_metrics] \n",
    "nodes = [] \n",
    "globs = globals().copy()\n",
    "for obj_name in globs:         \n",
    "    if isinstance(globs[obj_name], ExecNode):\n",
    "        nodes.append(globs[obj_name])\n",
    "\n",
    "\n",
    "dag = DAG(nodes,name = 'Real ETL example (using pokemon)',max_concurrency=3, debug = False, error_type_fatal= False)\n",
    "\n",
    "dag.execute()\n",
    "    \n",
    "from IPython.display import IFrame\n",
    "IFrame(src=f\"{dag.name}_states_run.html\", width='100%', height=600)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bda2cb47-5489-4909-aff9-e8880709abb1",
   "metadata": {},
   "source": [
    "# Final DAG\n",
    "\n",
    "If you run this tutorial you will get the dag html by yourself, here i will add a png version so you can check it out without running the tutorial:\n",
    "\n",
    "![Motivation](https://raw.githubusercontent.com/magralo/easydags/main/resource_readme/dag_tut_data_eng.png)\n",
    "              \n"
   ]
  }
 ],
 "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.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}