05-real-time-serving-pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Real-time Application Pipeline\n",
    "\n",
    "In this example, we define an application pipeline that accepts a user request, enriches the request with real-time features from the feature store, and feeds the features into a three-legged ensemble that uses the newly trained models.\n",
    "\n",
    "You would typically need to implement and deploy multiple microservices and com‐ plex logic to build such an application pipeline. But, with MLRun, you can define it in a few lines of code and deploy it automatically into elastic serverless functions. In addition, the MLRun serving framework will automatically support real-time feature imputing, model monitoring, and so on without requiring extra coding."
   ]
  },
  {
   "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"
     ]
    }
   ],
   "source": [
    "import mlrun\n",
    "project = mlrun.load_project(\n",
    "    name=\"fraud-demo\",\n",
    "    context=\"./\",\n",
    "    user_project=True,\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining a custom serving class\n",
    "\n",
    "MLRun has many built-in model-serving classes for different frameworks (Sklearn, Xgboost, PyTorch, TensorFlow, ONNX, Hugging Face models, and so on). You can also build your custom model serving class as demonstrated in Example 7-24. The serving class must support the load() method for loading the model and the predict() method for making a prediction. You can read MLRun documentation to see all the hooks and advanced usage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# mlrun: start-code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from cloudpickle import load\n",
    "from mlrun.serving.v2_serving import V2ModelServer\n",
    "\n",
    "class ClassifierModel(V2ModelServer):\n",
    "    \n",
    "    def load(self):\n",
    "        \"\"\"load and initialize the model and/or other elements\"\"\"\n",
    "        model_file, extra_data = self.get_model('.pkl')\n",
    "        self.model = load(open(model_file, 'rb'))\n",
    "        \n",
    "    def predict(self, body: dict) -> list:\n",
    "        \"\"\"Generate model predictions from sample\"\"\"\n",
    "        print(f\"Input -> {body['inputs']}\")\n",
    "        feats = np.asarray(body['inputs'])\n",
    "        result: np.ndarray = self.model.predict(feats)\n",
    "        return result.tolist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# mlrun: end-code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building an Application Pipeline with Enrichment and Ensemble\n",
    "\n",
    "MLRun serving can produce managed real-time serverless pipelines from various tasks, including MLRun models or standard model files. The pipelines use the Nuclio real-time serverless engine, which can be deployed anywhere. Nuclio is a high-performance open-source serverless framework focused on data, I/O, and compute-intensive workloads.\n",
    "\n",
    "The EnrichmentVotingEnsemble router class auto-enriches the request with data from the feature store. The router input accepts a list of inference requests (each request can be a dict or list of incoming features/keys). It enriches the request with data from the specified feature vector (feature_vector_uri), forwards the vector to one or more models in an ensemble, and returns an aggregated prediction value (for example, the average result across the three models).\n",
    "The features can often have null values (None, NaN, Inf). The Enrichment_ routers can substitute the null value with fixed or statistical value per fea ture. This is done through the `impute_policy` parameter, which accepts the impute policy per feature (where * is used to specify the default). The value can be a fixed number for constants or $mean, $max, $min, $std, $count for statistical values to substitute the value with the equivalent feature stats (taken from the feature store).\n",
    "The code in Example 7-24 defines a new serving function with the ClassifierModel class code (in serving.py) and a router topology (using the EnrichmentVotingEnsem ble router class) with three child models."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('transaction_fraud_rf',\n",
       "  {'updated': '2024-10-08 13:21:47.319818+00:00',\n",
       "   'project': 'fraud-demo-felipe',\n",
       "   'key': 'model',\n",
       "   'tree': 'e27f0d6d-8f14-4642-abee-43b9867631ef',\n",
       "   'tag': 'latest',\n",
       "   'labels': {'workflow-id': 'e27f0d6d-8f14-4642-abee-43b9867631ef',\n",
       "    'framework': 'sklearn'},\n",
       "   'iter': 1,\n",
       "   'hash': 'd31c0e672e4ac97438e866612c9fb02c1a3a1732'}),\n",
       " ('transaction_fraud_xgboost',\n",
       "  {'updated': '2024-10-08 13:21:18.851606+00:00',\n",
       "   'project': 'fraud-demo-felipe',\n",
       "   'key': 'model',\n",
       "   'tree': 'e27f0d6d-8f14-4642-abee-43b9867631ef',\n",
       "   'tag': 'latest',\n",
       "   'labels': {'workflow-id': 'e27f0d6d-8f14-4642-abee-43b9867631ef',\n",
       "    'framework': 'sklearn'},\n",
       "   'iter': 2,\n",
       "   'hash': 'd40c64ec2d081899089ec7e34288c87a175a848f'}),\n",
       " ('transaction_fraud_adaboost',\n",
       "  {'updated': '2024-10-08 13:21:22.048244+00:00',\n",
       "   'project': 'fraud-demo-felipe',\n",
       "   'key': 'model',\n",
       "   'tree': 'e27f0d6d-8f14-4642-abee-43b9867631ef',\n",
       "   'tag': 'latest',\n",
       "   'labels': {'workflow-id': 'e27f0d6d-8f14-4642-abee-43b9867631ef',\n",
       "    'framework': 'sklearn'},\n",
       "   'iter': 3,\n",
       "   'hash': 'db02f4acce087779cbc634285ea1647206a5fc84'})]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "[(m.spec.db_key, m.metadata.to_dict()) for m in project.list_models('', tag='latest')]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "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=\"800pt\" height=\"196pt\"\n",
       " viewBox=\"0.00 0.00 799.73 196.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 192)\">\n",
       "<title>mlrun&#45;flow</title>\n",
       "<polygon fill=\"white\" stroke=\"transparent\" points=\"-4,4 -4,-192 795.73,-192 795.73,4 -4,4\"/>\n",
       "<!-- _start -->\n",
       "<g id=\"node1\" class=\"node\">\n",
       "<title>_start</title>\n",
       "<polygon fill=\"lightgrey\" stroke=\"black\" points=\"364.89,-152.05 367.04,-152.15 369.17,-152.3 371.26,-152.49 373.32,-152.74 375.33,-153.03 377.29,-153.36 379.18,-153.75 381,-154.18 382.74,-154.65 384.4,-155.16 385.98,-155.71 387.45,-156.31 388.83,-156.94 390.1,-157.61 391.27,-158.31 392.33,-159.04 393.28,-159.8 394.11,-160.59 394.83,-161.41 395.43,-162.25 395.92,-163.11 396.29,-163.99 396.55,-164.89 396.7,-165.8 396.74,-166.72 396.67,-167.65 396.5,-168.59 396.23,-169.53 395.87,-170.47 395.41,-171.41 394.86,-172.35 394.23,-173.28 393.52,-174.2 392.74,-175.11 391.89,-176.01 390.97,-176.89 389.99,-177.75 388.96,-178.59 387.87,-179.41 386.74,-180.2 385.57,-180.96 384.36,-181.69 383.12,-182.39 381.84,-183.06 380.54,-183.69 379.22,-184.29 377.87,-184.84 376.51,-185.35 375.13,-185.82 373.74,-186.25 372.34,-186.64 370.93,-186.97 369.51,-187.26 368.09,-187.51 366.66,-187.7 365.23,-187.85 363.79,-187.95 362.36,-188 360.92,-188 359.49,-187.95 358.05,-187.85 356.62,-187.7 355.19,-187.51 353.77,-187.26 352.35,-186.97 350.94,-186.64 349.54,-186.25 348.15,-185.82 346.77,-185.35 345.41,-184.84 344.06,-184.29 342.74,-183.69 341.44,-183.06 340.16,-182.39 338.92,-181.69 337.71,-180.96 336.54,-180.2 335.41,-179.41 334.32,-178.59 333.29,-177.75 332.31,-176.89 331.39,-176.01 330.54,-175.11 329.76,-174.2 329.05,-173.28 328.42,-172.35 327.87,-171.41 327.41,-170.47 327.05,-169.53 326.78,-168.59 326.61,-167.65 326.54,-166.72 326.58,-165.8 326.73,-164.89 326.99,-163.99 327.36,-163.11 327.85,-162.25 328.46,-161.41 329.17,-160.59 330.01,-159.8 330.95,-159.04 332.01,-158.31 333.18,-157.61 334.45,-156.94 335.83,-156.31 337.31,-155.71 338.88,-155.16 340.54,-154.65 342.28,-154.18 344.1,-153.75 345.99,-153.36 347.95,-153.03 349.96,-152.74 352.02,-152.49 354.11,-152.3 356.24,-152.15 358.39,-152.05 360.56,-152 362.72,-152 364.89,-152.05\"/>\n",
       "<text text-anchor=\"middle\" x=\"361.64\" y=\"-166.3\" font-family=\"Times,serif\" font-size=\"14.00\">start</text>\n",
       "</g>\n",
       "<g id=\"node2\" class=\"node\">\n",
       "<title></title>\n",
       "<polygon fill=\"none\" stroke=\"black\" points=\"388.64,-86.54 388.64,-101.46 372.82,-112 350.46,-112 334.64,-101.46 334.64,-86.54 350.46,-76 372.82,-76 388.64,-86.54\"/>\n",
       "<polygon fill=\"none\" stroke=\"black\" points=\"392.64,-84.4 392.64,-103.6 374.04,-116 349.25,-116 330.64,-103.6 330.64,-84.4 349.25,-72 374.04,-72 392.64,-84.4\"/>\n",
       "</g>\n",
       "<!-- _start&#45;&gt; -->\n",
       "<g id=\"edge1\" class=\"edge\">\n",
       "<title>_start&#45;&gt;</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M361.64,-151.84C361.64,-144.16 361.64,-134.88 361.64,-126.05\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"365.14,-126.03 361.64,-116.03 358.14,-126.03 365.14,-126.03\"/>\n",
       "</g>\n",
       "<!-- transaction_fraud_rf -->\n",
       "<g id=\"node3\" class=\"node\">\n",
       "<title>transaction_fraud_rf</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"104.64\" cy=\"-18\" rx=\"104.78\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"104.64\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">transaction_fraud_rf</text>\n",
       "</g>\n",
       "<!-- &#45;&gt;transaction_fraud_rf -->\n",
       "<g id=\"edge2\" class=\"edge\">\n",
       "<title>&#45;&gt;transaction_fraud_rf</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M330.9,-84.15C290.22,-72.43 217.59,-51.52 165.59,-36.55\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"166.39,-33.14 155.81,-33.73 164.45,-39.86 166.39,-33.14\"/>\n",
       "</g>\n",
       "<!-- transaction_fraud_xgboost -->\n",
       "<g id=\"node4\" class=\"node\">\n",
       "<title>transaction_fraud_xgboost</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"361.64\" cy=\"-18\" rx=\"133.78\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"361.64\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">transaction_fraud_xgboost</text>\n",
       "</g>\n",
       "<!-- &#45;&gt;transaction_fraud_xgboost -->\n",
       "<g id=\"edge3\" class=\"edge\">\n",
       "<title>&#45;&gt;transaction_fraud_xgboost</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M361.64,-71.99C361.64,-64.06 361.64,-54.91 361.64,-46.48\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"365.14,-46.31 361.64,-36.31 358.14,-46.31 365.14,-46.31\"/>\n",
       "</g>\n",
       "<!-- transaction_fraud_adaboost -->\n",
       "<g id=\"node5\" class=\"node\">\n",
       "<title>transaction_fraud_adaboost</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"652.64\" cy=\"-18\" rx=\"139.18\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"652.64\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">transaction_fraud_adaboost</text>\n",
       "</g>\n",
       "<!-- &#45;&gt;transaction_fraud_adaboost -->\n",
       "<g id=\"edge4\" class=\"edge\">\n",
       "<title>&#45;&gt;transaction_fraud_adaboost</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M392.78,-85.08C437.78,-73.64 522.21,-52.17 582.62,-36.81\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"583.65,-40.16 592.48,-34.3 581.92,-33.37 583.65,-40.16\"/>\n",
       "</g>\n",
       "</g>\n",
       "</svg>\n"
      ],
      "text/plain": [
       "<graphviz.graphs.Digraph at 0x7fc6e54b5520>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create the serving function from your code above\n",
    "serving_fn = project.set_function('src/serving.py', name='test-function',\n",
    "                                  image=\"mlrun/mlrun\", kind=\"serving\")\n",
    "serving_fn.set_topology(\n",
    "    \"router\",\n",
    "    mlrun.serving.routers.EnrichmentVotingEnsemble(\n",
    "        feature_vector_uri=\"short\",\n",
    "        impute_policy={\"*\": \"$mean\"}),\n",
    ")\n",
    "# add the 3 trained models to the Ensemble\n",
    "for model in project.list_models('', tag='latest'):\n",
    "    name = model.spec.db_key\n",
    "    serving_fn.add_model(name, class_name=\"ClassifierModel\", model_path=model.uri)\n",
    "# Plot the ensemble configuration\n",
    "serving_fn.spec.graph.plot()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test the Application Pipeline Locally\n",
    "\n",
    "Before deploying the serving function, you can test it in the current notebook and check the model output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2024-10-08 13:41:43,004 [info] model transaction_fraud_rf was loaded\n",
      "> 2024-10-08 13:41:43,033 [info] model transaction_fraud_xgboost was loaded\n",
      "> 2024-10-08 13:41:43,064 [info] model transaction_fraud_adaboost was loaded\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Trying to unpickle estimator DecisionTreeClassifier from version 1.5.2 when using version 1.5.0. This might lead to breaking code or invalid results. Use at your own risk. For more info please refer to:\n",
      "https://scikit-learn.org/stable/model_persistence.html#security-maintainability-limitations\n",
      "Trying to unpickle estimator RandomForestClassifier from version 1.5.2 when using version 1.5.0. This might lead to breaking code or invalid results. Use at your own risk. For more info please refer to:\n",
      "https://scikit-learn.org/stable/model_persistence.html#security-maintainability-limitations\n",
      "Trying to unpickle estimator LogisticRegression from version 1.5.2 when using version 1.5.0. This might lead to breaking code or invalid results. Use at your own risk. For more info please refer to:\n",
      "https://scikit-learn.org/stable/model_persistence.html#security-maintainability-limitations\n",
      "Trying to unpickle estimator AdaBoostClassifier from version 1.5.2 when using version 1.5.0. This might lead to breaking code or invalid results. Use at your own risk. For more info please refer to:\n",
      "https://scikit-learn.org/stable/model_persistence.html#security-maintainability-limitations\n"
     ]
    }
   ],
   "source": [
    "# Create a mock server from the serving function\n",
    "local_server = serving_fn.to_mock_server()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Input -> [[82.38333699999998, 176.010817, 0.0, 38.79236982490288, 134.16, 417.81, 11.0, 37.982727272727274, 134.16, 1275.0599999999997, 44.0, 28.978636363636358, 90.0, 1.0, 2.0]]\n",
      "Input -> [[82.38333699999998, 176.010817, 0.0, 38.79236982490288, 134.16, 417.81, 11.0, 37.982727272727274, 134.16, 1275.0599999999997, 44.0, 28.978636363636358, 90.0, 1.0, 2.0]]\n",
      "Input -> [[82.38333699999998, 176.010817, 0.0, 38.79236982490288, 134.16, 417.81, 11.0, 37.982727272727274, 134.16, 1275.0599999999997, 44.0, 28.978636363636358, 90.0, 1.0, 2.0]]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "X does not have valid feature names, but RandomForestClassifier was fitted with feature names\n",
      "X does not have valid feature names, but LogisticRegression was fitted with feature names\n",
      "X does not have valid feature names, but AdaBoostClassifier was fitted with feature names\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'id': 'f7f45c8d7e3a4c6993d6e10cc0f91acd',\n",
       " 'model_name': 'VotingEnsemble',\n",
       " 'outputs': [0],\n",
       " 'model_version': 'v1'}"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Choose an id for your test\n",
    "sample_id = 'C1000148617'\n",
    "\n",
    "# Send your sample ID for prediction\n",
    "local_server.test(path='/v2/models/infer',\n",
    "            body={'inputs': [[sample_id]]})\n",
    "\n",
    "# notice the input vector is printed 3 times (once per child model) and is enriched with data from the feature store"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Accessing the real-time feature vector directly\n",
    "\n",
    "If you would like to access the real-time features directly from your application instead of using the EnrichmentVotingEnsemble, you can call the feature store `get_online_feature_service()` method. This method is used internally in the EnrichmentVotingEnsemble router class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'amount_max_2h': 82.38333699999998,\n",
       "  'amount_max_12h': 134.16,\n",
       "  'amount_max_24h': 134.16,\n",
       "  'amount_count_2h': 0.0,\n",
       "  'amount_count_12h': 11.0,\n",
       "  'amount_count_24h': 44.0,\n",
       "  'amount_sum_2h': 176.010817,\n",
       "  'amount_sum_12h': 417.81,\n",
       "  'amount_sum_24h': 1275.0599999999997,\n",
       "  'es_transportation_sum_14d': 90.0,\n",
       "  'es_health_sum_14d': 1.0,\n",
       "  'es_otherservices_sum_14d': 2.0,\n",
       "  'amount_avg_2h': 38.79236982490288,\n",
       "  'amount_avg_12h': 37.982727272727274,\n",
       "  'amount_avg_24h': 28.978636363636358}]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import mlrun.feature_store as fs\n",
    "\n",
    "# Create the online feature service\n",
    "svc = fs.get_feature_vector('short:latest').get_online_feature_service(impute_policy={\"*\": \"$mean\"})\n",
    "\n",
    "# Get sample feature vector\n",
    "sample_fv = svc.get([{'source': sample_id}])\n",
    "sample_fv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploying the function on the Kubernetes cluster\n",
    "\n",
    "You can now deploy the function. Once deployed, you get a function with http trigger that can be called from other locations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2024-10-08 13:41:43,535 [info] Starting remote function deploy\n",
      "2024-10-08 13:41:43  (info) Deploying function\n",
      "2024-10-08 13:41:44  (info) Building\n",
      "2024-10-08 13:41:44  (info) Staging files and preparing base images\n",
      "2024-10-08 13:41:44  (warn) Using user provided base image, runtime interpreter version is provided by the base image\n",
      "2024-10-08 13:41:44  (info) Building processor image\n",
      "2024-10-08 13:42:49  (info) Build complete\n",
      "2024-10-08 13:43:13  (info) Function deploy complete\n",
      "> 2024-10-08 13:43:14,979 [info] Successfully deployed function: {\"external_invocation_urls\":[\"fraud-demo-felipe-test-function.default-tenant.app.cust-cs-illl--3-6-0.iguazio-cd2.com/\"],\"internal_invocation_urls\":[\"nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080\"]}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'http://fraud-demo-felipe-test-function.default-tenant.app.cust-cs-illl--3-6-0.iguazio-cd2.com/'"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import os\n",
    "from mlrun.datastore.datastore_profile import DatastoreProfileV3io\n",
    "\n",
    "# Create and register TSDB profile\n",
    "tsdb_profile = DatastoreProfileV3io(\n",
    "    name=\"my-v3io-tsdb\",\n",
    ")\n",
    "project.register_datastore_profile(tsdb_profile)\n",
    "\n",
    "# Create and register stream profile\n",
    "stream_profile = DatastoreProfileV3io(\n",
    "    name=\"my-v3io-stream\",\n",
    "    v3io_access_key=mlrun.mlconf.get_v3io_access_key(),\n",
    ")\n",
    "project.register_datastore_profile(stream_profile)\n",
    "\n",
    "# Enable model monitoring\n",
    "serving_fn.set_tracking()\n",
    "\n",
    "project.set_model_monitoring_credentials(\n",
    "    tsdb_profile_name=tsdb_profile.name,\n",
    "    stream_profile_name=stream_profile.name,\n",
    ")\n",
    "\n",
    "# Deploy the serving function\n",
    "serving_fn.deploy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test the server\n",
    "\n",
    "You can test the serving function and examine the model output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2024-10-08 13:43:15,027 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'id': 'ee5635d0-bb82-4036-a71c-379696d9ed79',\n",
       " 'model_name': 'VotingEnsemble',\n",
       " 'outputs': [0],\n",
       " 'model_version': 'v1'}"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Choose an id for your test\n",
    "sample_id = 'C1000148617'\n",
    "\n",
    "model_inference_path = '/v2/models/infer'\n",
    "\n",
    "# Send your sample ID for prediction\n",
    "serving_fn.invoke(path='/v2/models/infer',\n",
    "                  body={'inputs': [[sample_id]]})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also directly query the feature store values, which are used in the enrichment."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simulate incoming data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the dataset\n",
    "data = mlrun.get_dataitem(mlrun.get_sample_path(\"data/fraud-demo-mlrun-fs-docs/data.csv\")).as_df()\n",
    "\n",
    "# use only first 10k\n",
    "data = data.sort_values(by='source', axis=0)[:10000]\n",
    "\n",
    "# keys\n",
    "sample_ids = data['source'].to_list()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2024-10-08 13:43:19,205 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': 'efa462b4-36ed-4f63-8d4d-8d6474fff08d', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:20,718 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': 'b930f851-fd9a-498a-8b7f-4e84f3bd42ec', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:21,987 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': 'c064d2e8-377a-4384-9a47-7c5881f83b0a', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:22,928 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': 'd4d6a214-fcf5-4c94-9551-db129c1f52f2', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:23,263 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': '72419b36-0e6e-4ade-8b23-70b3407115d6', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:24,141 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': 'aa27481c-5903-4f02-9ce4-5540fb45d775', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:25,491 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': 'fdc379d7-7a9c-452b-91ac-93b29abaffab', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:26,392 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': '015fbeab-7f98-4295-a05e-a2fb38d54b3a', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:27,417 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': '59733a3f-ffc8-4828-88e7-0e46ff9a129a', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n",
      "> 2024-10-08 13:43:28,915 [info] Invoking function: {\"method\":\"POST\",\"path\":\"http://nuclio-fraud-demo-felipe-test-function.default-tenant.svc.cluster.local:8080/v2/models/infer\"}\n",
      "{'id': '05e984c5-556d-415d-a4f4-e3eb389f2213', 'model_name': 'VotingEnsemble', 'outputs': [0], 'model_version': 'v1'}\n"
     ]
    }
   ],
   "source": [
    "from random import choice, uniform\n",
    "from time import sleep\n",
    "\n",
    "# Sending random requests\n",
    "for _ in range(10):\n",
    "    data_point = choice(sample_ids)\n",
    "    try:\n",
    "        resp = serving_fn.invoke(path=model_inference_path, body={'inputs': [[data_point]]})\n",
    "        print(resp)\n",
    "        sleep(uniform(0.2, 1.7))\n",
    "    except OSError:\n",
    "        pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "## Done!\n",
    "\n",
    "You've completed the fraud-detection demo. \n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}