03-ingest-with-feature-store.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Ingestion and Preparation Using a Feature Store\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the previous section, you created ML features manually on a small scale (using a notebook with Pandas). However, when deploying ML to production, the process of ingesting, transforming, storing, versioning, and serving data should be automated and run at scale. Organizations use dedicated data and ML engineering teams to productize and scale the data pipelines, which requires more time and resources and can lead to errors and inconsistencies. Furthermore, when the application works with real-time transactional data, the data processing is based on streaming technologies, forcing even more complex and duplicate implementations.\n",
    "\n",
    "Feature stores enable reuse and collaboration, where the same features can be engineered once and used in different applications. In addition, features can be updated and refreshed constantly, resulting in more accurate and up-to-date models and a simplified model retraining process.\n",
    "\n",
    "Some feature stores (like MLRun and Tecton) automate the development and deployment of the feature generation (transformation) data pipelines. This section will show you how to use MLRun to easily build feature engineering pipelines and deploy them as batch or real-time scenarios.\n",
    "\n",
    "Feature stores define groups of features (feature sets) that are ingested and transformed together from the same online or offline source. In the example, you will define three feature sets (credit transactions, user activities, and labels), each with its source, transformations, and target store definitions. Feature set definitions can be saved and later deployed in an application context, reading from offline (for example, object storage or a database) or online (for example, a Kafka stream or an HTTP endpoint) sources.\n",
    "\n",
    "After the features are stored and versioned in the feature store, they can be combined into feature vectors that the model training, serving, or monitoring applications will use.\n",
    "\n",
    "In MLRun, all the assets and resources (functions, workflows, data, features, models, and so on) are grouped into projects. Projects can be deployed as a whole, have access control and policies, and are usually mapped to a GIT repository for simple deployment and versioning. The first step in the notebook is to create a new project or load an existing project from the version control or database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The history saving thread hit an unexpected error (DatabaseError('database disk image is malformed')).History will not be written to the database.\n",
      "Both server & client are aligned (1.7.0rc43).\n"
     ]
    }
   ],
   "source": [
    "!/User/align_mlrun.sh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import mlrun\n",
    "\n",
    "project = mlrun.load_project(\n",
    "    name=\"fraud-demo\",\n",
    "    context=\"./\",\n",
    "    user_project=True,\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building the Credit Transactions Data Pipeline (Feature Set)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following part generates and aggregates new features from the credit transactions dataset and stores them in offline and online forms for training and real-time serving. First, define the feature set, its main entity (index), timestamp, and so on. Next, define the set of feature transformations (mappings, encoding, aggregations, custom operators, and so on). And finally, specify how or where to store the results (online and offline data stores) or use the system defaults."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "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>step</th>\n",
       "      <th>age</th>\n",
       "      <th>gender</th>\n",
       "      <th>zipcodeOri</th>\n",
       "      <th>zipMerchant</th>\n",
       "      <th>category</th>\n",
       "      <th>amount</th>\n",
       "      <th>fraud</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>source</th>\n",
       "      <th>target</th>\n",
       "      <th>device</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>274633</th>\n",
       "      <td>91</td>\n",
       "      <td>5</td>\n",
       "      <td>F</td>\n",
       "      <td>28007</td>\n",
       "      <td>28007</td>\n",
       "      <td>es_transportation</td>\n",
       "      <td>26.92</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:35.908872000</td>\n",
       "      <td>C1022153336</td>\n",
       "      <td>M1823072687</td>\n",
       "      <td>33832bb8607545df97632a7ab02d69c4</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>286902</th>\n",
       "      <td>94</td>\n",
       "      <td>2</td>\n",
       "      <td>M</td>\n",
       "      <td>28007</td>\n",
       "      <td>28007</td>\n",
       "      <td>es_transportation</td>\n",
       "      <td>48.22</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:54.202497913</td>\n",
       "      <td>C1006176917</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>fadd829c49e74ffa86c8da3be75ada53</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>416998</th>\n",
       "      <td>131</td>\n",
       "      <td>3</td>\n",
       "      <td>M</td>\n",
       "      <td>28007</td>\n",
       "      <td>28007</td>\n",
       "      <td>es_transportation</td>\n",
       "      <td>17.56</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:59.309667939</td>\n",
       "      <td>C1010936270</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>58d0422a50bc40c89d2b4977b2f1beea</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        step age gender  zipcodeOri  zipMerchant           category  amount  \\\n",
       "274633    91   5      F       28007        28007  es_transportation   26.92   \n",
       "286902    94   2      M       28007        28007  es_transportation   48.22   \n",
       "416998   131   3      M       28007        28007  es_transportation   17.56   \n",
       "\n",
       "        fraud                     timestamp       source       target  \\\n",
       "274633      0 2024-10-06 07:03:35.908872000  C1022153336  M1823072687   \n",
       "286902      0 2024-10-06 07:03:54.202497913  C1006176917   M348934600   \n",
       "416998      0 2024-10-06 07:03:59.309667939  C1010936270   M348934600   \n",
       "\n",
       "                                  device  \n",
       "274633  33832bb8607545df97632a7ab02d69c4  \n",
       "286902  fadd829c49e74ffa86c8da3be75ada53  \n",
       "416998  58d0422a50bc40c89d2b4977b2f1beea  "
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "from src.date_adjust import adjust_data_timespan\n",
    "\n",
    "# Fetch the transactions dataset from the server\n",
    "transactions_data = pd.read_csv(mlrun.get_sample_path(\"data/fraud-demo-mlrun-fs-docs/data.csv\"), parse_dates=['timestamp'])\n",
    "\n",
    "# use only first 10k\n",
    "transactions_data = transactions_data.sort_values(by='source', axis=0)[:10000]\n",
    "\n",
    "# Adjust the samples timestamp for the past 2 days\n",
    "transactions_data = adjust_data_timespan(transactions_data, new_period='2d')\n",
    "\n",
    "# Sorting after adjusting timestamps\n",
    "transactions_data = transactions_data.sort_values(by='timestamp', axis=0)\n",
    "\n",
    "# Preview\n",
    "transactions_data.head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Defining the transactions feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import MLRun's Feature Store\n",
    "import mlrun.feature_store as fstore\n",
    "from mlrun.feature_store.steps import OneHotEncoder, MapValues, DateExtractor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define the transactions FeatureSet\n",
    "transaction_set = fstore.FeatureSet(\"transactions\", \n",
    "                                 entities=[fstore.Entity(\"source\")], \n",
    "                                 timestamp_key='timestamp', \n",
    "                                 description=\"transactions feature set\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The MLRun feature set pipeline is implemented using a graph (DAG) of multiple\n",
    "connected steps, where the `.to()` operation links one step to the next in line. \n",
    "You can visually inspect the generated pipeline using `.plot()`, and test it with sample data to verify it produces the expected results using .preview()."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define and add value mapping\n",
    "main_categories = [\"es_transportation\", \"es_health\", \"es_otherservices\",\n",
    "       \"es_food\", \"es_hotelservices\", \"es_barsandrestaurants\",\n",
    "       \"es_tech\", \"es_sportsandtoys\", \"es_wellnessandbeauty\",\n",
    "       \"es_hyper\", \"es_fashion\", \"es_home\", \"es_contents\",\n",
    "       \"es_travel\", \"es_leisure\"]\n",
    "\n",
    "# One Hot Encode the newly defined mappings\n",
    "one_hot_encoder_mapping = {'category': main_categories,\n",
    "                           'gender': list(transactions_data.gender.unique())}\n",
    "\n",
    "# Define the graph steps\n",
    "transaction_set.graph\\\n",
    "    .to(DateExtractor(parts = ['hour', 'day_of_week'], timestamp_col = 'timestamp'))\\\n",
    "    .to(MapValues(mapping={'age': {'U': '0'}}, with_original_features=True))\\\n",
    "    .to(OneHotEncoder(mapping=one_hot_encoder_mapping))\n",
    "\n",
    "\n",
    "# Add aggregations for 2, 12, and 24 hour time windows\n",
    "transaction_set.add_aggregation(name='amount',\n",
    "                                column='amount',\n",
    "                                operations=['avg','sum', 'count','max'],\n",
    "                                windows=['2h', '12h', '24h'],\n",
    "                                period='1h')\n",
    "\n",
    "\n",
    "# Add the category aggregations over a 14 day window\n",
    "for category in main_categories:\n",
    "    transaction_set.add_aggregation(name=category,column=f'category_{category}',\n",
    "                                    operations=['sum'], windows=['14d'], period='1d')\n",
    "\n",
    "# Add default (offline-parquet & online-nosql) targets\n",
    "transaction_set.set_targets()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Visualize and test the transactions feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/svg+xml": [
       "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>\n",
       "<!DOCTYPE svg PUBLIC \"-//W3C//DTD SVG 1.1//EN\"\n",
       " \"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd\">\n",
       "<!-- Generated by graphviz version 2.43.0 (0)\n",
       " -->\n",
       "<!-- Title: mlrun&#45;flow Pages: 1 -->\n",
       "<svg width=\"907pt\" height=\"98pt\"\n",
       " viewBox=\"0.00 0.00 907.45 98.00\" xmlns=\"http://www.w3.org/2000/svg\" xmlns:xlink=\"http://www.w3.org/1999/xlink\">\n",
       "<g id=\"graph0\" class=\"graph\" transform=\"scale(1 1) rotate(0) translate(4 94)\">\n",
       "<title>mlrun&#45;flow</title>\n",
       "<polygon fill=\"white\" stroke=\"transparent\" points=\"-4,4 -4,-94 903.45,-94 903.45,4 -4,4\"/>\n",
       "<!-- _start -->\n",
       "<g id=\"node1\" class=\"node\">\n",
       "<title>_start</title>\n",
       "<polygon fill=\"lightgrey\" stroke=\"black\" points=\"38.55,-27.05 40.7,-27.15 42.83,-27.3 44.92,-27.49 46.98,-27.74 48.99,-28.03 50.95,-28.36 52.84,-28.75 54.66,-29.18 56.4,-29.65 58.06,-30.16 59.63,-30.71 61.11,-31.31 62.49,-31.94 63.76,-32.61 64.93,-33.31 65.99,-34.04 66.93,-34.8 67.77,-35.59 68.48,-36.41 69.09,-37.25 69.58,-38.11 69.95,-38.99 70.21,-39.89 70.36,-40.8 70.4,-41.72 70.33,-42.65 70.16,-43.59 69.89,-44.53 69.53,-45.47 69.07,-46.41 68.52,-47.35 67.89,-48.28 67.18,-49.2 66.4,-50.11 65.55,-51.01 64.63,-51.89 63.65,-52.75 62.62,-53.59 61.53,-54.41 60.4,-55.2 59.23,-55.96 58.02,-56.69 56.78,-57.39 55.5,-58.06 54.2,-58.69 52.88,-59.29 51.53,-59.84 50.17,-60.35 48.79,-60.82 47.4,-61.25 46,-61.64 44.59,-61.97 43.17,-62.26 41.75,-62.51 40.32,-62.7 38.89,-62.85 37.45,-62.95 36.02,-63 34.58,-63 33.15,-62.95 31.71,-62.85 30.28,-62.7 28.85,-62.51 27.43,-62.26 26.01,-61.97 24.6,-61.64 23.2,-61.25 21.81,-60.82 20.43,-60.35 19.07,-59.84 17.72,-59.29 16.4,-58.69 15.1,-58.06 13.82,-57.39 12.58,-56.69 11.37,-55.96 10.2,-55.2 9.07,-54.41 7.98,-53.59 6.95,-52.75 5.97,-51.89 5.05,-51.01 4.2,-50.11 3.42,-49.2 2.71,-48.28 2.08,-47.35 1.53,-46.41 1.07,-45.47 0.71,-44.53 0.44,-43.59 0.27,-42.65 0.2,-41.72 0.24,-40.8 0.39,-39.89 0.65,-38.99 1.02,-38.11 1.51,-37.25 2.11,-36.41 2.83,-35.59 3.66,-34.8 4.61,-34.04 5.67,-33.31 6.84,-32.61 8.11,-31.94 9.49,-31.31 10.97,-30.71 12.54,-30.16 14.2,-29.65 15.94,-29.18 17.76,-28.75 19.65,-28.36 21.61,-28.03 23.62,-27.74 25.68,-27.49 27.77,-27.3 29.9,-27.15 32.05,-27.05 34.22,-27 36.38,-27 38.55,-27.05\"/>\n",
       "<text text-anchor=\"middle\" x=\"35.3\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">start</text>\n",
       "</g>\n",
       "<!-- DateExtractor -->\n",
       "<g id=\"node2\" class=\"node\">\n",
       "<title>DateExtractor</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"183.94\" cy=\"-45\" rx=\"77.19\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"183.94\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">DateExtractor</text>\n",
       "</g>\n",
       "<!-- _start&#45;&gt;DateExtractor -->\n",
       "<g id=\"edge1\" class=\"edge\">\n",
       "<title>_start&#45;&gt;DateExtractor</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M69.94,-45C78.02,-45 87.01,-45 96.29,-45\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"96.46,-48.5 106.45,-45 96.45,-41.5 96.46,-48.5\"/>\n",
       "</g>\n",
       "<!-- MapValues -->\n",
       "<g id=\"node3\" class=\"node\">\n",
       "<title>MapValues</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"359.03\" cy=\"-45\" rx=\"61.99\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"359.03\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">MapValues</text>\n",
       "</g>\n",
       "<!-- DateExtractor&#45;&gt;MapValues -->\n",
       "<g id=\"edge2\" class=\"edge\">\n",
       "<title>DateExtractor&#45;&gt;MapValues</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M261.36,-45C269.86,-45 278.49,-45 286.92,-45\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"287.08,-48.5 297.08,-45 287.08,-41.5 287.08,-48.5\"/>\n",
       "</g>\n",
       "<!-- OneHotEncoder -->\n",
       "<g id=\"node4\" class=\"node\">\n",
       "<title>OneHotEncoder</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"541.92\" cy=\"-45\" rx=\"85.29\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"541.92\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">OneHotEncoder</text>\n",
       "</g>\n",
       "<!-- MapValues&#45;&gt;OneHotEncoder -->\n",
       "<g id=\"edge3\" class=\"edge\">\n",
       "<title>MapValues&#45;&gt;OneHotEncoder</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M421.02,-45C429.27,-45 437.9,-45 446.57,-45\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"446.75,-48.5 456.75,-45 446.75,-41.5 446.75,-48.5\"/>\n",
       "</g>\n",
       "<!-- Aggregates -->\n",
       "<g id=\"node5\" class=\"node\">\n",
       "<title>Aggregates</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"726.75\" cy=\"-45\" rx=\"63.89\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"726.75\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">Aggregates</text>\n",
       "</g>\n",
       "<!-- OneHotEncoder&#45;&gt;Aggregates -->\n",
       "<g id=\"edge4\" class=\"edge\">\n",
       "<title>OneHotEncoder&#45;&gt;Aggregates</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M627.11,-45C635.71,-45 644.39,-45 652.85,-45\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"653.04,-48.5 663.04,-45 653.04,-41.5 653.04,-48.5\"/>\n",
       "</g>\n",
       "<!-- parquet/parquet -->\n",
       "<g id=\"node6\" class=\"node\">\n",
       "<title>parquet/parquet</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M899.45,-86.73C899.45,-88.53 883.09,-90 862.95,-90 842.81,-90 826.45,-88.53 826.45,-86.73 826.45,-86.73 826.45,-57.27 826.45,-57.27 826.45,-55.47 842.81,-54 862.95,-54 883.09,-54 899.45,-55.47 899.45,-57.27 899.45,-57.27 899.45,-86.73 899.45,-86.73\"/>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M899.45,-86.73C899.45,-84.92 883.09,-83.45 862.95,-83.45 842.81,-83.45 826.45,-84.92 826.45,-86.73\"/>\n",
       "<text text-anchor=\"middle\" x=\"862.95\" y=\"-68.3\" font-family=\"Times,serif\" font-size=\"14.00\">parquet</text>\n",
       "</g>\n",
       "<!-- Aggregates&#45;&gt;parquet/parquet -->\n",
       "<g id=\"edge5\" class=\"edge\">\n",
       "<title>Aggregates&#45;&gt;parquet/parquet</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M779.25,-55.36C791.37,-57.8 804.21,-60.38 816.06,-62.77\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"815.69,-66.26 826.18,-64.8 817.07,-59.4 815.69,-66.26\"/>\n",
       "</g>\n",
       "<!-- nosql/nosql -->\n",
       "<g id=\"node7\" class=\"node\">\n",
       "<title>nosql/nosql</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M890.45,-32.73C890.45,-34.53 878.12,-36 862.95,-36 847.77,-36 835.45,-34.53 835.45,-32.73 835.45,-32.73 835.45,-3.27 835.45,-3.27 835.45,-1.47 847.77,0 862.95,0 878.12,0 890.45,-1.47 890.45,-3.27 890.45,-3.27 890.45,-32.73 890.45,-32.73\"/>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M890.45,-32.73C890.45,-30.92 878.12,-29.45 862.95,-29.45 847.77,-29.45 835.45,-30.92 835.45,-32.73\"/>\n",
       "<text text-anchor=\"middle\" x=\"862.95\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">nosql</text>\n",
       "</g>\n",
       "<!-- Aggregates&#45;&gt;nosql/nosql -->\n",
       "<g id=\"edge6\" class=\"edge\">\n",
       "<title>Aggregates&#45;&gt;nosql/nosql</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M779.25,-34.64C794.6,-31.55 811.1,-28.23 825.29,-25.38\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"826.05,-28.79 835.16,-23.39 824.67,-21.93 826.05,-28.79\"/>\n",
       "</g>\n",
       "</g>\n",
       "</svg>\n"
      ],
      "text/plain": [
       "<graphviz.graphs.Digraph at 0x7f7b9a5c7820>"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Plot the pipeline so you can see the different steps\n",
    "transaction_set.plot(rankdir=\"LR\", with_targets=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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>amount_sum_2h</th>\n",
       "      <th>amount_sum_12h</th>\n",
       "      <th>amount_sum_24h</th>\n",
       "      <th>amount_count_2h</th>\n",
       "      <th>amount_count_12h</th>\n",
       "      <th>amount_count_24h</th>\n",
       "      <th>amount_max_2h</th>\n",
       "      <th>amount_max_12h</th>\n",
       "      <th>amount_max_24h</th>\n",
       "      <th>amount_avg_2h</th>\n",
       "      <th>...</th>\n",
       "      <th>category_es_contents</th>\n",
       "      <th>category_es_travel</th>\n",
       "      <th>category_es_leisure</th>\n",
       "      <th>amount</th>\n",
       "      <th>fraud</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>target</th>\n",
       "      <th>device</th>\n",
       "      <th>timestamp_hour</th>\n",
       "      <th>timestamp_day_of_week</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>source</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>C1022153336</th>\n",
       "      <td>26.92</td>\n",
       "      <td>26.92</td>\n",
       "      <td>26.92</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>26.92</td>\n",
       "      <td>26.92</td>\n",
       "      <td>26.92</td>\n",
       "      <td>26.920000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>26.92</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:35.908872000</td>\n",
       "      <td>M1823072687</td>\n",
       "      <td>33832bb8607545df97632a7ab02d69c4</td>\n",
       "      <td>7</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1006176917</th>\n",
       "      <td>48.22</td>\n",
       "      <td>48.22</td>\n",
       "      <td>48.22</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>48.22</td>\n",
       "      <td>48.22</td>\n",
       "      <td>48.22</td>\n",
       "      <td>48.220000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>48.22</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:54.202497913</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>fadd829c49e74ffa86c8da3be75ada53</td>\n",
       "      <td>7</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1010936270</th>\n",
       "      <td>17.56</td>\n",
       "      <td>17.56</td>\n",
       "      <td>17.56</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>17.56</td>\n",
       "      <td>17.56</td>\n",
       "      <td>17.56</td>\n",
       "      <td>17.560000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>17.56</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:59.309667939</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>58d0422a50bc40c89d2b4977b2f1beea</td>\n",
       "      <td>7</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1033736586</th>\n",
       "      <td>4.50</td>\n",
       "      <td>4.50</td>\n",
       "      <td>4.50</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>4.50</td>\n",
       "      <td>4.50</td>\n",
       "      <td>4.50</td>\n",
       "      <td>4.500000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>4.50</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:04:11.602419118</td>\n",
       "      <td>M1823072687</td>\n",
       "      <td>30b269ae55984e5584f1dd5f642ac1a3</td>\n",
       "      <td>7</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1019071188</th>\n",
       "      <td>1.83</td>\n",
       "      <td>1.83</td>\n",
       "      <td>1.83</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.83</td>\n",
       "      <td>1.83</td>\n",
       "      <td>1.83</td>\n",
       "      <td>1.830000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1.83</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:04:45.491501001</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>97bee3503a984f59aa6139b59f933c0b</td>\n",
       "      <td>7</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1032715879</th>\n",
       "      <td>82.91</td>\n",
       "      <td>697.94</td>\n",
       "      <td>1681.25</td>\n",
       "      <td>3.0</td>\n",
       "      <td>24.0</td>\n",
       "      <td>56.0</td>\n",
       "      <td>54.55</td>\n",
       "      <td>70.47</td>\n",
       "      <td>85.97</td>\n",
       "      <td>27.636667</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>26.00</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:02:33.822725884</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>dcd0943b34254126a2d79d3ac652f268</td>\n",
       "      <td>7</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1005806982</th>\n",
       "      <td>31.14</td>\n",
       "      <td>31.14</td>\n",
       "      <td>330.61</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>5.0</td>\n",
       "      <td>31.14</td>\n",
       "      <td>31.14</td>\n",
       "      <td>119.50</td>\n",
       "      <td>31.140000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>31.14</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:02:38.186659022</td>\n",
       "      <td>M1741626453</td>\n",
       "      <td>c186fb979a77422fa06296c3ff2813f6</td>\n",
       "      <td>7</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1018783569</th>\n",
       "      <td>283.71</td>\n",
       "      <td>1005.10</td>\n",
       "      <td>1736.71</td>\n",
       "      <td>3.0</td>\n",
       "      <td>23.0</td>\n",
       "      <td>56.0</td>\n",
       "      <td>218.48</td>\n",
       "      <td>218.48</td>\n",
       "      <td>218.48</td>\n",
       "      <td>94.570000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>218.48</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:05.569182602</td>\n",
       "      <td>M1053599405</td>\n",
       "      <td>73bdcf0627aa4cc5b882f397cf0d1177</td>\n",
       "      <td>7</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1014847580</th>\n",
       "      <td>118.22</td>\n",
       "      <td>935.26</td>\n",
       "      <td>2062.69</td>\n",
       "      <td>5.0</td>\n",
       "      <td>31.0</td>\n",
       "      <td>68.0</td>\n",
       "      <td>34.93</td>\n",
       "      <td>79.16</td>\n",
       "      <td>89.85</td>\n",
       "      <td>23.644000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>34.93</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:12.551437958</td>\n",
       "      <td>M1823072687</td>\n",
       "      <td>1b216268a9bd4937912c49d16adb5b67</td>\n",
       "      <td>7</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1023803355</th>\n",
       "      <td>113.41</td>\n",
       "      <td>1063.61</td>\n",
       "      <td>2451.98</td>\n",
       "      <td>2.0</td>\n",
       "      <td>34.0</td>\n",
       "      <td>72.0</td>\n",
       "      <td>77.76</td>\n",
       "      <td>77.76</td>\n",
       "      <td>95.71</td>\n",
       "      <td>56.705000</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>35.65</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:35.908872000</td>\n",
       "      <td>M348934600</td>\n",
       "      <td>e2f1b14f73014143a94f7d9505638134</td>\n",
       "      <td>7</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>10000 rows × 56 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "             amount_sum_2h  amount_sum_12h  amount_sum_24h  amount_count_2h  \\\n",
       "source                                                                        \n",
       "C1022153336          26.92           26.92           26.92              1.0   \n",
       "C1006176917          48.22           48.22           48.22              1.0   \n",
       "C1010936270          17.56           17.56           17.56              1.0   \n",
       "C1033736586           4.50            4.50            4.50              1.0   \n",
       "C1019071188           1.83            1.83            1.83              1.0   \n",
       "...                    ...             ...             ...              ...   \n",
       "C1032715879          82.91          697.94         1681.25              3.0   \n",
       "C1005806982          31.14           31.14          330.61              1.0   \n",
       "C1018783569         283.71         1005.10         1736.71              3.0   \n",
       "C1014847580         118.22          935.26         2062.69              5.0   \n",
       "C1023803355         113.41         1063.61         2451.98              2.0   \n",
       "\n",
       "             amount_count_12h  amount_count_24h  amount_max_2h  \\\n",
       "source                                                           \n",
       "C1022153336               1.0               1.0          26.92   \n",
       "C1006176917               1.0               1.0          48.22   \n",
       "C1010936270               1.0               1.0          17.56   \n",
       "C1033736586               1.0               1.0           4.50   \n",
       "C1019071188               1.0               1.0           1.83   \n",
       "...                       ...               ...            ...   \n",
       "C1032715879              24.0              56.0          54.55   \n",
       "C1005806982               1.0               5.0          31.14   \n",
       "C1018783569              23.0              56.0         218.48   \n",
       "C1014847580              31.0              68.0          34.93   \n",
       "C1023803355              34.0              72.0          77.76   \n",
       "\n",
       "             amount_max_12h  amount_max_24h  amount_avg_2h  ...  \\\n",
       "source                                                      ...   \n",
       "C1022153336           26.92           26.92      26.920000  ...   \n",
       "C1006176917           48.22           48.22      48.220000  ...   \n",
       "C1010936270           17.56           17.56      17.560000  ...   \n",
       "C1033736586            4.50            4.50       4.500000  ...   \n",
       "C1019071188            1.83            1.83       1.830000  ...   \n",
       "...                     ...             ...            ...  ...   \n",
       "C1032715879           70.47           85.97      27.636667  ...   \n",
       "C1005806982           31.14          119.50      31.140000  ...   \n",
       "C1018783569          218.48          218.48      94.570000  ...   \n",
       "C1014847580           79.16           89.85      23.644000  ...   \n",
       "C1023803355           77.76           95.71      56.705000  ...   \n",
       "\n",
       "             category_es_contents  category_es_travel  category_es_leisure  \\\n",
       "source                                                                       \n",
       "C1022153336                     0                   0                    0   \n",
       "C1006176917                     0                   0                    0   \n",
       "C1010936270                     0                   0                    0   \n",
       "C1033736586                     0                   0                    0   \n",
       "C1019071188                     0                   0                    0   \n",
       "...                           ...                 ...                  ...   \n",
       "C1032715879                     0                   0                    0   \n",
       "C1005806982                     0                   0                    0   \n",
       "C1018783569                     0                   0                    0   \n",
       "C1014847580                     0                   0                    0   \n",
       "C1023803355                     0                   0                    0   \n",
       "\n",
       "             amount  fraud                     timestamp       target  \\\n",
       "source                                                                  \n",
       "C1022153336   26.92      0 2024-10-06 07:03:35.908872000  M1823072687   \n",
       "C1006176917   48.22      0 2024-10-06 07:03:54.202497913   M348934600   \n",
       "C1010936270   17.56      0 2024-10-06 07:03:59.309667939   M348934600   \n",
       "C1033736586    4.50      0 2024-10-06 07:04:11.602419118  M1823072687   \n",
       "C1019071188    1.83      0 2024-10-06 07:04:45.491501001   M348934600   \n",
       "...             ...    ...                           ...          ...   \n",
       "C1032715879   26.00      0 2024-10-08 07:02:33.822725884   M348934600   \n",
       "C1005806982   31.14      0 2024-10-08 07:02:38.186659022  M1741626453   \n",
       "C1018783569  218.48      0 2024-10-08 07:03:05.569182602  M1053599405   \n",
       "C1014847580   34.93      0 2024-10-08 07:03:12.551437958  M1823072687   \n",
       "C1023803355   35.65      0 2024-10-08 07:03:35.908872000   M348934600   \n",
       "\n",
       "                                       device  timestamp_hour  \\\n",
       "source                                                          \n",
       "C1022153336  33832bb8607545df97632a7ab02d69c4               7   \n",
       "C1006176917  fadd829c49e74ffa86c8da3be75ada53               7   \n",
       "C1010936270  58d0422a50bc40c89d2b4977b2f1beea               7   \n",
       "C1033736586  30b269ae55984e5584f1dd5f642ac1a3               7   \n",
       "C1019071188  97bee3503a984f59aa6139b59f933c0b               7   \n",
       "...                                       ...             ...   \n",
       "C1032715879  dcd0943b34254126a2d79d3ac652f268               7   \n",
       "C1005806982  c186fb979a77422fa06296c3ff2813f6               7   \n",
       "C1018783569  73bdcf0627aa4cc5b882f397cf0d1177               7   \n",
       "C1014847580  1b216268a9bd4937912c49d16adb5b67               7   \n",
       "C1023803355  e2f1b14f73014143a94f7d9505638134               7   \n",
       "\n",
       "             timestamp_day_of_week  \n",
       "source                              \n",
       "C1022153336                      6  \n",
       "C1006176917                      6  \n",
       "C1010936270                      6  \n",
       "C1033736586                      6  \n",
       "C1019071188                      6  \n",
       "...                            ...  \n",
       "C1032715879                      1  \n",
       "C1005806982                      1  \n",
       "C1018783569                      1  \n",
       "C1014847580                      1  \n",
       "C1023803355                      1  \n",
       "\n",
       "[10000 rows x 56 columns]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Preview the resulting features from ingesting sample data (transactions_data)\n",
    "transaction_set.preview(transactions_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After performing the ingestion process, you can see all of the different features that were created with the help of the UI, asshown in the image below."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Features Catalog - fraud prevention](images/features-catalog-transaction.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building the User Events Data Pipeline (Feature Set)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The events data pipeline is simpler and has a single transformation for encoding the event category (using one-hot encoding). The next part demonstrates defining the\n",
    "user events feature set, adding the transformation, and testing the feature set with sample data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define, visualize and test the events feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "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>source</th>\n",
       "      <th>event</th>\n",
       "      <th>timestamp</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>45553</th>\n",
       "      <td>C137986193</td>\n",
       "      <td>password_change</td>\n",
       "      <td>2024-10-06 07:03:46.328233000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>24134</th>\n",
       "      <td>C1940951230</td>\n",
       "      <td>details_change</td>\n",
       "      <td>2024-10-06 07:03:47.376415091</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>64444</th>\n",
       "      <td>C247537602</td>\n",
       "      <td>login</td>\n",
       "      <td>2024-10-06 07:03:49.031198103</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            source            event                     timestamp\n",
       "45553   C137986193  password_change 2024-10-06 07:03:46.328233000\n",
       "24134  C1940951230   details_change 2024-10-06 07:03:47.376415091\n",
       "64444   C247537602            login 2024-10-06 07:03:49.031198103"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Fetch the user_events dataset from the server\n",
    "user_events_data = pd.read_csv('https://s3.wasabisys.com/iguazio/data/fraud-demo-mlrun-fs-docs/events.csv', \n",
    "                               index_col=0, quotechar=\"\\'\", parse_dates=['timestamp'])\n",
    "\n",
    "# Adjust to the last 2 days to see the latest aggregations in the online feature vectors\n",
    "user_events_data = adjust_data_timespan(user_events_data, new_period='2d')\n",
    "\n",
    "# Preview\n",
    "user_events_data.head(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_events_set = fstore.FeatureSet(\"events\",\n",
    "                           entities=[fstore.Entity(\"source\")],\n",
    "                           timestamp_key='timestamp', \n",
    "                           description=\"user events feature set\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/svg+xml": [
       "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>\n",
       "<!DOCTYPE svg PUBLIC \"-//W3C//DTD SVG 1.1//EN\"\n",
       " \"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd\">\n",
       "<!-- Generated by graphviz version 2.43.0 (0)\n",
       " -->\n",
       "<!-- Title: mlrun&#45;flow Pages: 1 -->\n",
       "<svg width=\"394pt\" height=\"98pt\"\n",
       " viewBox=\"0.00 0.00 393.88 98.00\" xmlns=\"http://www.w3.org/2000/svg\" xmlns:xlink=\"http://www.w3.org/1999/xlink\">\n",
       "<g id=\"graph0\" class=\"graph\" transform=\"scale(1 1) rotate(0) translate(4 94)\">\n",
       "<title>mlrun&#45;flow</title>\n",
       "<polygon fill=\"white\" stroke=\"transparent\" points=\"-4,4 -4,-94 389.88,-94 389.88,4 -4,4\"/>\n",
       "<!-- _start -->\n",
       "<g id=\"node1\" class=\"node\">\n",
       "<title>_start</title>\n",
       "<polygon fill=\"lightgrey\" stroke=\"black\" points=\"38.55,-27.05 40.7,-27.15 42.83,-27.3 44.92,-27.49 46.98,-27.74 48.99,-28.03 50.95,-28.36 52.84,-28.75 54.66,-29.18 56.4,-29.65 58.06,-30.16 59.63,-30.71 61.11,-31.31 62.49,-31.94 63.76,-32.61 64.93,-33.31 65.99,-34.04 66.93,-34.8 67.77,-35.59 68.48,-36.41 69.09,-37.25 69.58,-38.11 69.95,-38.99 70.21,-39.89 70.36,-40.8 70.4,-41.72 70.33,-42.65 70.16,-43.59 69.89,-44.53 69.53,-45.47 69.07,-46.41 68.52,-47.35 67.89,-48.28 67.18,-49.2 66.4,-50.11 65.55,-51.01 64.63,-51.89 63.65,-52.75 62.62,-53.59 61.53,-54.41 60.4,-55.2 59.23,-55.96 58.02,-56.69 56.78,-57.39 55.5,-58.06 54.2,-58.69 52.88,-59.29 51.53,-59.84 50.17,-60.35 48.79,-60.82 47.4,-61.25 46,-61.64 44.59,-61.97 43.17,-62.26 41.75,-62.51 40.32,-62.7 38.89,-62.85 37.45,-62.95 36.02,-63 34.58,-63 33.15,-62.95 31.71,-62.85 30.28,-62.7 28.85,-62.51 27.43,-62.26 26.01,-61.97 24.6,-61.64 23.2,-61.25 21.81,-60.82 20.43,-60.35 19.07,-59.84 17.72,-59.29 16.4,-58.69 15.1,-58.06 13.82,-57.39 12.58,-56.69 11.37,-55.96 10.2,-55.2 9.07,-54.41 7.98,-53.59 6.95,-52.75 5.97,-51.89 5.05,-51.01 4.2,-50.11 3.42,-49.2 2.71,-48.28 2.08,-47.35 1.53,-46.41 1.07,-45.47 0.71,-44.53 0.44,-43.59 0.27,-42.65 0.2,-41.72 0.24,-40.8 0.39,-39.89 0.65,-38.99 1.02,-38.11 1.51,-37.25 2.11,-36.41 2.83,-35.59 3.66,-34.8 4.61,-34.04 5.67,-33.31 6.84,-32.61 8.11,-31.94 9.49,-31.31 10.97,-30.71 12.54,-30.16 14.2,-29.65 15.94,-29.18 17.76,-28.75 19.65,-28.36 21.61,-28.03 23.62,-27.74 25.68,-27.49 27.77,-27.3 29.9,-27.15 32.05,-27.05 34.22,-27 36.38,-27 38.55,-27.05\"/>\n",
       "<text text-anchor=\"middle\" x=\"35.3\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">start</text>\n",
       "</g>\n",
       "<!-- OneHotEncoder -->\n",
       "<g id=\"node2\" class=\"node\">\n",
       "<title>OneHotEncoder</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"191.74\" cy=\"-45\" rx=\"85.29\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"191.74\" y=\"-41.3\" font-family=\"Times,serif\" font-size=\"14.00\">OneHotEncoder</text>\n",
       "</g>\n",
       "<!-- _start&#45;&gt;OneHotEncoder -->\n",
       "<g id=\"edge1\" class=\"edge\">\n",
       "<title>_start&#45;&gt;OneHotEncoder</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M69.81,-45C77.82,-45 86.77,-45 96.08,-45\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"96.29,-48.5 106.29,-45 96.29,-41.5 96.29,-48.5\"/>\n",
       "</g>\n",
       "<!-- parquet/parquet -->\n",
       "<g id=\"node3\" class=\"node\">\n",
       "<title>parquet/parquet</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M385.88,-86.73C385.88,-88.53 369.52,-90 349.38,-90 329.24,-90 312.88,-88.53 312.88,-86.73 312.88,-86.73 312.88,-57.27 312.88,-57.27 312.88,-55.47 329.24,-54 349.38,-54 369.52,-54 385.88,-55.47 385.88,-57.27 385.88,-57.27 385.88,-86.73 385.88,-86.73\"/>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M385.88,-86.73C385.88,-84.92 369.52,-83.45 349.38,-83.45 329.24,-83.45 312.88,-84.92 312.88,-86.73\"/>\n",
       "<text text-anchor=\"middle\" x=\"349.38\" y=\"-68.3\" font-family=\"Times,serif\" font-size=\"14.00\">parquet</text>\n",
       "</g>\n",
       "<!-- OneHotEncoder&#45;&gt;parquet/parquet -->\n",
       "<g id=\"edge2\" class=\"edge\">\n",
       "<title>OneHotEncoder&#45;&gt;parquet/parquet</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M258.3,-56.37C273.13,-58.94 288.57,-61.62 302.39,-64.02\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"302.15,-67.53 312.6,-65.79 303.35,-60.64 302.15,-67.53\"/>\n",
       "</g>\n",
       "<!-- nosql/nosql -->\n",
       "<g id=\"node4\" class=\"node\">\n",
       "<title>nosql/nosql</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M376.88,-32.73C376.88,-34.53 364.56,-36 349.38,-36 334.21,-36 321.88,-34.53 321.88,-32.73 321.88,-32.73 321.88,-3.27 321.88,-3.27 321.88,-1.47 334.21,0 349.38,0 364.56,0 376.88,-1.47 376.88,-3.27 376.88,-3.27 376.88,-32.73 376.88,-32.73\"/>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M376.88,-32.73C376.88,-30.92 364.56,-29.45 349.38,-29.45 334.21,-29.45 321.88,-30.92 321.88,-32.73\"/>\n",
       "<text text-anchor=\"middle\" x=\"349.38\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">nosql</text>\n",
       "</g>\n",
       "<!-- OneHotEncoder&#45;&gt;nosql/nosql -->\n",
       "<g id=\"edge3\" class=\"edge\">\n",
       "<title>OneHotEncoder&#45;&gt;nosql/nosql</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M258.3,-33.63C276.56,-30.46 295.75,-27.13 311.7,-24.36\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"312.42,-27.79 321.68,-22.63 311.22,-20.89 312.42,-27.79\"/>\n",
       "</g>\n",
       "</g>\n",
       "</svg>\n"
      ],
      "text/plain": [
       "<graphviz.graphs.Digraph at 0x7f7b9a273220>"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Define and add value mapping\n",
    "events_mapping = {'event': list(user_events_data.event.unique())}\n",
    "\n",
    "# One Hot Encode\n",
    "user_events_set.graph.to(OneHotEncoder(mapping=events_mapping))\n",
    "\n",
    "# Add default (offline-parquet & online-nosql) targets\n",
    "user_events_set.set_targets()\n",
    "\n",
    "# Plot the pipeline so you can see the different steps\n",
    "user_events_set.plot(rankdir=\"LR\", with_targets=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "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>event_password_change</th>\n",
       "      <th>event_details_change</th>\n",
       "      <th>event_login</th>\n",
       "      <th>timestamp</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>source</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>C137986193</th>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:46.328233000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1940951230</th>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:47.376415091</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C247537602</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2024-10-06 07:03:49.031198103</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C470079617</th>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:50.321647428</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1142118359</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2024-10-06 07:03:51.111939830</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C71938921</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2024-10-08 07:03:36.877782776</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1790426542</th>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:41.948986974</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C662187574</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2024-10-08 07:03:43.369641178</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1461183154</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2024-10-08 07:03:44.067781428</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C129566217</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2024-10-08 07:03:46.328233000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>82092 rows × 4 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "             event_password_change  event_details_change  event_login  \\\n",
       "source                                                                  \n",
       "C137986193                       1                     0            0   \n",
       "C1940951230                      0                     1            0   \n",
       "C247537602                       0                     0            1   \n",
       "C470079617                       1                     0            0   \n",
       "C1142118359                      0                     0            1   \n",
       "...                            ...                   ...          ...   \n",
       "C71938921                        0                     0            1   \n",
       "C1790426542                      1                     0            0   \n",
       "C662187574                       0                     0            1   \n",
       "C1461183154                      0                     0            1   \n",
       "C129566217                       0                     0            1   \n",
       "\n",
       "                                timestamp  \n",
       "source                                     \n",
       "C137986193  2024-10-06 07:03:46.328233000  \n",
       "C1940951230 2024-10-06 07:03:47.376415091  \n",
       "C247537602  2024-10-06 07:03:49.031198103  \n",
       "C470079617  2024-10-06 07:03:50.321647428  \n",
       "C1142118359 2024-10-06 07:03:51.111939830  \n",
       "...                                   ...  \n",
       "C71938921   2024-10-08 07:03:36.877782776  \n",
       "C1790426542 2024-10-08 07:03:41.948986974  \n",
       "C662187574  2024-10-08 07:03:43.369641178  \n",
       "C1461183154 2024-10-08 07:03:44.067781428  \n",
       "C129566217  2024-10-08 07:03:46.328233000  \n",
       "\n",
       "[82092 rows x 4 columns]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_events_set.preview(user_events_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building The Target Labels Data Pipeline (Feature Set)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The third feature set defines how to create the target labels. Target labels are only used in training, so they are only stored in the offline store (during serving, the target value is predicted by the model). The basic pandas engine is used, since real-time processing is unnecessary, and the processing is simple and stateless."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define, visualize, and test the events feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_labels(df):\n",
    "    labels = df[['fraud','timestamp']].copy()\n",
    "    labels = labels.rename(columns={\"fraud\": \"label\"})\n",
    "    labels['timestamp'] = labels['timestamp'].astype(\"datetime64[ms]\")\n",
    "    labels['label'] = labels['label'].astype(int)\n",
    "    return labels"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/svg+xml": [
       "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>\n",
       "<!DOCTYPE svg PUBLIC \"-//W3C//DTD SVG 1.1//EN\"\n",
       " \"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd\">\n",
       "<!-- Generated by graphviz version 2.43.0 (0)\n",
       " -->\n",
       "<!-- Title: mlrun&#45;flow Pages: 1 -->\n",
       "<svg width=\"151pt\" height=\"193pt\"\n",
       " viewBox=\"0.00 0.00 150.99 193.25\" xmlns=\"http://www.w3.org/2000/svg\" xmlns:xlink=\"http://www.w3.org/1999/xlink\">\n",
       "<g id=\"graph0\" class=\"graph\" transform=\"scale(1 1) rotate(0) translate(4 189.25)\">\n",
       "<title>mlrun&#45;flow</title>\n",
       "<polygon fill=\"white\" stroke=\"transparent\" points=\"-4,4 -4,-189.25 146.99,-189.25 146.99,4 -4,4\"/>\n",
       "<!-- _start -->\n",
       "<g id=\"node1\" class=\"node\">\n",
       "<title>_start</title>\n",
       "<polygon fill=\"lightgrey\" stroke=\"black\" points=\"74.74,-149.3 76.89,-149.4 79.02,-149.55 81.12,-149.74 83.18,-149.99 85.19,-150.28 87.14,-150.61 89.03,-151 90.85,-151.43 92.6,-151.9 94.26,-152.41 95.83,-152.96 97.3,-153.56 98.68,-154.19 99.96,-154.86 101.12,-155.56 102.18,-156.29 103.13,-157.05 103.96,-157.84 104.68,-158.66 105.28,-159.5 105.77,-160.36 106.14,-161.24 106.4,-162.14 106.55,-163.05 106.59,-163.97 106.53,-164.9 106.36,-165.84 106.09,-166.78 105.72,-167.72 105.26,-168.66 104.72,-169.6 104.09,-170.53 103.38,-171.45 102.59,-172.36 101.74,-173.26 100.82,-174.14 99.84,-175 98.81,-175.84 97.73,-176.66 96.6,-177.45 95.42,-178.21 94.21,-178.94 92.97,-179.64 91.7,-180.31 90.39,-180.94 89.07,-181.54 87.73,-182.09 86.36,-182.6 84.98,-183.07 83.59,-183.5 82.19,-183.89 80.78,-184.22 79.36,-184.51 77.94,-184.76 76.51,-184.95 75.08,-185.1 73.65,-185.2 72.21,-185.25 70.78,-185.25 69.34,-185.2 67.91,-185.1 66.47,-184.95 65.05,-184.76 63.62,-184.51 62.2,-184.22 60.79,-183.89 59.39,-183.5 58,-183.07 56.62,-182.6 55.26,-182.09 53.92,-181.54 52.59,-180.94 51.29,-180.31 50.02,-179.64 48.77,-178.94 47.56,-178.21 46.39,-177.45 45.26,-176.66 44.18,-175.84 43.14,-175 42.16,-174.14 41.25,-173.26 40.39,-172.36 39.61,-171.45 38.9,-170.53 38.27,-169.6 37.72,-168.66 37.27,-167.72 36.9,-166.78 36.63,-165.84 36.46,-164.9 36.39,-163.97 36.43,-163.05 36.58,-162.14 36.84,-161.24 37.22,-160.36 37.71,-159.5 38.31,-158.66 39.03,-157.84 39.86,-157.05 40.81,-156.29 41.86,-155.56 43.03,-154.86 44.3,-154.19 45.68,-153.56 47.16,-152.96 48.73,-152.41 50.39,-151.9 52.13,-151.43 53.96,-151 55.85,-150.61 57.8,-150.28 59.81,-149.99 61.87,-149.74 63.97,-149.55 66.1,-149.4 68.25,-149.3 70.41,-149.25 72.58,-149.25 74.74,-149.3\"/>\n",
       "<text text-anchor=\"middle\" x=\"71.49\" y=\"-163.55\" font-family=\"Times,serif\" font-size=\"14.00\">start</text>\n",
       "</g>\n",
       "<!-- create_labels -->\n",
       "<g id=\"node2\" class=\"node\">\n",
       "<title>create_labels</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"71.49\" cy=\"-95.25\" rx=\"71.49\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"71.49\" y=\"-91.55\" font-family=\"Times,serif\" font-size=\"14.00\">create_labels</text>\n",
       "</g>\n",
       "<!-- _start&#45;&gt;create_labels -->\n",
       "<g id=\"edge1\" class=\"edge\">\n",
       "<title>_start&#45;&gt;create_labels</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M71.49,-148.95C71.49,-141.23 71.49,-131.96 71.49,-123.36\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"74.99,-123.35 71.49,-113.35 67.99,-123.35 74.99,-123.35\"/>\n",
       "</g>\n",
       "<!-- parquet/labels -->\n",
       "<g id=\"node3\" class=\"node\">\n",
       "<title>parquet/labels</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M100.99,-37.6C100.99,-39.68 87.77,-41.38 71.49,-41.38 55.22,-41.38 41.99,-39.68 41.99,-37.6 41.99,-37.6 41.99,-3.65 41.99,-3.65 41.99,-1.57 55.22,0.13 71.49,0.13 87.77,0.13 100.99,-1.57 100.99,-3.65 100.99,-3.65 100.99,-37.6 100.99,-37.6\"/>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M100.99,-37.6C100.99,-35.52 87.77,-33.83 71.49,-33.83 55.22,-33.83 41.99,-35.52 41.99,-37.6\"/>\n",
       "<text text-anchor=\"start\" x=\"49.99\" y=\"-21.43\" font-family=\"Times,serif\" font-size=\"14.00\">labels</text>\n",
       "<text text-anchor=\"start\" x=\"52.49\" y=\"-12.22\" font-family=\"Times,serif\" font-size=\"8.00\">(parquet)</text>\n",
       "</g>\n",
       "<!-- create_labels&#45;&gt;parquet/labels -->\n",
       "<g id=\"edge2\" class=\"edge\">\n",
       "<title>create_labels&#45;&gt;parquet/labels</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M71.49,-77.04C71.49,-69.4 71.49,-60.2 71.49,-51.5\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"74.99,-51.28 71.49,-41.28 67.99,-51.28 74.99,-51.28\"/>\n",
       "</g>\n",
       "</g>\n",
       "</svg>\n"
      ],
      "text/plain": [
       "<graphviz.graphs.Digraph at 0x7f7b805c28b0>"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from mlrun.datastore import ParquetTarget\n",
    "import os\n",
    "\n",
    "# Define the \"labels\" feature set, use \"pandas\" processing engine\n",
    "labels_set = fstore.FeatureSet(\"labels\", \n",
    "                           entities=[fstore.Entity(\"source\")], \n",
    "                           timestamp_key='timestamp',\n",
    "                           description=\"training labels\",\n",
    "                           engine=\"pandas\")\n",
    "\n",
    "labels_set.graph.to(name=\"create_labels\", handler=create_labels)\n",
    "\n",
    "\n",
    "# specify only Parquet (offline) target since its not used for real-time\n",
    "target = ParquetTarget(name='labels',path=f'v3io:///projects/{project.name}/target.parquet')\n",
    "labels_set.set_targets([target], with_defaults=False)\n",
    "\n",
    "# Visualize the feature set\n",
    "labels_set.plot(with_targets=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "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>label</th>\n",
       "      <th>timestamp</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>source</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>C1022153336</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:35.908</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1006176917</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:54.202</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1010936270</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:03:59.309</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1033736586</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:04:11.602</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1019071188</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-06 07:04:45.491</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1032715879</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:02:33.822</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1005806982</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:02:38.186</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1018783569</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:05.569</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1014847580</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:12.551</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>C1023803355</th>\n",
       "      <td>0</td>\n",
       "      <td>2024-10-08 07:03:35.908</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>10000 rows × 2 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "             label               timestamp\n",
       "source                                    \n",
       "C1022153336      0 2024-10-06 07:03:35.908\n",
       "C1006176917      0 2024-10-06 07:03:54.202\n",
       "C1010936270      0 2024-10-06 07:03:59.309\n",
       "C1033736586      0 2024-10-06 07:04:11.602\n",
       "C1019071188      0 2024-10-06 07:04:45.491\n",
       "...            ...                     ...\n",
       "C1032715879      0 2024-10-08 07:02:33.822\n",
       "C1005806982      0 2024-10-08 07:02:38.186\n",
       "C1018783569      0 2024-10-08 07:03:05.569\n",
       "C1014847580      0 2024-10-08 07:03:12.551\n",
       "C1023803355      0 2024-10-08 07:03:35.908\n",
       "\n",
       "[10000 rows x 2 columns]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "labels_set.preview(transactions_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ingesting Data Into the Feature Store\n",
    "\n",
    "Data must be ingested into the feature store before it can be used in training or serving. There are three ways to ingest data:\n",
    "\n",
    "1. Direct ingestion - ingest the data directly from the client/notebook (interactively) \n",
    "2. Batch/scheduled ingestion - create a service/job whithat that ingests data from the source (e.g. file, DB, ..)\n",
    "3. Real-time/Streaming ingestion - create an online service that accepts real-time events (from a stream, http, etc.) and push them into the feature store\n",
    "\n",
    "Direct and batch ingestion are achieved using the `ingest()` method, while real-time ingestion is done using the `deploy_ingestion_service()` method. Both methods are demonstrated in the following sections. The direct ingestion is great for development and testing while the real-time ingestion is mainly used in production."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Batch data ingestion (for tests and training)\n",
    "\n",
    "To run training or test the serving you need to ingest and transform the input datasets and store the results in the feature store. The simplest way is to use the `ingest()` method and specify the feature-set and the source (Dataframe, file, etc.).\n",
    "\n",
    "You can specify the desired target if you want to overwrite the default behavior. For example, set `targets=[ParquetTarget()]` to specify that the data will only be written to parquet files and will not be written to the NoSQL DB (meaning you cannot run real-time serving).\n",
    "\n",
    "The `ingest()` method has many other args/options, see the [documentation](https://docs.mlrun.org/en/stable/data-prep/ingest-data-fs.html?highlight=ingest) for details.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Ingest your transactions dataset through your defined pipeline\n",
    "transactions_df = fstore.ingest(transaction_set, transactions_data, \n",
    "                 infer_options=fstore.InferOptions.default())\n",
    "\n",
    "# Ingestion of your newly created events feature set\n",
    "events_df = fstore.ingest(user_events_set, user_events_data)\n",
    "\n",
    "# Ingest the labels feature set\n",
    "labels_df = fstore.ingest(labels_set, transactions_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transactions - Real-time data ingestion (for Production)\n",
    "\n",
    "When dealing with real-time aggregation, it's important to be able to update these aggregations in real-time.\n",
    "For this purpose, you create live serving functions that update the online feature store of the `transactions` \n",
    "FeatureSet and `Events` FeatureSet.\n",
    "\n",
    "Using MLRun's `serving` runtime, create a `nuclio` function loaded with your feature set's computational graph definition\n",
    "and an `HttpSource` to define the HTTP trigger.\n",
    "\n",
    "Notice that the implementation below does not require any rewrite of the pipeline logic."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transactions - Deploy real-time feature set ingestion pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create iguazio v3io stream and transactions push API endpoint\n",
    "transaction_stream = f'v3io:///projects/{project.name}/streams/transaction'\n",
    "transaction_pusher = mlrun.datastore.get_stream_pusher(transaction_stream)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[nuclio] 2024-10-08 07:06:23,308 workerAllocationMode was automatically set to 'static' because explicitAckMode is enabled\n",
      "> 2024-10-08 07:06:23,309 [warning] Setting function replicas to 1\n",
      "> 2024-10-08 07:06:23,311 [info] Starting remote function deploy\n",
      "2024-10-08 07:06:23  (info) Deploying function\n",
      "2024-10-08 07:06:23  (info) Building\n",
      "2024-10-08 07:06:24  (info) Staging files and preparing base images\n",
      "2024-10-08 07:06:24  (warn) Using user provided base image, runtime interpreter version is provided by the base image\n",
      "2024-10-08 07:06:24  (info) Building processor image\n",
      "2024-10-08 07:07:19  (info) Build complete\n",
      "2024-10-08 07:07:28  (info) Function deploy complete\n",
      "> 2024-10-08 07:07:34,645 [info] Successfully deployed function: {\"external_invocation_urls\":[\"fraud-demo-felipe-transactions-ingest.default-tenant.app.cust-cs-illl--3-6-0.iguazio-cd2.com/\"],\"internal_invocation_urls\":[\"nuclio-fraud-demo-felipe-transactions-ingest.default-tenant.svc.cluster.local:8080\"]}\n"
     ]
    }
   ],
   "source": [
    "# Define the source stream trigger (use v3io streams)\n",
    "# define the `key` and `time` fields (extracted from the Json message).\n",
    "source = mlrun.datastore.sources.StreamSource(path=transaction_stream , key_field='source', time_field='timestamp')\n",
    "\n",
    "# Deploy the transactions feature set's ingestion service over a real-time (Nuclio) serverless function\n",
    "# you can use the run_config parameter to pass function/service specific configuration\n",
    "transaction_set_endpoint = transaction_set.deploy_ingestion_service(source=source)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transactions - Test the feature set HTTP endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By defining your `transactions` feature set you can now use MLRun and Storey to deploy it as a live endpoint, ready to ingest new data!\n",
    "\n",
    "Using MLRun's `serving` runtime, create a nuclio function loaded with your feature set's computational graph definition \n",
    "and an `HttpSource` to define the HTTP trigger."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'step': 64,\n",
       " 'age': '3',\n",
       " 'gender': 'M',\n",
       " 'zipcodeOri': 28007,\n",
       " 'zipMerchant': 28007,\n",
       " 'category': 'es_sportsandtoys',\n",
       " 'amount': 80.94,\n",
       " 'fraud': 0,\n",
       " 'timestamp': '2024-10-08 07:07:34.697038',\n",
       " 'source': 'C1002658784',\n",
       " 'target': 'M17379832',\n",
       " 'device': '3d9d504623da4cc79e25731795251ccd'}"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import requests\n",
    "import json\n",
    "\n",
    "# Select a sample from the dataset and serialize it to JSON\n",
    "transaction_sample = json.loads(transactions_data.sample(1).to_json(orient='records'))[0]\n",
    "transaction_sample['timestamp'] = str(pd.Timestamp.now())\n",
    "transaction_sample"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'{\"id\": \"fc30989f-d8bd-4189-8a7a-fe4dee32a447\"}'"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Post the sample to the ingestion endpoint\n",
    "requests.post(transaction_set_endpoint[0], json=transaction_sample).text"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### User events - deploy the feature set live endpoint\n",
    "Deploy the events feature set's ingestion service using the feature set and all the previously defined resources."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create iguazio v3io stream and transactions push API endpoint\n",
    "events_stream = f'v3io:///projects/{project.name}/streams/events'\n",
    "events_pusher = mlrun.datastore.get_stream_pusher(events_stream)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[nuclio] 2024-10-08 07:07:34,841 workerAllocationMode was automatically set to 'static' because explicitAckMode is enabled\n",
      "> 2024-10-08 07:07:34,842 [warning] Setting function replicas to 1\n",
      "> 2024-10-08 07:07:34,844 [info] Starting remote function deploy\n",
      "2024-10-08 07:07:35  (info) Deploying function\n",
      "2024-10-08 07:07:35  (info) Building\n",
      "2024-10-08 07:07:35  (info) Staging files and preparing base images\n",
      "2024-10-08 07:07:35  (warn) Using user provided base image, runtime interpreter version is provided by the base image\n",
      "2024-10-08 07:07:35  (info) Building processor image\n",
      "2024-10-08 07:08:40  (info) Build complete\n",
      "2024-10-08 07:08:48  (info) Function deploy complete\n",
      "> 2024-10-08 07:08:56,048 [info] Successfully deployed function: {\"external_invocation_urls\":[\"fraud-demo-felipe-events-ingest.default-tenant.app.cust-cs-illl--3-6-0.iguazio-cd2.com/\"],\"internal_invocation_urls\":[\"nuclio-fraud-demo-felipe-events-ingest.default-tenant.svc.cluster.local:8080\"]}\n"
     ]
    }
   ],
   "source": [
    "# Define the source stream trigger (use v3io streams)\n",
    "# define the `key` and `time` fields (extracted from the Json message).\n",
    "source = mlrun.datastore.sources.StreamSource(path=events_stream , key_field='source', time_field='timestamp')\n",
    "\n",
    "# Deploy the transactions feature set's ingestion service over a real-time (Nuclio) serverless function\n",
    "# you can use the run_config parameter to pass function/service specific configuration\n",
    "events_set_endpoint = user_events_set.deploy_ingestion_service(source=source)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### User Events - Test the feature set HTTP endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'source': 'C1663162700',\n",
       " 'event': 'password_change',\n",
       " 'timestamp': '2024-10-08 07:08:56.098197'}"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Select a sample from the events dataset and serialize it to JSON\n",
    "user_events_sample = json.loads(user_events_data.sample(1).to_json(orient='records'))[0]\n",
    "user_events_sample['timestamp'] = str(pd.Timestamp.now())\n",
    "user_events_sample"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'{\"id\": \"63367c2a-019e-43e0-bd7e-f3285f708eba\"}'"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Post the sample to the ingestion endpoint\n",
    "requests.post(events_set_endpoint[0], json=user_events_sample).text"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Done!\n",
    "\n",
    "You've completed the third part - the data-ingestion with the feature store.\n",
    "Proceed to [Part 4](04-train-test-pipeline.ipynb) to learn how to train an ML model using the feature store data."
   ]
  }
 ],
 "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.5"
  },
  "toc-showtags": false
 },
 "nbformat": 4,
 "nbformat_minor": 4
}