vectorization04.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "source": [
    "# introduction to the vectorizing principle\n",
    "\n",
    "## part 4: defining a data pipeline\n",
    "\n",
    "Here, we are focusing on data transformation code used for curating data or training and evaluating a model. In this case, a data pipeline can be used as a means of splitting the work in pieces that are conceptionally relevant for creating, operating, or maintaining the code. One can see a data pipeline as a directed acyclic graph (DAG) of tasks. Often a hierarchical notion of grouping the tasks is useful for example grouping them in stages. Typical stages of a data pipeline may be:\n",
    "\n",
    "1. raw data ingestion\n",
    "2. early cleaning for easier inspection\n",
    "3. transformation in the best possible form for subject matter reasoning\n",
    "4. feature enrichment\n",
    "5. model training\n",
    "6. model evaluation\n",
    "\n",
    "Data pipeline orchestration tools can help with tracing/monitoring different executions of the pipeline. They may help with communication between tasks as well as materializations/dematerializations to persistent storage (files or databases). They may introduce transactionality features like showing the result of a stage only on success. And they can help with caching and automatic cache-invalidation. It helps to think of intermediate data within a pipeline as caches since they (should) only depend on external inputs into the pipeline and the code running it. Often a notion of caching is implemented in an ad-hoc fasion with manual control but this causes trouble if multiple instances of the pipeline should be operated in order to ensure an agile development process working with small, medium, large input data both with per-user and team shared caches. For most data analytic pipelines, the algorithm is not fixed upfront. It is developed as part of the creation process. Thus, a high development iteration speed is key! Good software design and reproduction/testing strategies are needed to keep up the speed long enough to create maximum value.\n",
    "\n",
    "#### define tasks:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "from pydiverse.pipedag.materialize import Blob, Table, materialize\n",
    "import lightgbm\n",
    "from sklearn.metrics import mean_squared_error\n",
    "from sklearn.model_selection import train_test_split\n",
    "import math\n",
    "\n",
    "@materialize(version=\"1.0.0\")\n",
    "def read_input_data():\n",
    "    data_df = pd.read_csv('data/taxi_data/train.csv.gz')\n",
    "    return Table(data_df, name=\"input_data\")\n",
    "\n",
    "\n",
    "@materialize(version=\"1.0.0\", input_type=pd.DataFrame)\n",
    "def feature_trip_distance(df: pd.DataFrame):\n",
    "    start_lat = np.radians(df[\"pickup_latitude\"])\n",
    "    start_lng = np.radians(df[\"pickup_longitude\"])\n",
    "    dest_lat = np.radians(df[\"dropoff_latitude\"])\n",
    "    dest_lng = np.radians(df[\"dropoff_longitude\"])\n",
    "\n",
    "    d = (\n",
    "        np.sin(dest_lat / 2 - start_lat / 2) ** 2\n",
    "        + np.cos(start_lat)\n",
    "        * np.cos(dest_lat)\n",
    "        * np.sin(dest_lng / 2 - start_lng / 2) ** 2\n",
    "    )\n",
    "\n",
    "    return Table(pd.DataFrame(\n",
    "        dict(\n",
    "            id=df[\"id\"],\n",
    "            # 6,371 km is the earth radius\n",
    "            haversine_distance = 2 * 6371 * np.arcsin(np.sqrt(d))\n",
    "        )\n",
    "    ), name=\"trip_distance\")\n",
    "\n",
    "@materialize(version=\"1.0.0\", input_type=pd.DataFrame)\n",
    "def feature_split_pickup_datetime(df: pd.DataFrame):\n",
    "    tpep_pickup_datetime = pd.to_datetime(df[\"pickup_datetime\"])\n",
    "\n",
    "    return Table(pd.DataFrame(\n",
    "        dict(\n",
    "            id=df[\"id\"],\n",
    "            pickup_dayofweek=tpep_pickup_datetime.dt.dayofweek,\n",
    "            pickup_hour=tpep_pickup_datetime.dt.hour,\n",
    "            pickup_minute=tpep_pickup_datetime.dt.minute,\n",
    "        )\n",
    "    ), name=\"pickup_datetime\")\n",
    "\n",
    "@materialize(nout=2, version=\"1.0.0\", input_type=pd.DataFrame)\n",
    "def get_feature_df(df: pd.DataFrame, features: list[pd.DataFrame], target_col=\"trip_duration\"):\n",
    "    final_df = df[[\"id\"] + [col for col in df.columns if col != target_col and df[col].dtype in (int, float, bool)]]\n",
    "    for feature_df in features:\n",
    "        final_df = final_df.merge(feature_df, on=\"id\")\n",
    "    return (\n",
    "        Table(final_df, name=\"features\"),\n",
    "        Table(df[[\"id\", target_col]], name=\"target\"),\n",
    "    )\n",
    "\n",
    "\n",
    "def combine_features(data_df):\n",
    "    features = [\n",
    "        feature_trip_distance(data_df),\n",
    "        feature_split_pickup_datetime(data_df),\n",
    "    ]\n",
    "    return get_feature_df(data_df, features)\n",
    "\n",
    "\n",
    "@materialize(nout=4, version=\"1.0.0\", input_type=pd.DataFrame)\n",
    "def split_train_test(features_df: pd.DataFrame, target_df: pd.DataFrame):\n",
    "    features_df.sort_values(\"id\", inplace=True)\n",
    "    target_df.sort_values(\"id\", inplace=True)\n",
    "    (\n",
    "        features_train,\n",
    "        features_test,\n",
    "        target_train,\n",
    "        target_test,\n",
    "    ) = train_test_split(features_df, target_df, test_size=0.1)\n",
    "    return (\n",
    "        Table(features_train, name=\"features_train\"),\n",
    "        Table(features_test, name=\"features_test\"),\n",
    "        Table(target_train, name=\"target_train\"),\n",
    "        Table(target_test, name=\"target_test\"),\n",
    "    )\n",
    "\n",
    "@materialize(version=\"1.0.0\", input_type=pd.DataFrame)\n",
    "def train_model(features_train: pd.DataFrame, target_train: pd.DataFrame):\n",
    "    features_train.sort_values(\"id\", inplace=True)\n",
    "    target_train.sort_values(\"id\", inplace=True)\n",
    "    del features_train[\"id\"]\n",
    "    del target_train[\"id\"]\n",
    "    model = lightgbm.LGBMRegressor(objective=\"regression_l1\")\n",
    "    model.fit(features_train, target_train)\n",
    "    return Blob(model, name=\"model\")\n",
    "\n",
    "\n",
    "@materialize(version=\"1.0.0\", input_type=pd.DataFrame)\n",
    "def evaluate_model(features_test: pd.DataFrame, target_test: pd.DataFrame, model: lightgbm.LGBMRegressor):\n",
    "    features_test.sort_values(\"id\", inplace=True)\n",
    "    target_test.sort_values(\"id\", inplace=True)\n",
    "    del features_test[\"id\"]\n",
    "    del target_test[\"id\"]\n",
    "    predicted = model.predict(features_test)\n",
    "    print(model.score(features_test, target_test))\n",
    "    print(math.sqrt(mean_squared_error(target_test, predicted)))\n",
    "    lightgbm.plot_importance(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "source": [
    "#### define pipeline:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "from pydiverse.pipedag.core import Flow, PipedagConfig, Stage\n",
    "def get_pipeline():\n",
    "    with Flow(\"vectorization\") as flow:\n",
    "        with Stage(\"01_raw_input\"):\n",
    "            data_df = read_input_data()\n",
    "        with Stage(\"02_features\"):\n",
    "            features_df, target_df = combine_features(data_df)\n",
    "            (\n",
    "                features_train,\n",
    "                features_test,\n",
    "                target_train,\n",
    "                target_test,\n",
    "            ) = split_train_test(features_df, target_df)\n",
    "        with Stage(\"03_model\"):\n",
    "            model = train_model(features_train, target_train)\n",
    "        with Stage(\"04_evaluation\"):\n",
    "            evaluate_model(features_test, target_test, model)\n",
    "    return flow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "source": [
    "#### setup logging:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "import logging\n",
    "from pydiverse.pipedag.util.structlog import setup_logging\n",
    "setup_logging(log_level=logging.INFO)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "source": [
    "#### run pipeline:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\u001b[2m2024-08-06 09:17:14.010071\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mInitialized SQL Table Store   \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m \u001b[36mengine_url\u001b[0m=\u001b[35mduckdb:////tmp/pipedag/vectorization/db.duckdb\u001b[0m \u001b[36mschema_prefix\u001b[0m=\u001b[35m\u001b[0m \u001b[36mschema_suffix\u001b[0m=\u001b[35m\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.017183\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mInitialized SQL Table Store   \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m \u001b[36mengine_url\u001b[0m=\u001b[35mduckdb:////tmp/pipedag/vectorization/db.duckdb\u001b[0m \u001b[36mschema_prefix\u001b[0m=\u001b[35m\u001b[0m \u001b[36mschema_suffix\u001b[0m=\u001b[35m\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.017604\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mStarting IPCServer            \u001b[0m [\u001b[0m\u001b[1m\u001b[34mRunContextServer\u001b[0m]\u001b[0m \u001b[36maddress\u001b[0m=\u001b[35mtcp://127.0.0.1:63222\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.027818\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA IF NOT EXISTS pipedag_metadata\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.033285\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA IF NOT EXISTS \"01_raw_input\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.033973\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mDROP SCHEMA IF EXISTS \"01_raw_input__even\" CASCADE\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.034700\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA \"01_raw_input__even\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.051313\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"01_raw_input__even\".input_data AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"01_raw_input\".input_data\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.056203\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'read_input_data'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'read_input_data' 0x31a4c1010 (id: 0)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.056930\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'read_input_data'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'read_input_data' 0x31a4c1010 (id: 0)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.057634\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mCommitting stage              \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 01_raw_input>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.058843\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mStage is cache valid          \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 01_raw_input>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.063353\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 01_raw_input>\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.COMPLETED: 1>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.065566\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA IF NOT EXISTS \"02_features\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.066091\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mDROP SCHEMA IF EXISTS \"02_features__even\" CASCADE\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.066620\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA \"02_features__even\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.074873\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".trip_distance AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".trip_distance\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.078321\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'feature_trip_distance'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'feature_trip_distance' 0x31a0dd650 (id: 2)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.079013\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'feature_trip_distance'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'feature_trip_distance' 0x31a0dd650 (id: 2)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.085453\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".pickup_datetime AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".pickup_datetime\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.088898\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'feature_split_pickup_datetime'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'feature_split_pickup_datetime' 0x112d030d0 (id: 3)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.089461\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'feature_split_pickup_datetime'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'feature_split_pickup_datetime' 0x112d030d0 (id: 3)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.096282\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".features AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".features\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.101519\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".target AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".target\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.105450\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'get_feature_df'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'get_feature_df' 0x112d01650 (id: 4)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.106085\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'get_feature_df'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'get_feature_df' 0x112d01650 (id: 4)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.112154\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".features_train AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".features_train\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.119981\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".features_test AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".features_test\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.125914\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".target_train AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".target_train\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.131816\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE VIEW \"02_features__even\".target_test AS\n",
      "    \u001b[35mSELECT * \n",
      "    \u001b[35mFROM \"02_features\".target_test\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.136216\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'split_train_test'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'split_train_test' 0x1073d3650 (id: 5)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.136918\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'split_train_test'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'split_train_test' 0x1073d3650 (id: 5)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.137583\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mCommitting stage              \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 02_features>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.138788\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mStage is cache valid          \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 02_features>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.142633\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 02_features>\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.COMPLETED: 1>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.144854\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA IF NOT EXISTS \"03_model\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.145410\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mDROP SCHEMA IF EXISTS \"03_model__odd\" CASCADE\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.145917\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA \"03_model__odd\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.153047\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'train_model'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'train_model' 0x1073d2110 (id: 7)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.153747\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'train_model'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'train_model' 0x1073d2110 (id: 7)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.154615\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mCommitting stage              \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 03_model>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.155717\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mStage is cache valid          \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 03_model>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.159452\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 03_model>\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.COMPLETED: 1>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.161194\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA IF NOT EXISTS \"04_evaluation\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.161703\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mDROP SCHEMA IF EXISTS \"04_evaluation__odd\" CASCADE\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.162176\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mExecuting sql                 \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m\n",
      "    [\u001b[36mquery\u001b[0m]\n",
      "    \u001b[35mCREATE SCHEMA \"04_evaluation__odd\"\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.168816\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFound task in cache. Using cached result.\u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'evaluate_model'\u001b[0m]\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'evaluate_model' 0x112d0a6d0 (id: 9)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.169549\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mTask 'evaluate_model'\u001b[0m]\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.CACHE_VALID: 2>\u001b[0m \u001b[36mtask\u001b[0m=\u001b[35m<Task 'evaluate_model' 0x112d0a6d0 (id: 9)>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.170142\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mCommitting stage              \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 04_evaluation>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.171414\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mStage is cache valid          \u001b[0m [\u001b[0m\u001b[1m\u001b[34mDuckDBTableStore\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 04_evaluation>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.174551\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mTask finished successfully    \u001b[0m [\u001b[0m\u001b[1m\u001b[34mCommit Stage\u001b[0m]\u001b[0m \u001b[36mstage\u001b[0m=\u001b[35m<Stage: 04_evaluation>\u001b[0m \u001b[36mstate\u001b[0m=\u001b[35m<FinalTaskState.COMPLETED: 1>\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.175798\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mFlow visualization            \u001b[0m [\u001b[0m\u001b[1m\u001b[34mFlow\u001b[0m]\u001b[0m \u001b[36murl\u001b[0m=\u001b[35mhttps://kroki.io/graphviz/svg/eNqtksFOhDAURff9ClK3miAwjhOCWz_CmKa0r9hMh5L2VWPM_LtgO4jRxZiBzQ235HAeD6k7x4eX7DH7ID608UaY4BEcy2-Z429M90PA8dzwFkxDly2ticd3A423RsuatJ2wxrqGXuVfV5GPT8SqNVzsa6Jsjz-KPHtKYAdcRiyTHDm9zpQ25gSEe6XE3VjGF9LpDCR9rsmR_KFeMAUcgwO_MP8uVxAvZ_EEZX4wGtmgxT4M0wiA-gBnj1H84qHTI0d75L04n1PNnA7wNDCT6mzAZgbEgdBx3TMEj5etpGQHK8Es9pGaFZaxnZ2jbeRepFsxeOUmcNS2Xzgv6xXEd7N4wsL_3fPs5iEr6phlymr6o2KWKaspN9OCx9ym3E0fL-aRfAL0RVLe\u001b[0m\n",
      "\u001b[2m2024-08-06 09:17:14.380539\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mStopped IPCServer             \u001b[0m [\u001b[0m\u001b[1m\u001b[34mRunContextServer\u001b[0m]\u001b[0m \u001b[36maddress\u001b[0m=\u001b[35mtcp://127.0.0.1:63222\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "flow = get_pipeline()\n",
    "result = flow.run()\n",
    "assert result.successful"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "source": [
    "Next: [vectorization05a.ipynb](vectorization05a.ipynb): generating SQL and dataframe transformation code with one syntax"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}