{ "cells": [ { "cell_type": "markdown", "id": "bee9388f-8ffc-4fcd-930f-197ec3c2dd96", "metadata": {}, "source": [ "# Test client" ] }, { "cell_type": "code", "execution_count": 44, "id": "eceb50b4-c516-4224-a0b1-efd31bb78c29", "metadata": {}, "outputs": [], "source": [ "import yaml\n", "def update_username(username):\n", " path = 'client/feature_store.yaml'\n", " with open(path, 'r') as file:\n", " config = yaml.safe_load(file) or {}\n", " config['auth']['username'] = username\n", " with open(path, 'w') as file:\n", " yaml.safe_dump(config, file, default_flow_style=False)" ] }, { "cell_type": "markdown", "id": "08a4020a-10ad-476a-af25-26a09d3d4786", "metadata": {}, "source": [ "# Update test user\n", "Use one of `reader`, `writer`, `batch_admin` or `admin` (password is fixed) as the current `username`." ] }, { "cell_type": "code", "execution_count": 45, "id": "564849f9-c95a-4278-9fa7-fa09694e5d93", "metadata": {}, "outputs": [], "source": [ "username = 'reader'\n", "update_username(username)" ] }, { "cell_type": "code", "execution_count": 46, "id": "6ffb2c42-5a5d-495c-92c5-0729f0144fb8", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "auth:\n", " auth_discovery_url: http://0.0.0.0:9999/realms/rbac_example/.well-known/openid-configuration\n", " client_id: app\n", " client_secret: REDACTED\n", " password: password\n", " type: oidc\n", " username: reader\n", "entity_key_serialization_version: 2\n", "offline_store:\n", " host: localhost\n", " port: 8815\n", " type: remote\n", "online_store:\n", " path: http://localhost:6566\n", " type: remote\n", "project: rbac\n", "registry:\n", " path: localhost:6570\n", " registry_type: remote\n" ] } ], "source": [ "!cat client/feature_store.yaml" ] }, { "cell_type": "markdown", "id": "664b6f52-d8cf-4145-bf7a-fcce111a34da", "metadata": {}, "source": [ "## Updating logger\n", "The following is needed to log in the notebook the output the messages logged by th Feast application." ] }, { "cell_type": "code", "execution_count": 47, "id": "3a6fe206-63f8-486f-88cb-b4e888cb6855", "metadata": {}, "outputs": [], "source": [ "import logging\n", "import sys\n", "from io import StringIO\n", "logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')\n", "logger = logging.getLogger()" ] }, { "cell_type": "markdown", "id": "a1eb1495-1f38-4165-a6a4-26a2087f1635", "metadata": {}, "source": [ "## Setup Feast client\n", "Initialize the Feast store from the [client configuration](./client/feature_store.yaml)" ] }, { "cell_type": "code", "execution_count": 48, "id": "b2292e78-cf30-441c-b67f-36e1f1a81923", "metadata": {}, "outputs": [], "source": [ "from feast.feature_store import FeatureStore" ] }, { "cell_type": "code", "execution_count": 49, "id": "bb653327-9eb3-448f-b320-625337851522", "metadata": {}, "outputs": [], "source": [ "store = FeatureStore(repo_path=\"client\")" ] }, { "cell_type": "markdown", "id": "7e826371-3df5-483a-878d-ce79e8b907e3", "metadata": {}, "source": [ "## Basic validation\n", "Verify the authorization config and run some GET APIs on the registry." ] }, { "cell_type": "code", "execution_count": 50, "id": "a59979af-a438-436d-918c-3174d94ade5b", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Authorization config is: {'auth_discovery_url': 'http://0.0.0.0:9999/realms/rbac_example/.well-known/openid-configuration', 'client_id': 'app', 'client_secret': 'REDACTED', 'password': 'password', 'type': 'oidc', 'username': 'reader'}\n" ] } ], "source": [ "print(f\"Authorization config is: {store.config.auth}\")" ] }, { "cell_type": "code", "execution_count": 51, "id": "bf0af19c-6609-4cb4-86f3-a976528c3966", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Entity found driver\n" ] } ], "source": [ "for e in store.list_entities():\n", " print(f\"Entity found {e.name}\")" ] }, { "cell_type": "code", "execution_count": 52, "id": "0494a65f-64bf-45f0-a772-ee6d8b89c91e", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "FeatureView found driver_hourly_stats of type FeatureView\n", "FeatureView found driver_hourly_stats_fresh of type FeatureView\n", "FeatureView found transformed_conv_rate_fresh of type OnDemandFeatureView\n", "FeatureView found transformed_conv_rate of type OnDemandFeatureView\n" ] } ], "source": [ "for fv in store.list_all_feature_views():\n", " print(f\"FeatureView found {fv.name} of type {type(fv).__name__}\")" ] }, { "cell_type": "code", "execution_count": 53, "id": "0832822f-e954-4d43-a96f-de5cf05acb2b", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "FeatureService found driver_activity_v1 of type FeatureService\n", "FeatureService found driver_activity_v3 of type FeatureService\n", "FeatureService found driver_activity_v2 of type FeatureService\n" ] } ], "source": [ "for fs in store.list_feature_services():\n", " print(f\"FeatureService found {fs.name} of type {type(fs).__name__}\")" ] }, { "cell_type": "code", "execution_count": 54, "id": "98fd0767-4305-4b18-a50b-298fa7103815", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "NAME TYPES NAME_PATTERN ACTIONS ROLES REQUIRED_TAGS\n", "read_permission Project DESCRIBE reader -\n", " FeatureView\n", " OnDemandFeatureView\n", " BatchFeatureView\n", " StreamFeatureView\n", " Entity\n", " FeatureService\n", " DataSource\n", " ValidationReference\n", " SavedDataset\n", " Permission\n", "write_fresh_permission FeatureView .*_fresh WRITE_ONLINE fresh_writer -\n", "offline_permission FeatureView CREATE batch_admin -\n", " OnDemandFeatureView DESCRIBE\n", " FeatureService UPDATE\n", " DELETE\n", " WRITE_OFFLINE\n", " READ_OFFLINE\n", "admin_permission Project CREATE store_admin -\n", " FeatureView DESCRIBE\n", " OnDemandFeatureView UPDATE\n", " BatchFeatureView DELETE\n", " StreamFeatureView READ_ONLINE\n", " Entity READ_OFFLINE\n", " FeatureService WRITE_ONLINE\n", " DataSource WRITE_OFFLINE\n", " ValidationReference\n", " SavedDataset\n", " Permission\n" ] } ], "source": [ "!feast -c client permissions list" ] }, { "cell_type": "markdown", "id": "ad2d56ee-e7a9-463e-a597-932c10f8df1c", "metadata": {}, "source": [ "## Validating with test_workflow.py\n", "The following test functions were copied from the `test_workflow.py` template but we added `try` blocks to print only \n", "the relevant error messages, since we expect to receive errors from the permission enforcement modules." ] }, { "cell_type": "code", "execution_count": 55, "id": "930f7e8c-c2a0-4425-99c2-c9958a5a7632", "metadata": {}, "outputs": [], "source": [ "import subprocess\n", "from datetime import datetime\n", "\n", "import pandas as pd\n", "\n", "from feast import FeatureStore\n", "from feast.data_source import PushMode\n", "\n", "def fetch_historical_features_entity_df(store: FeatureStore, for_batch_scoring: bool):\n", " # Note: see https://docs.feast.dev/getting-started/concepts/feature-retrieval for more details on how to retrieve\n", " # for all entities in the offline store instead\n", " entity_df = pd.DataFrame.from_dict(\n", " {\n", " # entity's join key -> entity values\n", " \"driver_id\": [1001, 1002, 1003],\n", " # \"event_timestamp\" (reserved key) -> timestamps\n", " \"event_timestamp\": [\n", " datetime(2021, 4, 12, 10, 59, 42),\n", " datetime(2021, 4, 12, 8, 12, 10),\n", " datetime(2021, 4, 12, 16, 40, 26),\n", " ],\n", " # (optional) label name -> label values. Feast does not process these\n", " \"label_driver_reported_satisfaction\": [1, 5, 3],\n", " # values we're using for an on-demand transformation\n", " \"val_to_add\": [1, 2, 3],\n", " \"val_to_add_2\": [10, 20, 30],\n", " }\n", " )\n", " # For batch scoring, we want the latest timestamps\n", " if for_batch_scoring:\n", " entity_df[\"event_timestamp\"] = pd.to_datetime(\"now\", utc=True)\n", "\n", " try:\n", " training_df = store.get_historical_features(\n", " entity_df=entity_df,\n", " features=[\n", " \"driver_hourly_stats:conv_rate\",\n", " \"driver_hourly_stats:acc_rate\",\n", " \"driver_hourly_stats:avg_daily_trips\",\n", " \"transformed_conv_rate:conv_rate_plus_val1\",\n", " \"transformed_conv_rate:conv_rate_plus_val2\",\n", " ],\n", " ).to_df()\n", " print(training_df.head())\n", " except Exception as e:\n", " print(f\"Failed to run `store.get_historical_features`: {e}\")\n", "\n", "\n", "def fetch_online_features(store, source: str = \"\"):\n", " entity_rows = [\n", " # {join_key: entity_value}\n", " {\n", " \"driver_id\": 1001,\n", " \"val_to_add\": 1000,\n", " \"val_to_add_2\": 2000,\n", " },\n", " {\n", " \"driver_id\": 1002,\n", " \"val_to_add\": 1001,\n", " \"val_to_add_2\": 2002,\n", " },\n", " ]\n", " if source == \"feature_service\":\n", " try:\n", " features_to_fetch = store.get_feature_service(\"driver_activity_v1\")\n", " except Exception as e:\n", " print(f\"Failed to run `store.get_feature_service`: {e}\")\n", " elif source == \"push\":\n", " try:\n", " features_to_fetch = store.get_feature_service(\"driver_activity_v3\")\n", " except Exception as e:\n", " print(f\"Failed to run `store.get_feature_service`: {e}\")\n", " else:\n", " features_to_fetch = [\n", " \"driver_hourly_stats:acc_rate\",\n", " \"transformed_conv_rate:conv_rate_plus_val1\",\n", " \"transformed_conv_rate:conv_rate_plus_val2\",\n", " ]\n", " try:\n", " returned_features = store.get_online_features(\n", " features=features_to_fetch,\n", " entity_rows=entity_rows,\n", " ).to_dict()\n", " for key, value in sorted(returned_features.items()):\n", " print(key, \" : \", value)\n", " except Exception as e:\n", " print(f\"Failed to run `store.get_online_features`: {e}\")" ] }, { "cell_type": "code", "execution_count": 56, "id": "86359ae5-e723-4976-89bb-e772f597ed60", "metadata": {}, "outputs": [], "source": [ "store = FeatureStore(repo_path=\"client\")" ] }, { "cell_type": "markdown", "id": "c0fed355-a1ac-4515-ae27-9d0feca886f4", "metadata": {}, "source": [ "### Historical features" ] }, { "cell_type": "code", "execution_count": 57, "id": "e18dba03-6199-4b48-a9cb-23e3fa51a505", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "--- Historical features for training ---\n", "Failed to run `store.get_historical_features`: Permission error:\n", "Permission offline_permission denied execution of ['READ_OFFLINE'] to FeatureView:driver_hourly_stats: Requires roles ['batch_admin'],Permission admin_permission denied execution of ['READ_OFFLINE'] to FeatureView:driver_hourly_stats: Requires roles ['store_admin']. Detail: Python exception: FeastPermissionError. gRPC client debug context: UNKNOWN:Error received from peer ipv6:%5B::1%5D:8815 {grpc_message:\"Permission error:\\nPermission offline_permission denied execution of [\\'READ_OFFLINE\\'] to FeatureView:driver_hourly_stats: Requires roles [\\'batch_admin\\'],Permission admin_permission denied execution of [\\'READ_OFFLINE\\'] to FeatureView:driver_hourly_stats: Requires roles [\\'store_admin\\']. Detail: Python exception: FeastPermissionError\", grpc_status:2, created_time:\"2024-09-09T08:52:22.529654+02:00\"}. Client context: IOError: Server never sent a data message. Detail: Internal\n", "\n", "--- Historical features for batch scoring ---\n", "Failed to run `store.get_historical_features`: Permission error:\n", "Permission offline_permission denied execution of ['READ_OFFLINE'] to FeatureView:driver_hourly_stats: Requires roles ['batch_admin'],Permission admin_permission denied execution of ['READ_OFFLINE'] to FeatureView:driver_hourly_stats: Requires roles ['store_admin']. Detail: Python exception: FeastPermissionError. gRPC client debug context: UNKNOWN:Error received from peer ipv6:%5B::1%5D:8815 {created_time:\"2024-09-09T08:52:23.51953+02:00\", grpc_status:2, grpc_message:\"Permission error:\\nPermission offline_permission denied execution of [\\'READ_OFFLINE\\'] to FeatureView:driver_hourly_stats: Requires roles [\\'batch_admin\\'],Permission admin_permission denied execution of [\\'READ_OFFLINE\\'] to FeatureView:driver_hourly_stats: Requires roles [\\'store_admin\\']. Detail: Python exception: FeastPermissionError\"}. Client context: IOError: Server never sent a data message. Detail: Internal\n" ] } ], "source": [ "print(\"\\n--- Historical features for training ---\")\n", "fetch_historical_features_entity_df(store, for_batch_scoring=False)\n", "\n", "print(\"\\n--- Historical features for batch scoring ---\")\n", "fetch_historical_features_entity_df(store, for_batch_scoring=True)" ] }, { "cell_type": "markdown", "id": "83bdd1a1-7071-4c51-bf69-9b2bade572a1", "metadata": {}, "source": [ "### Materialization" ] }, { "cell_type": "code", "execution_count": 58, "id": "baeed80c-d2bf-4ac2-ae97-dc689c32e797", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "--- Load features into online store ---\n", "Materializing \u001b[1m\u001b[32m2\u001b[0m feature views to \u001b[1m\u001b[32m2024-09-09 08:52:23+02:00\u001b[0m into the \u001b[1m\u001b[32mremote\u001b[0m online store.\n", "\n", "\u001b[1m\u001b[32mdriver_hourly_stats\u001b[0m from \u001b[1m\u001b[32m2024-09-09 10:50:53+02:00\u001b[0m to \u001b[1m\u001b[32m2024-09-09 08:52:23+02:00\u001b[0m:\n", "Failed to run `store.materialize_incremental`: Permission error:\n", "Permission admin_permission denied execution of ['READ_OFFLINE'] to FileSource:driver_hourly_stats_source: Requires roles ['store_admin']. Detail: Python exception: FeastPermissionError. gRPC client debug context: UNKNOWN:Error received from peer ipv6:%5B::1%5D:8815 {created_time:\"2024-09-09T08:52:24.551895+02:00\", grpc_status:2, grpc_message:\"Permission error:\\nPermission admin_permission denied execution of [\\'READ_OFFLINE\\'] to FileSource:driver_hourly_stats_source: Requires roles [\\'store_admin\\']. Detail: Python exception: FeastPermissionError\"}. Client context: IOError: Server never sent a data message. Detail: Internal\n" ] } ], "source": [ "print(\"\\n--- Load features into online store ---\")\n", "try:\n", " store.materialize_incremental(end_date=datetime.now())\n", "except Exception as e:\n", " print(f\"Failed to run `store.materialize_incremental`: {e}\")" ] }, { "cell_type": "markdown", "id": "f3ef1e87-a98e-447e-893a-d10e205d87c5", "metadata": {}, "source": [ "### Online features" ] }, { "cell_type": "code", "execution_count": 59, "id": "feb552de-77da-4177-bc4e-4c882ca91fe8", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "--- Online features ---\n", "Failed to run `store.get_online_features`: Permission error:\n", "Permission admin_permission denied execution of ['READ_ONLINE'] to FeatureView:driver_hourly_stats: Requires roles ['store_admin']\n", "\n", "--- Online features retrieved (instead) through a feature service---\n", "Failed to run `store.get_online_features`: Permission error:\n", "Permission admin_permission denied execution of ['READ_ONLINE'] to FeatureView:driver_hourly_stats: Requires roles ['store_admin']\n", "\n", "--- Online features retrieved (using feature service v3, which uses a feature view with a push source---\n", "Failed to run `store.get_online_features`: Permission error:\n", "Permission admin_permission denied execution of ['READ_ONLINE'] to FeatureView:driver_hourly_stats: Requires roles ['store_admin']\n" ] } ], "source": [ "print(\"\\n--- Online features ---\")\n", "fetch_online_features(store)\n", "\n", "print(\"\\n--- Online features retrieved (instead) through a feature service---\")\n", "fetch_online_features(store, source=\"feature_service\")\n", "\n", "print(\n", " \"\\n--- Online features retrieved (using feature service v3, which uses a feature view with a push source---\"\n", ")\n", "fetch_online_features(store, source=\"push\")" ] }, { "cell_type": "markdown", "id": "7ce5704c-86ef-4d00-a111-b86e853f2cca", "metadata": {}, "source": [ "### Stream push" ] }, { "cell_type": "code", "execution_count": 60, "id": "e53317fc-8e6b-4dc3-89ca-28d6be04b98a", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "--- Simulate a stream event ingestion of the hourly stats df ---\n", " driver_id event_timestamp created conv_rate \\\n", "0 1001 2024-09-09 08:52:33.038542 2024-09-09 08:52:33.038547 1.0 \n", "\n", " acc_rate avg_daily_trips \n", "0 1.0 1000 \n", "Failed to run `store.push`: \n", "\n", "--- Online features again with updated values from a stream push---\n", "Failed to run `store.get_online_features`: Permission error:\n", "Permission admin_permission denied execution of ['READ_ONLINE'] to FeatureView:driver_hourly_stats: Requires roles ['store_admin']\n" ] } ], "source": [ "print(\"\\n--- Simulate a stream event ingestion of the hourly stats df ---\")\n", "event_df = pd.DataFrame.from_dict(\n", " {\n", " \"driver_id\": [1001],\n", " \"event_timestamp\": [\n", " datetime.now(),\n", " ],\n", " \"created\": [\n", " datetime.now(),\n", " ],\n", " \"conv_rate\": [1.0],\n", " \"acc_rate\": [1.0],\n", " \"avg_daily_trips\": [1000],\n", " }\n", ")\n", "print(event_df)\n", "try:\n", " store.push(\"driver_stats_push_source\", event_df, to=PushMode.ONLINE_AND_OFFLINE)\n", "except Exception as e:\n", " print(f\"Failed to run `store.push`: {e}\") \n", "\n", "print(\"\\n--- Online features again with updated values from a stream push---\")\n", "fetch_online_features(store, source=\"push\")" ] }, { "cell_type": "markdown", "id": "5709f71b-ddff-4048-9db1-98d4090326e1", "metadata": {}, "source": [ "**Note** If you see the following error, it is likely due to the issue [#4392: Remote registry client does not map application errors](https://github.com/feast-dev/feast/issues/4392):\n", "```\n", "Feature view driver_hourly_stats_fresh does not exist in project rbac\n", "```" ] }, { "cell_type": "code", "execution_count": null, "id": "573d9e29-4ba8-41f4-b6a1-82a24d4550b5", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.9" } }, "nbformat": 4, "nbformat_minor": 5 }