00-intro/workshop.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div align=\"center\">\n",
    "  <img src=\"https://raw.githubusercontent.com/flyteorg/static-resources/main/flyte/readme/flyte_and_lf.png\" alt=\"Flyte and LF AI & Data Logo\" width=\"250\">\n",
    "</div>\n",
    "\n",
    "<h1 align=\"center\">\n",
    "  Flyte School\n",
    "</h1>\n",
    "\n",
    "<h3 align=\"center\">\n",
    "  Module 0: A Practical Introduction to Machine Learning Orchestration\n",
    "</h3>\n",
    "\n",
    "This module provides a hands-on introduction to Flyte, an open source machine\n",
    "learning and data processing orchestration platform. We'll cover the basics of\n",
    "Flyte and how to use it locally for iterative development.\n",
    "\n",
    "### Learning Objectives\n",
    "\n",
    "- 🧱 Understand how tasks, workflows, and launchplans can be used to manage the\n",
    "  flow of data and models through a sequence of computations.\n",
    "- 💻 Learn how to spin up a local Flyte cluster on their own workstation and run\n",
    "  workflows on it for local debugging.\n",
    "- 🔄 Programmatically run tasks and workflows through the CLI or in a Python\n",
    "  runtime to quickly iterate on your code.\n",
    "\n",
    "### Outline\n",
    "\n",
    "| 🔤 Introduction to Flyte | [45 minutes] |\n",
    "| --- | --- |\n",
    "| **Environment Setup** | Setting up your virtual development environment |\n",
    "| **Flyte Basics** | Tasks, Workflows, and Launch Plans: the building blocks of Flyte |\n",
    "| **`pyflyte run`** | Run tasks and workflows locally or on a Flyte cluster  |\n",
    "| **Flyte Console** | A tour of the Flyte console to view workflow progress and status  |\n",
    "| **`FlyteRemote`** | Programmatically run tasks and workflows  |\n",
    "| **Scheduling Launch Plans** | Run your workflows on a schedule |\n",
    "| **Flyte Programming Model** | A mental model of Flyte |"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 🌍 Environment Setup\n",
    "\n",
    "Follow the instructions in the setup instructions of the [README](./README.md).\n",
    "\n",
    "We'll be using some environment variables throughout this workshop, so let's\n",
    "export them right now:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from pathlib import Path\n",
    "\n",
    "os.environ[\"FLYTECTL_CONFIG\"] = str(Path.home() / \".flyte/config-sandbox.yaml\")\n",
    "os.environ[\"IMAGE\"] = \"ghcr.io/unionai-oss/flyte-school:00-intro-latest\""
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 🔤 Flyte Basics\n",
    "\n",
    "Tasks, Workflows, and Launch Plans: the building blocks of Flyte."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 📦 Tasks as Containerized Functions\n",
    "\n",
    "A core building block for type-safety, statelessness, and reproducibility."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's take a closer look at Flyte tasks.\n",
    "\n",
    "<image src=\"static/flyte_tasks.png\" width=\"400px\">\n",
    "\n",
    "We're going to use `flytekit` to write our tasks.\n",
    "\n",
    "`flytekit` is the Python SDK for Flyte. It's the way data scientists, ML engineers, data engineers, and data analysts write code that will eventually run on a Flyte cluster.\n",
    "\n",
    "The `@task`-decorated function looks deceptively simple:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello world\n"
     ]
    }
   ],
   "source": [
    "from flytekit import task\n",
    "\n",
    "@task\n",
    "def hello():\n",
    "    print(\"hello world\")\n",
    "\n",
    "hello()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we want the task to do anything useful, we want to give it some inputs and\n",
    "have it produce some outputs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.0"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@task\n",
    "def square(x: float) -> float:\n",
    "    return x ** 2\n",
    "\n",
    "square(x=2.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> ℹ️ In the above code, notice two things:\n",
    ">\n",
    "> - **Tasks functions are strongly typed:** this not only provides type safety,\n",
    ">   it allows Flyte to analyze a sequence of tasks that depend on each other\n",
    ">   and determine their compatibility.\n",
    "> - **Tasks arguments must be kwargs:** this is a current constraint of Flyte tasks\n",
    ">   that may be removed in a future release."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**📝 Exercise**:\n",
    "1. Try calling `square` with a different data type.\n",
    "2. Modify the code in the `square` function to output a different data type."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 🤔 Why Containerized Functions?\n",
    "\n",
    "What does type-safety have to do with containers?\n",
    "\n",
    "Flyte treats tasks as containerized functions. This means that each task runs on\n",
    "their own isolated container in the Flyte cluster. Coupled with strongly-typed\n",
    "interfaces, tasks are essentially ✨microservices✨ that can be composed together\n",
    "to form workflows.\n",
    "\n",
    "This entails the following benefits:\n",
    "\n",
    "- **Statelessness:** each task is stateless, which means that it can be run\n",
    "  multiple times without side effects.\n",
    "- **Reproducibility:** each task is reproducible, which means that it can be\n",
    "  run multiple times with the same inputs and produce the same outputs.\n",
    "- **Portability:** each task is portable, which means that it can be run on\n",
    "  any Flyte cluster, or any infrastructure that can run containers.\n",
    "- **Heterogeneity:** each task in a workflow can be written in any language and\n",
    "  can be run with varying compute requirements."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 🔀 Workflows and Promises\n",
    "\n",
    "How Flyte workflows construct an execution graph of tasks."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's take a closer look at Flyte workflows:\n",
    "\n",
    "<image src=\"static/flyte_workflows.png\" width=\"500px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Flyte `@workflow` is basically a domain-specific language (DSL) that builds an\n",
    "execution graph that uses tasks as the building blocks for more complex pipelines."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5.0"
      ]
     },
     "execution_count": 71,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from flytekit import workflow\n",
    "\n",
    "@task\n",
    "def error(x: list[float], y: list[float]) -> list[float]:\n",
    "    return [xi - yi for xi, yi in zip(x, y)]\n",
    "\n",
    "@task\n",
    "def squares(x: list[float]) -> list[float]:\n",
    "    out = [xi ** 2 for xi in x]\n",
    "    return out\n",
    "\n",
    "@task\n",
    "def sum_task(x: list[float]) -> float:\n",
    "    return sum(x)\n",
    "\n",
    "@workflow\n",
    "def sum_of_squares(x: list[float], y: list[float]) -> float:\n",
    "    errors = error(x=x, y=y)\n",
    "    squared = squares(x=errors)\n",
    "    return sum_task(x=squared)\n",
    "\n",
    "\n",
    "sum_of_squares(x=[1.0, 2.0, 4.0], y=[1.0, 3.0, 6.0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**📝 Exercise:**\n",
    "\n",
    "- In the `sum_of_squares` workflow function, print out the `squared` variable. What\n",
    "  do you expect to see? What do you actually see?\n",
    "- Try invoking `sum_of_squares` with different data types.\n",
    "- Modify one of the types in any of the tasks used in the `sum_of_squares` workflow."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's take a look at the [workflows/example_intro.py](./workflows/example_intro.py) script.\n",
    "\n",
    "In it, you'll see a simple pipeline that uses the penguins dataset to train a\n",
    "penguin species classifier. This script introduces three core concepts in Flyte:\n",
    "\n",
    "- `tasks`: the basic unit of compute in Flyte.\n",
    "- `workflows`: an execution graph of tasks.\n",
    "- `launchplans`: a mechanism for executing and reusing workflows.\n",
    "\n",
    "You can run this workflow locally with:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DefaultNamedTupleOutput(o0=LogisticRegression(C=0.1, max_iter=5000), o1=0.989010989010989, o2=1.0)\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "python workflows/example_intro.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Iterating on Workflows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `pyflyte run`\n",
    "\n",
    "Run tasks and workflows locally or on a Flyte cluster."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run to run a workflow locally, execute the following command on on your terminal:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DefaultNamedTupleOutput(o0=LogisticRegression(C=0.01, max_iter=2500), o1=0.9743589743589743, o2=1.0)\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte run \\\n",
    "    workflows/example_intro.py training_workflow \\\n",
    "    --hyperparameters '{\"C\": 0.01}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is great for the local debugging experience, but what if we want to run this\n",
    "workflow on an actual Flyte cluster?\n",
    "\n",
    "`pyflyte run` also supports this use case through the `--remote` flag."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Promise(node:n2.o0)\n",
      "Go to http://localhost:30080/console/projects/flytesnacks/domains/development/executions/f5f8d11a0f59046d9aa5 to see execution in the console.\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte --config $FLYTECTL_CONFIG \\\n",
    "    run --remote \\\n",
    "    --image $IMAGE \\\n",
    "    workflows/example_intro.py training_workflow \\\n",
    "    --hyperparameters '{\"C\": 0.01}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Notice how we're providing two extra flags:\n",
    "- `--config`: this is the path to the Flyte config file, which points `pyflyte run`\n",
    "  to the Flyte cluster endpoint.\n",
    "- `--remote`: this flag tells `pyflyte run` that we want to run the workflow on\n",
    "  a flyte cluster.\n",
    "\n",
    "Once you execute this command, you should see a message that looks like this:\n",
    "\n",
    "```\n",
    "Go to http://localhost:30080/console/projects/flytesnacks/domains/development/executions/<execution_id> to see execution in the console.\n",
    "```\n",
    "\n",
    "Where `<execution_id>` is the execution id of the workflow\n",
    "execution."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**📝 Exercise:**\n",
    "\n",
    "- Run a task with `pyflyte run` locally."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 🖥️ Flyte Console\n",
    "\n",
    "A tour of the Flyte console to view workflow progress and status."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's go to the URL provided by the `pyflyte run` command to see the execution progressing.\n",
    "\n",
    "The `flyteconsole` is the UI component of the Flyte stack. It provides a way to visualize workflows, launch them from the browser, and obtain useful metadata about Flyte entities and their corresponding executions.\n",
    "\n",
    "<image src=\"https://raw.githubusercontent.com/flyteorg/static-resources/main/flytesnacks/getting_started/getting_started_console.gif\" width=\"1000px\">\n",
    "\n",
    "> ℹ️ There are a few things to note about the console:\n",
    ">\n",
    "> - It exposes comprehensive metadata about the workflow as it progresses.\n",
    "> - It visualizes the execution graph of the workflow in the **Graph View**.\n",
    "> - It profiles the runtime of the workflow in the **Timeline View**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 🎮 `FlyteRemote`\n",
    "\n",
    "Programmatically run tasks and workflows."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also run workflows programmatically using the `FlyteRemote` class. This\n",
    "is useful for:\n",
    "\n",
    "- 📓 Running Flyte tasks/workflows within a Jupyter notebook\n",
    "- 🤖 Running Flyte tasks/workflows as a microservice\n",
    "- 🚢 Integrating Flyte into your CI/CD pipelines, or \n",
    "\n",
    "The code below illustrates how we can import the workflow functions into a Python\n",
    "runtime and execute them on a Flyte cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'http://localhost:30080/console/projects/flytesnacks/domains/development/executions/f8627c6c5a4f44e40949'"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from workflows import example_intro\n",
    "from workflows.utils import get_remote\n",
    "\n",
    "remote = get_remote()\n",
    "execution = remote.execute_local_workflow(\n",
    "    example_intro.training_workflow,\n",
    "    inputs={\n",
    "        \"hyperparameters\": example_intro.Hyperparameters(C=0.1, max_iter=5000),\n",
    "        \"test_size\": 0.2,\n",
    "        \"random_state\": 11,\n",
    "    }\n",
    ")\n",
    "remote.generate_console_url(execution)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution = remote.wait(execution)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style>#sk-container-id-1 {color: black;}#sk-container-id-1 pre{padding: 0;}#sk-container-id-1 div.sk-toggleable {background-color: white;}#sk-container-id-1 label.sk-toggleable__label {cursor: pointer;display: block;width: 100%;margin-bottom: 0;padding: 0.3em;box-sizing: border-box;text-align: center;}#sk-container-id-1 label.sk-toggleable__label-arrow:before {content: \"▸\";float: left;margin-right: 0.25em;color: #696969;}#sk-container-id-1 label.sk-toggleable__label-arrow:hover:before {color: black;}#sk-container-id-1 div.sk-estimator:hover label.sk-toggleable__label-arrow:before {color: black;}#sk-container-id-1 div.sk-toggleable__content {max-height: 0;max-width: 0;overflow: hidden;text-align: left;background-color: #f0f8ff;}#sk-container-id-1 div.sk-toggleable__content pre {margin: 0.2em;color: black;border-radius: 0.25em;background-color: #f0f8ff;}#sk-container-id-1 input.sk-toggleable__control:checked~div.sk-toggleable__content {max-height: 200px;max-width: 100%;overflow: auto;}#sk-container-id-1 input.sk-toggleable__control:checked~label.sk-toggleable__label-arrow:before {content: \"▾\";}#sk-container-id-1 div.sk-estimator input.sk-toggleable__control:checked~label.sk-toggleable__label {background-color: #d4ebff;}#sk-container-id-1 div.sk-label input.sk-toggleable__control:checked~label.sk-toggleable__label {background-color: #d4ebff;}#sk-container-id-1 input.sk-hidden--visually {border: 0;clip: rect(1px 1px 1px 1px);clip: rect(1px, 1px, 1px, 1px);height: 1px;margin: -1px;overflow: hidden;padding: 0;position: absolute;width: 1px;}#sk-container-id-1 div.sk-estimator {font-family: monospace;background-color: #f0f8ff;border: 1px dotted black;border-radius: 0.25em;box-sizing: border-box;margin-bottom: 0.5em;}#sk-container-id-1 div.sk-estimator:hover {background-color: #d4ebff;}#sk-container-id-1 div.sk-parallel-item::after {content: \"\";width: 100%;border-bottom: 1px solid gray;flex-grow: 1;}#sk-container-id-1 div.sk-label:hover label.sk-toggleable__label {background-color: #d4ebff;}#sk-container-id-1 div.sk-serial::before {content: \"\";position: absolute;border-left: 1px solid gray;box-sizing: border-box;top: 0;bottom: 0;left: 50%;z-index: 0;}#sk-container-id-1 div.sk-serial {display: flex;flex-direction: column;align-items: center;background-color: white;padding-right: 0.2em;padding-left: 0.2em;position: relative;}#sk-container-id-1 div.sk-item {position: relative;z-index: 1;}#sk-container-id-1 div.sk-parallel {display: flex;align-items: stretch;justify-content: center;background-color: white;position: relative;}#sk-container-id-1 div.sk-item::before, #sk-container-id-1 div.sk-parallel-item::before {content: \"\";position: absolute;border-left: 1px solid gray;box-sizing: border-box;top: 0;bottom: 0;left: 50%;z-index: -1;}#sk-container-id-1 div.sk-parallel-item {display: flex;flex-direction: column;z-index: 1;position: relative;background-color: white;}#sk-container-id-1 div.sk-parallel-item:first-child::after {align-self: flex-end;width: 50%;}#sk-container-id-1 div.sk-parallel-item:last-child::after {align-self: flex-start;width: 50%;}#sk-container-id-1 div.sk-parallel-item:only-child::after {width: 0;}#sk-container-id-1 div.sk-dashed-wrapped {border: 1px dashed gray;margin: 0 0.4em 0.5em 0.4em;box-sizing: border-box;padding-bottom: 0.4em;background-color: white;}#sk-container-id-1 div.sk-label label {font-family: monospace;font-weight: bold;display: inline-block;line-height: 1.2em;}#sk-container-id-1 div.sk-label-container {text-align: center;}#sk-container-id-1 div.sk-container {/* jupyter's `normalize.less` sets `[hidden] { display: none; }` but bootstrap.min.css set `[hidden] { display: none !important; }` so we also need the `!important` here to be able to override the default hidden behavior on the sphinx rendered scikit-learn.org. See: https://github.com/scikit-learn/scikit-learn/issues/21755 */display: inline-block !important;position: relative;}#sk-container-id-1 div.sk-text-repr-fallback {display: none;}</style><div id=\"sk-container-id-1\" class=\"sk-top-container\"><div class=\"sk-text-repr-fallback\"><pre>LogisticRegression(C=0.1, max_iter=5000)</pre><b>In a Jupyter environment, please rerun this cell to show the HTML representation or trust the notebook. <br />On GitHub, the HTML representation is unable to render, please try loading this page with nbviewer.org.</b></div><div class=\"sk-container\" hidden><div class=\"sk-item\"><div class=\"sk-estimator sk-toggleable\"><input class=\"sk-toggleable__control sk-hidden--visually\" id=\"sk-estimator-id-1\" type=\"checkbox\" checked><label for=\"sk-estimator-id-1\" class=\"sk-toggleable__label sk-toggleable__label-arrow\">LogisticRegression</label><div class=\"sk-toggleable__content\"><pre>LogisticRegression(C=0.1, max_iter=5000)</pre></div></div></div></div></div>"
      ],
      "text/plain": [
       "LogisticRegression(C=0.1, max_iter=5000)"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from sklearn.linear_model import LogisticRegression\n",
    "\n",
    "clf = execution.outputs.get(\"o0\", LogisticRegression)\n",
    "clf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>bill_length_mm</th>\n",
       "      <th>bill_depth_mm</th>\n",
       "      <th>flipper_length_mm</th>\n",
       "      <th>body_mass_g</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>111</th>\n",
       "      <td>45.6</td>\n",
       "      <td>20.3</td>\n",
       "      <td>191.0</td>\n",
       "      <td>4600.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>158</th>\n",
       "      <td>45.4</td>\n",
       "      <td>14.6</td>\n",
       "      <td>211.0</td>\n",
       "      <td>4800.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>291</th>\n",
       "      <td>50.5</td>\n",
       "      <td>19.6</td>\n",
       "      <td>201.0</td>\n",
       "      <td>4050.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>311</th>\n",
       "      <td>47.5</td>\n",
       "      <td>16.8</td>\n",
       "      <td>199.0</td>\n",
       "      <td>3900.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>186</th>\n",
       "      <td>49.1</td>\n",
       "      <td>14.8</td>\n",
       "      <td>220.0</td>\n",
       "      <td>5150.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     bill_length_mm  bill_depth_mm  flipper_length_mm  body_mass_g\n",
       "111            45.6           20.3              191.0       4600.0\n",
       "158            45.4           14.6              211.0       4800.0\n",
       "291            50.5           19.6              201.0       4050.0\n",
       "311            47.5           16.8              199.0       3900.0\n",
       "186            49.1           14.8              220.0       5150.0"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from palmerpenguins import load_penguins\n",
    "from workflows import example_intro\n",
    "\n",
    "data = load_penguins().dropna().sample(5, random_state=123)\n",
    "features = data[example_intro.FEATURES]\n",
    "features.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Predictions ['Adelie' 'Gentoo' 'Chinstrap' 'Chinstrap' 'Gentoo']\n",
      "Ground truth ['Adelie' 'Gentoo' 'Chinstrap' 'Chinstrap' 'Gentoo']\n"
     ]
    }
   ],
   "source": [
    "targets = data[example_intro.TARGET]\n",
    "print(\"Predictions\", clf.predict(features))\n",
    "print(\"Ground truth\", targets.values)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ⏱️ Scheduling Launchplans\n",
    "\n",
    "Run your workflows on a schedule."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A launchplan is a wrapper around a `workflow` function, providing additional\n",
    "functionality such as scheduling workflows on a particular cadence.\n",
    "\n",
    "First, let's register our workflow module with `pyflyte register`. This is\n",
    "a command-line tool for you to package and register python modules that contain\n",
    "Flyte tasks, workflows, and launchplans."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "\n",
    "os.environ[\"VERSION\"] = datetime.now().strftime(\"%Y%m%d%H%M%S\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Running pyflyte register from /Users/nielsbantilan/git/flyte-school/00-intro with images ImageConfig(default_image=Image(name='default', fqn='ghcr.io/unionai-oss/flyte-school', tag='00-intro-latest'), images=[Image(name='default', fqn='ghcr.io/unionai-oss/flyte-school', tag='00-intro-latest')]) and image destination folder /root on 1 package(s) ('/Users/nielsbantilan/git/flyte-school/00-intro/workflows',)\n",
      "Registering against localhost:30080\n",
      "Detected Root /Users/nielsbantilan/git/flyte-school/00-intro, using this to create deployable package...\n",
      "No output path provided, using a temporary directory at /var/folders/4q/frdnh9l10h53gggw1m59gr9m0000gp/T/tmp9eul9iff instead\n",
      "Loading packages ['workflows'] under source root /Users/nielsbantilan/git/flyte-school/00-intro\n",
      "Promise(node:n2.o0)\n",
      "Successfully serialized 7 flyte objects\n",
      "[✔] Registration workflows.example_intro.get_data type TASK successful with version 20230720123757\n",
      "[✔] Registration workflows.example_intro.split_data type TASK successful with version 20230720123757\n",
      "[✔] Registration workflows.example_intro.train_model type TASK successful with version 20230720123757\n",
      "[✔] Registration workflows.example_intro.evaluate type TASK successful with version 20230720123757\n",
      "[✔] Registration workflows.example_intro.training_workflow type WORKFLOW successful with version 20230720123757\n",
      "[✔] Registration workflows.example_intro.training_workflow type LAUNCH_PLAN successful with version 20230720123757\n",
      "[✔] Registration scheduled_training_workflow type LAUNCH_PLAN successful with version 20230720123757\n",
      "Successfully registered 7 entities\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte register --image $IMAGE workflows --version $VERSION"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Recall in the `example_intro.py` script that we defined a launch plan called\n",
    "`scheduled_training_workflow`. You can activate scheduled launchplans from the CLI:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, get the latest launchplan version:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  version: \"20230720123757\"\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "flytectl get launchplan \\\n",
    "    --project flytesnacks \\\n",
    "    --domain development \\\n",
    "    scheduled_training_workflow \\\n",
    "    --output yaml --latest  \\\n",
    "    | grep 'version:' -m 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Expected output:\n",
    "```\n",
    "version: <version>\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the `flytectl` CLI, activate the launchplan:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "updated launchplan successfully on scheduled_training_workflow"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "flytectl update launchplan \\\n",
    "    -p flytesnacks -d development \\\n",
    "    scheduled_training_workflow \\\n",
    "    --version $VERSION --activate"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Make sure it's activated:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  state: ACTIVE\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "flytectl get launchplan \\\n",
    "    -p flytesnacks -d development \\\n",
    "    scheduled_training_workflow \\\n",
    "    --output yaml --latest \\\n",
    "    | grep ' state:'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Expected output:\n",
    "```\n",
    "state: ACTIVE\n",
    "``````"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also use `FlyteRemote` to activate launchplans in Python:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "activated scheduled_training_workflow\n"
     ]
    }
   ],
   "source": [
    "from workflows.utils import get_remote\n",
    "\n",
    "remote = get_remote()\n",
    "launch_plan = remote.fetch_launch_plan(name=\"scheduled_training_workflow\")\n",
    "remote.client.update_launch_plan(launch_plan.id, \"ACTIVE\")\n",
    "print(\"activated scheduled_training_workflow\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get the execution for the most recent scheduled run:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<FlyteLiteral id { project: \"flytesnacks\" domain: \"development\" name: \"f00599e3691638724000\" } spec { launch_plan { resource_type: LAUNCH_PLAN project: \"flytesnacks\" domain: \"development\" name: \"scheduled_training_workflow\" version: \"20230720123757\" } metadata { mode: SCHEDULED scheduled_at { seconds: 1689871200 } system_metadata { } } labels { } annotations { } auth_role { } } closure { phase: RUNNING started_at { seconds: 1689871200 nanos: 84181000 } duration { } created_at { seconds: 1689871200 nanos: 74376000 } updated_at { seconds: 1689871200 nanos: 84181000 } }>\n"
     ]
    }
   ],
   "source": [
    "recent_executions = [\n",
    "    ex for ex in remote.recent_executions()\n",
    "    if ex.spec.launch_plan.name == \"scheduled_training_workflow\"\n",
    "]\n",
    "\n",
    "scheduled_execution = None\n",
    "if recent_executions:\n",
    "    scheduled_execution = recent_executions[0]\n",
    "    scheduled_execution = remote.sync(scheduled_execution)\n",
    "    \n",
    "print(scheduled_execution)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style>#sk-container-id-2 {color: black;}#sk-container-id-2 pre{padding: 0;}#sk-container-id-2 div.sk-toggleable {background-color: white;}#sk-container-id-2 label.sk-toggleable__label {cursor: pointer;display: block;width: 100%;margin-bottom: 0;padding: 0.3em;box-sizing: border-box;text-align: center;}#sk-container-id-2 label.sk-toggleable__label-arrow:before {content: \"▸\";float: left;margin-right: 0.25em;color: #696969;}#sk-container-id-2 label.sk-toggleable__label-arrow:hover:before {color: black;}#sk-container-id-2 div.sk-estimator:hover label.sk-toggleable__label-arrow:before {color: black;}#sk-container-id-2 div.sk-toggleable__content {max-height: 0;max-width: 0;overflow: hidden;text-align: left;background-color: #f0f8ff;}#sk-container-id-2 div.sk-toggleable__content pre {margin: 0.2em;color: black;border-radius: 0.25em;background-color: #f0f8ff;}#sk-container-id-2 input.sk-toggleable__control:checked~div.sk-toggleable__content {max-height: 200px;max-width: 100%;overflow: auto;}#sk-container-id-2 input.sk-toggleable__control:checked~label.sk-toggleable__label-arrow:before {content: \"▾\";}#sk-container-id-2 div.sk-estimator input.sk-toggleable__control:checked~label.sk-toggleable__label {background-color: #d4ebff;}#sk-container-id-2 div.sk-label input.sk-toggleable__control:checked~label.sk-toggleable__label {background-color: #d4ebff;}#sk-container-id-2 input.sk-hidden--visually {border: 0;clip: rect(1px 1px 1px 1px);clip: rect(1px, 1px, 1px, 1px);height: 1px;margin: -1px;overflow: hidden;padding: 0;position: absolute;width: 1px;}#sk-container-id-2 div.sk-estimator {font-family: monospace;background-color: #f0f8ff;border: 1px dotted black;border-radius: 0.25em;box-sizing: border-box;margin-bottom: 0.5em;}#sk-container-id-2 div.sk-estimator:hover {background-color: #d4ebff;}#sk-container-id-2 div.sk-parallel-item::after {content: \"\";width: 100%;border-bottom: 1px solid gray;flex-grow: 1;}#sk-container-id-2 div.sk-label:hover label.sk-toggleable__label {background-color: #d4ebff;}#sk-container-id-2 div.sk-serial::before {content: \"\";position: absolute;border-left: 1px solid gray;box-sizing: border-box;top: 0;bottom: 0;left: 50%;z-index: 0;}#sk-container-id-2 div.sk-serial {display: flex;flex-direction: column;align-items: center;background-color: white;padding-right: 0.2em;padding-left: 0.2em;position: relative;}#sk-container-id-2 div.sk-item {position: relative;z-index: 1;}#sk-container-id-2 div.sk-parallel {display: flex;align-items: stretch;justify-content: center;background-color: white;position: relative;}#sk-container-id-2 div.sk-item::before, #sk-container-id-2 div.sk-parallel-item::before {content: \"\";position: absolute;border-left: 1px solid gray;box-sizing: border-box;top: 0;bottom: 0;left: 50%;z-index: -1;}#sk-container-id-2 div.sk-parallel-item {display: flex;flex-direction: column;z-index: 1;position: relative;background-color: white;}#sk-container-id-2 div.sk-parallel-item:first-child::after {align-self: flex-end;width: 50%;}#sk-container-id-2 div.sk-parallel-item:last-child::after {align-self: flex-start;width: 50%;}#sk-container-id-2 div.sk-parallel-item:only-child::after {width: 0;}#sk-container-id-2 div.sk-dashed-wrapped {border: 1px dashed gray;margin: 0 0.4em 0.5em 0.4em;box-sizing: border-box;padding-bottom: 0.4em;background-color: white;}#sk-container-id-2 div.sk-label label {font-family: monospace;font-weight: bold;display: inline-block;line-height: 1.2em;}#sk-container-id-2 div.sk-label-container {text-align: center;}#sk-container-id-2 div.sk-container {/* jupyter's `normalize.less` sets `[hidden] { display: none; }` but bootstrap.min.css set `[hidden] { display: none !important; }` so we also need the `!important` here to be able to override the default hidden behavior on the sphinx rendered scikit-learn.org. See: https://github.com/scikit-learn/scikit-learn/issues/21755 */display: inline-block !important;position: relative;}#sk-container-id-2 div.sk-text-repr-fallback {display: none;}</style><div id=\"sk-container-id-2\" class=\"sk-top-container\"><div class=\"sk-text-repr-fallback\"><pre>LogisticRegression(C=0.1, max_iter=1000)</pre><b>In a Jupyter environment, please rerun this cell to show the HTML representation or trust the notebook. <br />On GitHub, the HTML representation is unable to render, please try loading this page with nbviewer.org.</b></div><div class=\"sk-container\" hidden><div class=\"sk-item\"><div class=\"sk-estimator sk-toggleable\"><input class=\"sk-toggleable__control sk-hidden--visually\" id=\"sk-estimator-id-2\" type=\"checkbox\" checked><label for=\"sk-estimator-id-2\" class=\"sk-toggleable__label sk-toggleable__label-arrow\">LogisticRegression</label><div class=\"sk-toggleable__content\"><pre>LogisticRegression(C=0.1, max_iter=1000)</pre></div></div></div></div></div>"
      ],
      "text/plain": [
       "LogisticRegression(C=0.1, max_iter=1000)"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from sklearn.linear_model import LogisticRegression\n",
    "\n",
    "scheduled_execution = remote.wait(scheduled_execution)\n",
    "clf = scheduled_execution.outputs.get(\"o0\", LogisticRegression)\n",
    "clf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Deactivate the schedule with the `flytectl` CLI:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "updated launchplan successfully on scheduled_training_workflow"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "flytectl update launchplan \\\n",
    "    -p flytesnacks -d development \\\n",
    "    scheduled_training_workflow \\\n",
    "    --version $VERSION --archive"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Or with `FlyteRemote`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "deactivated scheduled_training_workflow\n"
     ]
    }
   ],
   "source": [
    "remote.client.update_launch_plan(launch_plan.id, \"INACTIVE\")\n",
    "print(\"deactivated scheduled_training_workflow\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 🛩️ Flyte Programming Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<image src=\"static/flyte_programming_model_overview.png\" width=\"800px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To give you a sense of what's happening when we do `pyflyte run`, let's run just the\n",
    "`get_data` task from `example_intro.py`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Go to http://localhost:30080/console/projects/flytesnacks/domains/development/executions/fe1f234aa7f5a49a09f2 to see execution in the console.\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte run --remote \\\n",
    "    --image $IMAGE \\\n",
    "    workflows/example_intro.py get_data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we go to the flyteconsole link provided by the pyflyte run command, we can dig into the guts of a task container execution:\n",
    "\n",
    "> - In the **Executions** tab, go to the **Logs** link, which will take you to the Kubernetes dashboard logs.\n",
    "> - If you go to the Pod metadata description, you can see the **Arguments** that are provided as the entrypoint to the task container."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Usage: pyflyte-fast-execute [OPTIONS] [TASK_EXECUTE_CMD]...\n",
      "\n",
      "  Downloads a compressed code distribution specified by additional-\n",
      "  distribution and then calls the underlying task execute command for the\n",
      "  updated code.\n",
      "\n",
      "Options:\n",
      "  --additional-distribution TEXT\n",
      "  --dest-dir TEXT\n",
      "  --help                          Show this message and exit.\n"
     ]
    }
   ],
   "source": [
    "!pyflyte-fast-execute --help"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ⌨️ Type System"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Flyte type system is responsible for a lot of Flyte's production-grade\n",
    "qualities:\n",
    "\n",
    "- 👟 Run-time type-checking.\n",
    "- 📦 Serialization/deserialization of IO between tasks.\n",
    "- 📝 Type-checking of workflows at compile-time.\n",
    "\n",
    "This type system is language-agnostic and is implemented in the `flyteidl`\n",
    "protobuf format. This means that Flyte SDKs can be written in any language.\n",
    "Currently `flytekit` is the Python implementation, but there are also\n",
    "Java, Scala, and Javascript SDKs.\n",
    "\n",
    "<image src=\"static/flyte_type_system.png\" width=\"400px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 🔀 How Data Flows in Flyte"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If tasks run in their own containers inside the Flyte cluster, how is data passed between them?\n",
    "\n",
    "`flytekit` needs to convert all the Python types into something that the `flyteidl`\n",
    "protobuf type system can understand. This is done by the `flytekit` type engine.\n",
    "We'll learn more about how this works later, but at a high-level, it looks like\n",
    "this:\n",
    "\n",
    "<image src=\"static/flyte_data_flow.png\" width=\"500px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To see this diagram in action, let's kick off another workflow execution:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'http://localhost:30080/console/projects/flytesnacks/domains/development/executions/f29b9a98f19b24330a9a'"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from workflows import example_intro\n",
    "from workflows.utils import get_remote\n",
    "\n",
    "remote = get_remote()\n",
    "execution = remote.execute_local_workflow(\n",
    "    example_intro.training_workflow,\n",
    "    inputs={\n",
    "        \"hyperparameters\": example_intro.Hyperparameters(C=0.1, max_iter=5000),\n",
    "        \"test_size\": 0.2,\n",
    "        \"random_state\": 11,\n",
    "    }\n",
    ")\n",
    "remote.generate_console_url(execution)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the execution completes, we can see the inputs and outputs of each task\n",
    "in the workflow execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution = remote.wait(execution)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get the node executions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dict_keys(['end-node', 'n0', 'n1', 'n2', 'n3', 'n4', 'start-node'])"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "execution.node_executions.keys()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get the output of the `get_data` node:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'s3://my-s3-bucket/data/tp/f7e7704f07a4a4d84b98-n0-0/8dae9f6bb49d3546d0240fb07baa43e2'"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "get_data_node = execution.node_executions[\"n0\"]\n",
    "get_data_node.outputs[\"o0\"]._literal_sd.uri"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load the data up using the `remote.remote_context()` context manager."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>species</th>\n",
       "      <th>bill_length_mm</th>\n",
       "      <th>bill_depth_mm</th>\n",
       "      <th>flipper_length_mm</th>\n",
       "      <th>body_mass_g</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Adelie</td>\n",
       "      <td>39.1</td>\n",
       "      <td>18.7</td>\n",
       "      <td>181.0</td>\n",
       "      <td>3750.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Adelie</td>\n",
       "      <td>39.5</td>\n",
       "      <td>17.4</td>\n",
       "      <td>186.0</td>\n",
       "      <td>3800.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Adelie</td>\n",
       "      <td>40.3</td>\n",
       "      <td>18.0</td>\n",
       "      <td>195.0</td>\n",
       "      <td>3250.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  species  bill_length_mm  bill_depth_mm  flipper_length_mm  body_mass_g\n",
       "0  Adelie            39.1           18.7              181.0       3750.0\n",
       "1  Adelie            39.5           17.4              186.0       3800.0\n",
       "2  Adelie            40.3           18.0              195.0       3250.0"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "\n",
    "get_data_node = execution.node_executions[\"n0\"]\n",
    "with remote.remote_context():\n",
    "    sd = get_data_node.outputs[\"o0\"]\n",
    "    data = sd.open(pd.DataFrame).all()\n",
    "\n",
    "data.head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get the input of the `split_data` node:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'s3://my-s3-bucket/data/w8/f29b9a98f19b24330a9a-n0-0/896e2d26405a8744c6005206da53068b'"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "split_data_node = execution.node_executions[\"n1\"]\n",
    "split_data_node.inputs[\"data\"]._literal_sd.uri"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Make sure they're the same"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test passed! ✅\n"
     ]
    }
   ],
   "source": [
    "assert split_data_node.inputs[\"data\"]._literal_sd.uri == get_data_node.outputs[\"o0\"]._literal_sd.uri\n",
    "print(\"Test passed! ✅\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Flyte sandbox cluster actually runs a Minio object store that stores all of\n",
    "the inputs and outputs. We can find them on the Minio dashboard http://localhost:30080/minio, or we can get it through the terminal:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sh\n",
    "export AWS_ACCESS_KEY_ID=minio\n",
    "export AWS_SECRET_ACCESS_KEY=miniostorage\n",
    "aws --endpoint-url http://localhost:30002 s3 ls <s3_uri>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 🌳 Lifecycle of a Workflow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When you run a workflow locally, flytekit just runs the tasks in a Python runtime. However, when you run the workflow on a Flyte cluster, a lot of things are happening under the hood.\n",
    "\n",
    "`flytepropeller` is the core engine in the Flyte stack that orchestrates:\n",
    "- ➡️ the execution of tasks in a particular sequence.\n",
    "- 📊 the management of data dependencies between tasks.\n",
    "- 💿 the compute infrastructure needed to run a task.\n",
    "- ... and much more.\n",
    "\n",
    "This is a very deep topic that we don't have time to cover in this workshop, but\n",
    "here's a high-level overview of what happens:\n",
    "\n",
    "<image src=\"static/flyte_workflow_lifecycle.png\" width=\"500px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 🧑‍💻 Development Lifecycle Overview"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Typically, working with Flyte as a data scientist looks like this:\n",
    "\n",
    "- 💻 Create and test tasks/workflows locally with `pyflyte run`\n",
    "- 📦 Build a container for your tasks/workflows\n",
    "- 🔁 Iterate on a Flyte cluster with `pyflyte run --remote`\n",
    "- 🚀 Deploy to production on Flyte cluster using `pyflyte register --non-fast`\n",
    "\n",
    "We haven't gone through the build step, but it would look something like this:\n",
    "\n",
    "```bash\n",
    "make docker-build\n",
    "```\n",
    "\n",
    "This make target is in the [`Makefile`](./Makefile), which defines how to build\n",
    "and push the docker image.\n",
    "\n",
    "If we go to the [`Dockerfile`](./Dockerfile), we can see that this packages up\n",
    "all of the Flyte workflow code and the third-party dependencies into the image.\n",
    "\n",
    "Now let's understand what `pyflyte register` is."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `pyflyte register`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By default, `pyflyte register` zips up all of the source code of your Flyte\n",
    "application.\n",
    "\n",
    "Flyte supports rapid iteration during development via \"fast registration\" via\n",
    "`pyflyte register`. This zips up all of the source code of your Flyte \n",
    "application and bypasses the need to re-build a docker image with your updated\n",
    "code in it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Running pyflyte register from /Users/nielsbantilan/git/flyte-school/00-intro with images ImageConfig(default_image=Image(name='default', fqn='ghcr.io/unionai-oss/flyte-school', tag='00-intro-latest'), images=[Image(name='default', fqn='ghcr.io/unionai-oss/flyte-school', tag='00-intro-latest')]) and image destination folder /root on 1 package(s) ('/Users/nielsbantilan/git/flyte-school/00-intro/workflows',)\n",
      "Registering against localhost:30080\n",
      "Detected Root /Users/nielsbantilan/git/flyte-school/00-intro, using this to create deployable package...\n",
      "No output path provided, using a temporary directory at /var/folders/4q/frdnh9l10h53gggw1m59gr9m0000gp/T/tmpjkdpsihu instead\n",
      "Computed version is acp5-WGxARLHZgGZA4FtSg==\n",
      "Loading packages ['workflows'] under source root /Users/nielsbantilan/git/flyte-school/00-intro\n",
      "Promise(node:n2.o0)\n",
      "Successfully serialized 7 flyte objects\n",
      "[✔] Registration workflows.example_intro.get_data type TASK successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "[✔] Registration workflows.example_intro.split_data type TASK successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "[✔] Registration workflows.example_intro.train_model type TASK successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "[✔] Registration workflows.example_intro.evaluate type TASK successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "[✔] Registration workflows.example_intro.training_workflow type WORKFLOW successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "[✔] Registration workflows.example_intro.training_workflow type LAUNCH_PLAN successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "[✔] Registration scheduled_training_workflow type LAUNCH_PLAN successful with version acp5-WGxARLHZgGZA4FtSg==\n",
      "Successfully registered 7 entities\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte register \\\n",
    "    --image $IMAGE \\\n",
    "    --project flytesnacks \\\n",
    "    --domain development \\\n",
    "    --version myversion0 \\\n",
    "    workflows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "> ℹ️ **Note**: You can provide an explicit `--version` flag, but by default `flytekit` will create a version string for you."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's register a new version of the `example_intro.py` workflow with `pyflyte run`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Go to http://localhost:30080/console/projects/flytesnacks/domains/development/executions/f3659537700944e469b8 to see execution in the console.\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte run --remote \\\n",
    "    --image $IMAGE \\\n",
    "    workflows/example_intro.py training_workflow \\\n",
    "    --hyperparameters '{\"C\": 0.01}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "If we go to the Flyte console link provided the `pyflyte run`, we can understand\n",
    "what happened during the execution by going to the **Task** tag and looking at\n",
    "the `--additional-distribution` flag in the container arguments. The `tar.gz` file\n",
    "is the packaged-up Flyte code in our repository, which `flytekit` understands\n",
    "is the actual task code that we want to run."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `pyflyte register --non-fast`\n",
    "\n",
    "Unlike `pyflyte register`, `pyflyte register --non-fast` *will not* package up\n",
    "the user code and simply use the source code that is present in the image when\n",
    "it was originally built.\n",
    "\n",
    "This is the **recommended** way to deploy to production because it ensures that\n",
    "the source code that is running in production is the same as the source code in\n",
    "the image, since the `tar.gz` *could in theory* be swapped out by a malicious\n",
    "actor.\n",
    "\n",
    "Since we're deploying to production, let's specify the `--domain production` flag."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Running pyflyte register from /Users/nielsbantilan/git/flyte-school/00-intro with images ImageConfig(default_image=Image(name='default', fqn='ghcr.io/unionai-oss/flyte-school', tag='00-intro-latest'), images=[Image(name='default', fqn='ghcr.io/unionai-oss/flyte-school', tag='00-intro-latest')]) and image destination folder /root on 1 package(s) ('/Users/nielsbantilan/git/flyte-school/00-intro/workflows',)\n",
      "Registering against localhost:30080\n",
      "Detected Root /Users/nielsbantilan/git/flyte-school/00-intro, using this to create deployable package...\n",
      "Loading packages ['workflows'] under source root /Users/nielsbantilan/git/flyte-school/00-intro\n",
      "Successfully serialized 7 flyte objects\n",
      "[✔] Registration workflows.example_intro.get_data type TASK successful with version prod-v0\n",
      "[✔] Registration workflows.example_intro.split_data type TASK successful with version prod-v0\n",
      "[✔] Registration workflows.example_intro.train_model type TASK successful with version prod-v0\n",
      "[✔] Registration workflows.example_intro.evaluate type TASK successful with version prod-v0\n",
      "[✔] Registration workflows.example_intro.training_workflow type WORKFLOW successful with version prod-v0\n",
      "[✔] Registration workflows.example_intro.training_workflow type LAUNCH_PLAN successful with version prod-v0\n",
      "[✔] Registration scheduled_training_workflow type LAUNCH_PLAN successful with version prod-v0\n",
      "Successfully registered 7 entities\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "pyflyte register \\\n",
    "    --image $IMAGE \\\n",
    "    --project flytesnacks \\\n",
    "    --domain production \\\n",
    "    --non-fast \\\n",
    "    --version prod-v0 \\\n",
    "    workflows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusion\n",
    "\n",
    "🎉 Congrats! 🎉\n",
    "\n",
    "You've made it through to the end of this module! To recap, we've:\n",
    "\n",
    "- 🧱 Learned the basics constructs of Flyte: tasks, workflows, and launchplans.\n",
    "- 🔀 Understood how Flyte orchestrates execution graphs, data, and compute infrastructure.\n",
    "- ⌨️ Understood how the Flyte type system provides type-safety as well as IO abstraction.\n",
    "- 👟 Learned how to run tasks/workflows programmatically through the `pyflyte run` CLI and Python `FlyteRemote` client.\n",
    "\n",
    "### Resources\n",
    "\n",
    "<a href=\"https://docs.flyte.org/en/latest/\">\n",
    "    <img src=\"https://img.shields.io/badge/Flyte-Docs-purple?style=for-the-badge\" alt=\"Flyte Docs\" />\n",
    "</a>\n",
    "<a href=\"https://slack.flyte.org\">\n",
    "    <img src=\"https://img.shields.io/badge/Slack-Chat-pink?style=for-the-badge&logo=slack\" alt=\"Flyte Slack\" />\n",
    "</a>\n",
    "<a href=\"https://github.com/flyteorg/flyte\">\n",
    "    <img src=\"https://img.shields.io/badge/Flyte-Github%20Repo-black?style=for-the-badge&logo=github\" alt=\"Flyte Repo\" />\n",
    "</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<iframe src=\"https://docs.google.com/presentation/d/e/2PACX-1vRpMyFw3lfocwbDXcT3XhLEROjZ5mYVCrmQ5WHxikX4m6tlf00q8FgDgwFZSUwheM0luy-KR6Djw8yz/embed?start=false&loop=false&delayms=3000\" frameborder=\"0\" width=\"960\" height=\"569\" allowfullscreen=\"true\" mozallowfullscreen=\"true\" webkitallowfullscreen=\"true\"></iframe>"
   ]
  }
 ],
 "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.9.16"
  },
  "vscode": {
   "interpreter": {
    "hash": "1c1aa76aeeda18f711a0c7c067c98817deca9b2faedcaffac8fadd364058945b"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}