From 986b1bc54db55c88af8bada350fa93181d41b245 Mon Sep 17 00:00:00 2001 From: Khor Shu Heng Date: Thu, 22 Oct 2020 10:52:18 +0800 Subject: [PATCH] Add more explanations to the demo notebook, use local file system instead of GCS by default Signed-off-by: Khor Shu Heng --- examples/minimal/Feast 101.ipynb | 904 ++++++++++++++---------- infra/docker-compose/docker-compose.yml | 3 +- 2 files changed, 543 insertions(+), 364 deletions(-) diff --git a/examples/minimal/Feast 101.ipynb b/examples/minimal/Feast 101.ipynb index 196ad4fa2c7..bfe1dc3315c 100644 --- a/examples/minimal/Feast 101.ipynb +++ b/examples/minimal/Feast 101.ipynb @@ -18,31 +18,17 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "## Setup" + "## Introduction" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "0. Install docker, kubernetes (minikube or Docker Desktop), helm\n", - "1. git clone https://github.com/feast-dev/feast && cd feast\n", - "2. kubectl create secret generic feast-postgresql --from-literal=postgresql-password=password\n", - "3. kubectl create secret generic feast-gcp-service-account --from-file=credentials.json=/path/to/key.json\n", - "4. helm install demo infra/charts/feast --values infra/charts/feast/values-demo.yaml\n", - "5. kubectl get pods\n", - "```\n", - "NAME READY STATUS RESTARTS AGE\n", - "demo-feast-core-7f75dc4d48-dzxhb 1/1 Running 1 24m\n", - "demo-feast-jupyter-66bd6bc54f-fjxvh 1/1 Running 0 24m\n", - "demo-feast-online-serving-68d89cc996-xvxrj 1/1 Running 4 24m\n", - "demo-postgresql-0 1/1 Running 0 24m\n", - "demo-prometheus-statsd-exporter-799f847b6b-6472n 1/1 Running 0 24m\n", - "demo-redis-master-0 1/1 Running 0 24m\n", - "demo-redis-slave-0 1/1 Running 0 24m\n", - "demo-redis-slave-1 1/1 Running 0 22m\n", - "```\n", - "6. kubectl port-forward demo-feast-jupyter-66bd6bc54f-fjxvh 8888:8888" + "For this quick start, we will:\n", + "1. Register two driver features, one for driver statistics, the other for driver trips. Driver statistics are updated on daily basis, whereas driver trips are updated in real time.\n", + "2. Creates a driver dataset, then use Feast SDK to retrieve the features corresponding to these drivers from an offline store.\n", + "3. Store the features in an online store (Redis), and retrieve the features via Feast SDK." ] }, { @@ -60,44 +46,69 @@ ] }, { - "cell_type": "code", - "execution_count": null, + "cell_type": "markdown", "metadata": {}, - "outputs": [], "source": [ - "import os" + "Configurations can be provided in three different methods:" ] }, { "cell_type": "code", - "execution_count": 54, + "execution_count": null, "metadata": {}, "outputs": [], "source": [ - "# os.environ['FEAST_SPARK_LAUNCHER'] = 'standalone'\n", - "# os.environ['FEAST_SPARK_STANDALONE_MASTER'] = 'local[*]'\n", - "# os.environ['FEAST_SPARK_HOME'] = os.path.dirname(pyspark.__file__)\n", - "# os.environ['FEAST_SPARK_EXTRA_OPTIONS'] = '--jars https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop2-latest.jar'\\\n", - "# ' --conf spark.hadoop.fs.gs.impl=com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem'" + "# Using environmental variables\n", + "# os.environ[\"FEAST_CORE_URL\"] = \"core:6565\"\n", + "# os.environ[\"FEAST_SERVING_URL\"] = \"online_serving:6566\"\n", + "\n", + "# Provide a map during client initialization\n", + "# options = {\n", + "# \"FEAST_CORE_URL\": \"core:6565\",\n", + "# \"FEAST_SERVING_URL\": \"online_serving:6566\", \n", + "# }\n", + "# client = Client(options)\n", + "\n", + "# As keyword arguments, without the `FEAST` prefix\n", + "# client = Client(core_url=\"core:6565\", serving_url=\"online_serving:6566\")" ] }, { - "cell_type": "code", - "execution_count": null, + "cell_type": "markdown", "metadata": {}, - "outputs": [], "source": [ - "os.environ['FEAST_SPARK_STAGING_LOCATION'] = \"gs://feast-templocation-kf-feast/demo/staging/\"\n", - "os.environ['FEAST_HISTORICAL_FEATURE_OUTPUT_LOCATION'] = \"gs://feast-templocation-kf-feast/demo/output\"" + "If you are following the quick start guide, all required configurations to follow the remainder of the tutorial should have been setup, in the form of environmental variables, as showned below. The configuration values may differ depending on the environment. For a full list of configurable values and explanation, please refer to the user guide." ] }, { "cell_type": "code", - "execution_count": 4, + "execution_count": 1, "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "{'FEAST_CORE_URL': 'core:6565',\n", + " 'FEAST_HISTORICAL_FEATURE_OUTPUT_FORMAT': 'parquet',\n", + " 'FEAST_HISTORICAL_FEATURE_OUTPUT_LOCATION': 'file:///tmp/historical_feature_output',\n", + " 'FEAST_REDIS_HOST': 'redis',\n", + " 'FEAST_SERVING_URL': 'online_serving:6566',\n", + " 'FEAST_SPARK_EXTRA_OPTIONS': '--jars '\n", + " 'https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop2-latest.jar '\n", + " '--conf '\n", + " 'spark.hadoop.fs.gs.impl=com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem',\n", + " 'FEAST_SPARK_HOME': '/usr/local/spark',\n", + " 'FEAST_SPARK_LAUNCHER': 'standalone',\n", + " 'FEAST_SPARK_STAGING_LOCATION': 'file:///tmp/staging',\n", + " 'FEAST_SPARK_STANDALONE_MASTER': 'local'}\n" + ] + } + ], "source": [ - "# os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = \"/path/to/key\"" + "import os\n", + "from pprint import pprint\n", + "pprint({key: value for key, value in os.environ.items() if key.startswith(\"FEAST_\")})" ] }, { @@ -109,7 +120,7 @@ }, { "cell_type": "code", - "execution_count": 71, + "execution_count": 2, "metadata": {}, "outputs": [], "source": [ @@ -122,31 +133,30 @@ }, { "cell_type": "code", - "execution_count": 65, + "execution_count": 3, "metadata": {}, "outputs": [], "source": [ - "CORE_HOST = os.getenv(\"DEMO_FEAST_CORE_SERVICE_HOST\", \"localhost\")\n", - "SERVING_HOST = os.getenv(\"DEMO_FEAST_ONLINE_SERVING_SERVICE_HOST\", \"localhost\")\n", - "REDIS_HOST = os.getenv('DEMO_REDIS_MASTER_SERVICE_HOST', 'localhost')\n", - "\n", - "client = Client(\n", - " core_url=f\"{CORE_HOST}:6565\",\n", - " serving_url=f\"{SERVING_HOST}:6566\", \n", - " redis_host=REDIS_HOST\n", - ")" + "client = Client()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "### Declare Features" + "### Declare Features and Entities" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Entity defines the primary key(s) associated with one or more feature tables. The entity must be registered before declaring the associated feature tables. " ] }, { "cell_type": "code", - "execution_count": 66, + "execution_count": 4, "metadata": {}, "outputs": [], "source": [ @@ -155,7 +165,7 @@ }, { "cell_type": "code", - "execution_count": 68, + "execution_count": 5, "metadata": {}, "outputs": [], "source": [ @@ -226,19 +236,44 @@ "```" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Feature tables group the features together and describe how they can be retrieved. The following examples assume that the feature tables are stored on the local file system, and is accessible from the Spark cluster. If you have setup a GCP service account, you may use GCS instead as the file source." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "`batch_source` defines where the historical features are stored. It is also possible to have an optional `stream_source`, which the feature values are delivered continuously.\n", + "\n", + "For now we will define only `batch_source` for both `driver_statistics` and `driver_trips`, and demonstrate the usage of `stream_source` in later part of the tutorial." + ] + }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], - "source": [] + "source": [ + "# Replace this with a GCS bucket accessible by the service account associated with this notebook\n", + "# datalake_gcs_bucket = \"\"" + ] }, { "cell_type": "code", - "execution_count": 69, + "execution_count": 6, "metadata": {}, "outputs": [], "source": [ + "# GCS Source\n", + "# driver_statistics_source_uri = f\"gs://{datalake_gcs_bucket}/driver_statistics\"\n", + "\n", + "# Local File System Source\n", + "driver_statistics_source_uri = f\"file:///home/jovyan/driver_statistics\"\n", + "\n", "driver_statistics = FeatureTable(\n", " name = \"driver_statistics\",\n", " entities = [\"driver_id\"],\n", @@ -251,7 +286,7 @@ " event_timestamp_column=\"datetime\",\n", " created_timestamp_column=\"created\",\n", " file_format=ParquetFormat(),\n", - " file_url=\"gs://feast-demo-data-lake/driver_statistics\",\n", + " file_url=driver_statistics_source_uri,\n", " date_partition_column=\"date\"\n", " )\n", ")" @@ -259,10 +294,17 @@ }, { "cell_type": "code", - "execution_count": 70, + "execution_count": 7, "metadata": {}, "outputs": [], "source": [ + "# GCS Source\n", + "# driver_trips_source_uri = f\"gs://{datalake_gcs_bucket}/driver_trips\"\n", + "\n", + "# Local File System Source\n", + "driver_trips_source_uri = f\"file:///home/jovyan/driver_trips\"\n", + "\n", + "\n", "driver_trips = FeatureTable(\n", " name = \"driver_trips\",\n", " entities = [\"driver_id\"],\n", @@ -273,7 +315,7 @@ " event_timestamp_column=\"datetime\",\n", " created_timestamp_column=\"created\",\n", " file_format=ParquetFormat(),\n", - " file_url=\"gs://feast-demo-data-lake/driver_trips\",\n", + " file_url=driver_trips_source_uri,\n", " date_partition_column=\"date\"\n", " )\n", ")" @@ -288,7 +330,7 @@ }, { "cell_type": "code", - "execution_count": 11, + "execution_count": 8, "metadata": {}, "outputs": [], "source": [ @@ -299,7 +341,7 @@ }, { "cell_type": "code", - "execution_count": 12, + "execution_count": 9, "metadata": {}, "outputs": [ { @@ -311,12 +353,12 @@ " entities:\n", " - driver_id\n", " features:\n", - " - name: acc_rate\n", - " valueType: FLOAT\n", " - name: conv_rate\n", " valueType: FLOAT\n", " - name: avg_daily_trips\n", " valueType: INT32\n", + " - name: acc_rate\n", + " valueType: FLOAT\n", " batchSource:\n", " type: BATCH_FILE\n", " eventTimestampColumn: datetime\n", @@ -325,9 +367,9 @@ " fileOptions:\n", " fileFormat:\n", " parquetFormat: {}\n", - " fileUrl: gs://feast-demo-data-lake/driver_statistics\n", + " fileUrl: file:///home/jovyan/driver_statistics\n", "meta:\n", - " createdTimestamp: '2020-10-20T06:52:16Z'\n", + " createdTimestamp: '2020-10-22T02:15:16Z'\n", "\n", "spec:\n", " name: driver_trips\n", @@ -344,9 +386,9 @@ " fileOptions:\n", " fileFormat:\n", " parquetFormat: {}\n", - " fileUrl: gs://feast-demo-data-lake/driver_trips\n", + " fileUrl: file:///home/jovyan/driver_trips\n", "meta:\n", - " createdTimestamp: '2020-10-20T14:01:44Z'\n", + " createdTimestamp: '2020-10-22T02:15:16Z'\n", "\n" ] } @@ -363,9 +405,16 @@ "### Populating batch source" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Feast is agnostic to how the batch source is populated, as long as it complies to the Feature Table specification. Therefore, any existing ETL tools can be used for the purpose of data ingestion. Alternatively, you can also use Feast SDK to ingest a Panda Dataframe to the batch source." + ] + }, { "cell_type": "code", - "execution_count": 13, + "execution_count": 10, "metadata": {}, "outputs": [], "source": [ @@ -376,7 +425,7 @@ }, { "cell_type": "code", - "execution_count": 14, + "execution_count": 11, "metadata": {}, "outputs": [], "source": [ @@ -386,7 +435,7 @@ }, { "cell_type": "code", - "execution_count": 15, + "execution_count": 12, "metadata": {}, "outputs": [], "source": [ @@ -408,7 +457,7 @@ }, { "cell_type": "code", - "execution_count": 16, + "execution_count": 13, "metadata": {}, "outputs": [], "source": [ @@ -431,7 +480,7 @@ }, { "cell_type": "code", - "execution_count": 17, + "execution_count": 14, "metadata": {}, "outputs": [], "source": [ @@ -442,43 +491,7 @@ }, { "cell_type": "code", - "execution_count": 18, - "metadata": {}, - "outputs": [ - { - "data": { - "text/plain": [ - "driver_id int64\n", - "conv_rate float32\n", - "acc_rate float32\n", - "avg_daily_trips int32\n", - "datetime datetime64[ns]\n", - "created datetime64[ns]\n", - "dtype: object" - ] - }, - "execution_count": 18, - "metadata": {}, - "output_type": "execute_result" - } - ], - "source": [ - "stats_df.dtypes" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "#!gsutil -m rm -r 'gs://feast-demo-data-lake/driver_statistics/'\n", - "#!gsutil -m rm -r 'gs://feast-demo-data-lake/driver_trips/'" - ] - }, - { - "cell_type": "code", - "execution_count": 27, + "execution_count": 15, "metadata": {}, "outputs": [ { @@ -498,91 +511,73 @@ ] }, { - "cell_type": "code", - "execution_count": 28, + "cell_type": "markdown", "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-09/0db7e166e0c54d8ca24325df642ae07f.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-10/567b6633d2e645a3af9e5db39d44f120.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-11/fb78ef63c2b14ca093ddd7c700e9abef.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-12/9a36a19e056c4b7b998b6221492e4c6c.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-13/9c50b1c80cda40759da776bbafeef793.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-14/e997f7bf9fc34bbca3198d91a4cbf2fe.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-15/fca72f76472948e6a884632f99233845.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-16/ce5634d71dd346a3963780a0b7bbac0a.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-17/c3dffa56ad164058abb1c3775b22f8fd.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-18/d1f8f71e05974fec9d649e5bae49f7e1.parquet\n", - "gs://feast-demo-data-lake/driver_statistics/date=2020-10-19/7044934bdf1241f5aa090fdce535dc45.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-09/147b217e8a2648c1969e20aa795ed531.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-10/04257e2d12cb4804a24986c4ee91e2da.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-11/d5881c5c85714e459b95c21049daac57.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-12/681d05e35fd14ae494836e3ba9cb370a.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-13/fe315291b8114f3b9e93156d916fd4dc.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-14/40486361bf1842bcbc162d002d103543.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-15/8c98651a0a5444919ba6e50bbb726ec7.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-16/7152dfd71d004a07a84221d26ba39a73.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-17/8997c36d944040ea90699b00aeca5a24.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-18/0380ecff082e4175aa45c845c3ce96c2.parquet\n", - "gs://feast-demo-data-lake/driver_trips/date=2020-10-19/b9546e52a2a74ac0a206cac34b789eec.parquet\n" - ] - } - ], "source": [ - "!gsutil ls 'gs://feast-demo-data-lake/driver_statistics/**'\n", - "!gsutil ls 'gs://feast-demo-data-lake/driver_trips/**'" + "## Historical Retrieval For Training" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Historical Retrieval For Training" + "### Point-in-time correction" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "### Point-in-time correction" + "![Point In Time](./images/pit-2.png)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "![Point In Time](./images/pit-2.png)" + "Feast joins the features to the entities based on the following conditions:\n", + "1. Entity primary key(s) value matches.\n", + "2. Feature event timestamp is the closest match possible to the entity event timestamp,\n", + " but must not be more recent than the entity event timestamp, and the difference must\n", + " not be greater than the maximum age specified in the feature table, unless the maximum age is not specified.\n", + "3. If more than one feature table rows satisfy condition 1 and 2, feature row with the\n", + " most recent created timestamp will be chosen.\n", + "4. If none of the above conditions are satisfied, the feature rows will have null values." ] }, { "cell_type": "code", - "execution_count": 58, + "execution_count": 16, "metadata": {}, "outputs": [], "source": [ "import gcsfs\n", - "from pyarrow.parquet import ParquetDataset" + "from pyarrow.parquet import ParquetDataset\n", + "from urllib.parse import urlparse" ] }, { "cell_type": "code", - "execution_count": 59, + "execution_count": 17, "metadata": {}, "outputs": [], "source": [ - "def read_remote_parquet(path):\n", - " fs = gcsfs.GCSFileSystem()\n", - " files = [\"gs://\" + path for path in gcsfs.GCSFileSystem().glob(path)]\n", - " ds = ParquetDataset(files, filesystem=fs)\n", - " return ds.read().to_pandas()" + "def read_parquet(uri):\n", + " parsed_uri = urlparse(uri)\n", + " if parsed_uri.scheme == \"file\":\n", + " return pd.read_parquet(parsed_uri.path)\n", + " elif parsed_uri.scheme == \"gs\":\n", + " fs = gcsfs.GCSFileSystem()\n", + " files = [\"gs://\" + path for path in gcsfs.GCSFileSystem().glob(uri + '/part-*')]\n", + " ds = ParquetDataset(files, filesystem=fs)\n", + " return ds.read().to_pandas()\n", + " else:\n", + " raise ValueError(\"Unsupported scheme\")" ] }, { "cell_type": "code", - "execution_count": 77, + "execution_count": 18, "metadata": {}, "outputs": [ { @@ -613,53 +608,53 @@ " \n", " \n", " 0\n", - " 330184\n", - " 2020-10-19 09:07:43\n", + " 153811\n", + " 2020-10-19 15:21:20\n", " \n", " \n", " 1\n", - " 333896\n", - " 2020-10-18 21:19:01\n", + " 195236\n", + " 2020-10-18 19:07:53\n", " \n", " \n", " 2\n", - " 522128\n", - " 2020-10-18 00:10:16\n", + " 673969\n", + " 2020-10-18 20:06:02\n", " \n", " \n", " 3\n", - " 789025\n", - " 2020-10-18 19:41:26\n", + " 832049\n", + " 2020-10-18 00:10:00\n", " \n", " \n", " 4\n", - " 836898\n", - " 2020-10-18 19:25:35\n", + " 85798\n", + " 2020-10-18 22:09:50\n", " \n", " \n", " 5\n", - " 61720\n", - " 2020-10-19 15:08:43\n", + " 809726\n", + " 2020-10-19 00:19:19\n", " \n", " \n", " 6\n", - " 43893\n", - " 2020-10-17 17:31:12\n", + " 468897\n", + " 2020-10-18 19:15:25\n", " \n", " \n", " 7\n", - " 390750\n", - " 2020-10-19 05:27:16\n", + " 309579\n", + " 2020-10-19 11:31:34\n", " \n", " \n", " 8\n", - " 99001\n", - " 2020-10-19 03:28:45\n", + " 619146\n", + " 2020-10-19 02:00:57\n", " \n", " \n", " 9\n", - " 794802\n", - " 2020-10-18 15:09:06\n", + " 161665\n", + " 2020-10-18 02:56:24\n", " \n", " \n", "\n", @@ -667,19 +662,19 @@ ], "text/plain": [ " driver_id event_timestamp\n", - "0 330184 2020-10-19 09:07:43\n", - "1 333896 2020-10-18 21:19:01\n", - "2 522128 2020-10-18 00:10:16\n", - "3 789025 2020-10-18 19:41:26\n", - "4 836898 2020-10-18 19:25:35\n", - "5 61720 2020-10-19 15:08:43\n", - "6 43893 2020-10-17 17:31:12\n", - "7 390750 2020-10-19 05:27:16\n", - "8 99001 2020-10-19 03:28:45\n", - "9 794802 2020-10-18 15:09:06" + "0 153811 2020-10-19 15:21:20\n", + "1 195236 2020-10-18 19:07:53\n", + "2 673969 2020-10-18 20:06:02\n", + "3 832049 2020-10-18 00:10:00\n", + "4 85798 2020-10-18 22:09:50\n", + "5 809726 2020-10-19 00:19:19\n", + "6 468897 2020-10-18 19:15:25\n", + "7 309579 2020-10-19 11:31:34\n", + "8 619146 2020-10-19 02:00:57\n", + "9 161665 2020-10-18 02:56:24" ] }, - "execution_count": 77, + "execution_count": 18, "metadata": {}, "output_type": "execute_result" } @@ -696,10 +691,11 @@ }, { "cell_type": "code", - "execution_count": 81, + "execution_count": 19, "metadata": {}, "outputs": [], "source": [ + "# get_historical_features will return immediately once the Spark job has been submitted succesfully.\n", "job = client.get_historical_features(\n", " feature_refs=[\n", " \"driver_statistics:avg_daily_trips\",\n", @@ -713,27 +709,17 @@ }, { "cell_type": "code", - "execution_count": 82, + "execution_count": 20, "metadata": {}, - "outputs": [ - { - "data": { - "text/plain": [ - "'gs://feast-templocation-kf-feast/demo/output'" - ] - }, - "execution_count": 82, - "metadata": {}, - "output_type": "execute_result" - } - ], + "outputs": [], "source": [ - "job.get_output_file_uri()" + "# get_output_file_uri will block until the Spark job is completed.\n", + "output_file_uri = job.get_output_file_uri()" ] }, { "cell_type": "code", - "execution_count": 83, + "execution_count": 21, "metadata": {}, "outputs": [ { @@ -759,159 +745,159 @@ " \n", " driver_id\n", " event_timestamp\n", - " driver_statistics__acc_rate\n", " driver_statistics__conv_rate\n", " driver_statistics__avg_daily_trips\n", + " driver_statistics__acc_rate\n", " driver_trips__trips_today\n", " \n", " \n", " \n", " \n", " 0\n", - " 522128\n", - " 2020-10-18 00:10:16\n", - " 0.013687\n", - " 0.162499\n", - " 892\n", - " 883\n", + " 619146\n", + " 2020-10-19 02:00:57\n", + " 0.772141\n", + " 789\n", + " 0.343955\n", + " 954.0\n", " \n", " \n", " 1\n", - " 330184\n", - " 2020-10-19 09:07:43\n", - " 0.788955\n", - " 0.836066\n", - " 912\n", - " 642\n", + " 153811\n", + " 2020-10-19 15:21:20\n", + " 0.755364\n", + " 993\n", + " 0.758154\n", + " 632.0\n", " \n", " \n", " 2\n", - " 390750\n", - " 2020-10-19 05:27:16\n", - " 0.061170\n", - " 0.715991\n", - " 865\n", - " 119\n", + " 809726\n", + " 2020-10-19 00:19:19\n", + " 0.806590\n", + " 899\n", + " 0.017062\n", + " 414.0\n", " \n", " \n", " 3\n", - " 836898\n", - " 2020-10-18 19:25:35\n", - " 0.882056\n", - " 0.061671\n", - " 155\n", - " 573\n", + " 85798\n", + " 2020-10-18 22:09:50\n", + " 0.564337\n", + " 485\n", + " 0.960244\n", + " 778.0\n", " \n", " \n", " 4\n", - " 61720\n", - " 2020-10-19 15:08:43\n", - " 0.958883\n", - " 0.400128\n", - " 113\n", - " 415\n", + " 832049\n", + " 2020-10-18 00:10:00\n", + " 0.335728\n", + " 275\n", + " 0.125462\n", + " 837.0\n", " \n", " \n", " 5\n", - " 99001\n", - " 2020-10-19 03:28:45\n", - " 0.790018\n", - " 0.855180\n", - " 644\n", - " 972\n", + " 195236\n", + " 2020-10-18 19:07:53\n", + " 0.941593\n", + " 228\n", + " 0.494385\n", + " 719.0\n", " \n", " \n", " 6\n", - " 333896\n", - " 2020-10-18 21:19:01\n", - " 0.315527\n", - " 0.015839\n", - " 275\n", - " 377\n", + " 161665\n", + " 2020-10-18 02:56:24\n", + " 0.844367\n", + " 505\n", + " 0.996184\n", + " 868.0\n", " \n", " \n", " 7\n", - " 43893\n", - " 2020-10-17 17:31:12\n", - " 0.316299\n", - " 0.044608\n", - " 209\n", - " 805\n", + " 468897\n", + " 2020-10-18 19:15:25\n", + " 0.814076\n", + " 136\n", + " 0.763832\n", + " 539.0\n", " \n", " \n", " 8\n", - " 794802\n", - " 2020-10-18 15:09:06\n", - " 0.661202\n", - " 0.471721\n", - " 770\n", - " 403\n", + " 673969\n", + " 2020-10-18 20:06:02\n", + " 0.493867\n", + " 327\n", + " 0.363619\n", + " NaN\n", " \n", " \n", " 9\n", - " 789025\n", - " 2020-10-18 19:41:26\n", - " 0.391900\n", - " 0.729488\n", - " 891\n", - " 43\n", + " 309579\n", + " 2020-10-19 11:31:34\n", + " 0.229143\n", + " 979\n", + " 0.619767\n", + " 867.0\n", " \n", " \n", "\n", "" ], "text/plain": [ - " driver_id event_timestamp driver_statistics__acc_rate \\\n", - "0 522128 2020-10-18 00:10:16 0.013687 \n", - "1 330184 2020-10-19 09:07:43 0.788955 \n", - "2 390750 2020-10-19 05:27:16 0.061170 \n", - "3 836898 2020-10-18 19:25:35 0.882056 \n", - "4 61720 2020-10-19 15:08:43 0.958883 \n", - "5 99001 2020-10-19 03:28:45 0.790018 \n", - "6 333896 2020-10-18 21:19:01 0.315527 \n", - "7 43893 2020-10-17 17:31:12 0.316299 \n", - "8 794802 2020-10-18 15:09:06 0.661202 \n", - "9 789025 2020-10-18 19:41:26 0.391900 \n", + " driver_id event_timestamp driver_statistics__conv_rate \\\n", + "0 619146 2020-10-19 02:00:57 0.772141 \n", + "1 153811 2020-10-19 15:21:20 0.755364 \n", + "2 809726 2020-10-19 00:19:19 0.806590 \n", + "3 85798 2020-10-18 22:09:50 0.564337 \n", + "4 832049 2020-10-18 00:10:00 0.335728 \n", + "5 195236 2020-10-18 19:07:53 0.941593 \n", + "6 161665 2020-10-18 02:56:24 0.844367 \n", + "7 468897 2020-10-18 19:15:25 0.814076 \n", + "8 673969 2020-10-18 20:06:02 0.493867 \n", + "9 309579 2020-10-19 11:31:34 0.229143 \n", "\n", - " driver_statistics__conv_rate driver_statistics__avg_daily_trips \\\n", - "0 0.162499 892 \n", - "1 0.836066 912 \n", - "2 0.715991 865 \n", - "3 0.061671 155 \n", - "4 0.400128 113 \n", - "5 0.855180 644 \n", - "6 0.015839 275 \n", - "7 0.044608 209 \n", - "8 0.471721 770 \n", - "9 0.729488 891 \n", + " driver_statistics__avg_daily_trips driver_statistics__acc_rate \\\n", + "0 789 0.343955 \n", + "1 993 0.758154 \n", + "2 899 0.017062 \n", + "3 485 0.960244 \n", + "4 275 0.125462 \n", + "5 228 0.494385 \n", + "6 505 0.996184 \n", + "7 136 0.763832 \n", + "8 327 0.363619 \n", + "9 979 0.619767 \n", "\n", " driver_trips__trips_today \n", - "0 883 \n", - "1 642 \n", - "2 119 \n", - "3 573 \n", - "4 415 \n", - "5 972 \n", - "6 377 \n", - "7 805 \n", - "8 403 \n", - "9 43 " + "0 954.0 \n", + "1 632.0 \n", + "2 414.0 \n", + "3 778.0 \n", + "4 837.0 \n", + "5 719.0 \n", + "6 868.0 \n", + "7 539.0 \n", + "8 NaN \n", + "9 867.0 " ] }, - "execution_count": 83, + "execution_count": 21, "metadata": {}, "output_type": "execute_result" } ], "source": [ - "read_remote_parquet(job.get_output_file_uri() + '/part-*')" + "read_parquet(output_file_uri)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "... Train your model here ..." + "The retrieved result can now be used for model training." ] }, { @@ -921,9 +907,16 @@ "## Populating Online Storage with Batch Ingestion" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "In order to populate the online storage, we can use Feast SDK to start a Spark batch job which will extract the features from the batch source, then load the features to an online store." + ] + }, { "cell_type": "code", - "execution_count": 62, + "execution_count": 22, "metadata": {}, "outputs": [], "source": [ @@ -936,45 +929,53 @@ }, { "cell_type": "code", - "execution_count": 27, + "execution_count": 23, "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "" + "" ] }, - "execution_count": 27, + "execution_count": 23, "metadata": {}, "output_type": "execute_result" } ], "source": [ + "# It will take some time before the Spark Job is completed\n", "job.get_status()" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Once the job is completed, the SDK can be used to retrieve the result from the online store." + ] + }, { "cell_type": "code", - "execution_count": 30, + "execution_count": 24, "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "[{'driver_id': 416975},\n", - " {'driver_id': 139796},\n", - " {'driver_id': 667201},\n", - " {'driver_id': 459097},\n", - " {'driver_id': 549040},\n", - " {'driver_id': 775871},\n", - " {'driver_id': 232140},\n", - " {'driver_id': 137533},\n", - " {'driver_id': 353207},\n", - " {'driver_id': 258085}]" + "[{'driver_id': 100651},\n", + " {'driver_id': 561506},\n", + " {'driver_id': 304334},\n", + " {'driver_id': 997026},\n", + " {'driver_id': 420223},\n", + " {'driver_id': 546179},\n", + " {'driver_id': 308686},\n", + " {'driver_id': 888397},\n", + " {'driver_id': 731657},\n", + " {'driver_id': 684814}]" ] }, - "execution_count": 30, + "execution_count": 24, "metadata": {}, "output_type": "execute_result" } @@ -987,18 +988,49 @@ }, { "cell_type": "code", - "execution_count": 31, + "execution_count": 25, "metadata": {}, - "outputs": [], + "outputs": [ + { + "data": { + "text/plain": [ + "{'driver_id': [100651,\n", + " 561506,\n", + " 304334,\n", + " 997026,\n", + " 420223,\n", + " 546179,\n", + " 308686,\n", + " 888397,\n", + " 731657,\n", + " 684814],\n", + " 'driver_statistics:avg_daily_trips': [614,\n", + " 71,\n", + " 801,\n", + " 341,\n", + " 158,\n", + " 747,\n", + " 509,\n", + " 568,\n", + " 844,\n", + " 315]}" + ] + }, + "execution_count": 25, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ "features = client.get_online_features(\n", " feature_refs=[\"driver_statistics:avg_daily_trips\"],\n", - " entity_rows=entities_sample).to_dict()" + " entity_rows=entities_sample).to_dict()\n", + "features" ] }, { "cell_type": "code", - "execution_count": 32, + "execution_count": 26, "metadata": {}, "outputs": [ { @@ -1029,53 +1061,53 @@ " \n", " \n", " 0\n", - " 416975\n", - " 526\n", + " 100651\n", + " 614\n", " \n", " \n", " 1\n", - " 139796\n", - " 329\n", + " 561506\n", + " 71\n", " \n", " \n", " 2\n", - " 667201\n", - " 875\n", + " 304334\n", + " 801\n", " \n", " \n", " 3\n", - " 459097\n", - " 260\n", + " 997026\n", + " 341\n", " \n", " \n", " 4\n", - " 549040\n", - " 867\n", + " 420223\n", + " 158\n", " \n", " \n", " 5\n", - " 775871\n", - " 122\n", + " 546179\n", + " 747\n", " \n", " \n", " 6\n", - " 232140\n", - " 699\n", + " 308686\n", + " 509\n", " \n", " \n", " 7\n", - " 137533\n", - " 756\n", + " 888397\n", + " 568\n", " \n", " \n", " 8\n", - " 353207\n", - " 861\n", + " 731657\n", + " 844\n", " \n", " \n", " 9\n", - " 258085\n", - " 441\n", + " 684814\n", + " 315\n", " \n", " \n", "\n", @@ -1083,19 +1115,19 @@ ], "text/plain": [ " driver_id driver_statistics:avg_daily_trips\n", - "0 416975 526\n", - "1 139796 329\n", - "2 667201 875\n", - "3 459097 260\n", - "4 549040 867\n", - "5 775871 122\n", - "6 232140 699\n", - "7 137533 756\n", - "8 353207 861\n", - "9 258085 441" + "0 100651 614\n", + "1 561506 71\n", + "2 304334 801\n", + "3 997026 341\n", + "4 420223 158\n", + "5 546179 747\n", + "6 308686 509\n", + "7 888397 568\n", + "8 731657 844\n", + "9 684814 315" ] }, - "execution_count": 32, + "execution_count": 26, "metadata": {}, "output_type": "execute_result" } @@ -1108,7 +1140,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ - ".. Run your production prediction here .." + "The features can now be used as an input to the trained model." ] }, { @@ -1118,9 +1150,16 @@ "## Ingestion from Streaming (real-time) Source" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "With a streaming source, we can use Feast SDK to launch a Spark streaming job that continuously update the online store. First, we will update `driver_trips` feature table such that a new streaming source is added." + ] + }, { "cell_type": "code", - "execution_count": 33, + "execution_count": 27, "metadata": {}, "outputs": [], "source": [ @@ -1134,16 +1173,17 @@ }, { "cell_type": "code", - "execution_count": 34, + "execution_count": 28, "metadata": {}, "outputs": [], "source": [ + "# Change this to any Kafka broker addresses which is accessible by the spark cluster\n", "KAFKA_BROKER = \"kafka:9092\"" ] }, { "cell_type": "code", - "execution_count": 35, + "execution_count": 29, "metadata": {}, "outputs": [], "source": [ @@ -1163,7 +1203,7 @@ }, { "cell_type": "code", - "execution_count": 36, + "execution_count": 30, "metadata": {}, "outputs": [], "source": [ @@ -1177,9 +1217,16 @@ "client.apply_feature_table(driver_trips)" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Start the streaming job and send avro record to Kafka:" + ] + }, { "cell_type": "code", - "execution_count": 37, + "execution_count": 31, "metadata": {}, "outputs": [], "source": [ @@ -1190,7 +1237,7 @@ }, { "cell_type": "code", - "execution_count": 38, + "execution_count": 32, "metadata": {}, "outputs": [], "source": [ @@ -1210,7 +1257,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 33, "metadata": {}, "outputs": [], "source": [ @@ -1231,18 +1278,38 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 34, "metadata": {}, - "outputs": [], + "outputs": [ + { + "data": { + "text/plain": [ + "[{'driver_id': 684814},\n", + " {'driver_id': 60009},\n", + " {'driver_id': 94674},\n", + " {'driver_id': 977538},\n", + " {'driver_id': 861519},\n", + " {'driver_id': 85798},\n", + " {'driver_id': 594463},\n", + " {'driver_id': 132034},\n", + " {'driver_id': 268644},\n", + " {'driver_id': 369157}]" + ] + }, + "execution_count": 34, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ - "entities_sample = np.random.choice(entities, 10)\n", + "entities_sample = np.random.choice(entities, 10, replace=False)\n", "entities_sample = [{\"driver_id\": e} for e in entities_sample]\n", "entities_sample" ] }, { "cell_type": "code", - "execution_count": null, + "execution_count": 35, "metadata": {}, "outputs": [], "source": [ @@ -1253,19 +1320,130 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 36, "metadata": {}, - "outputs": [], + "outputs": [ + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
driver_iddriver_statistics:avg_daily_tripsdriver_trips:trips_today
0684814315None
160009266None
29467480None
3977538685None
4861519140None
585798485None
6594463315None
7132034188None
8268644849None
9369157129None
\n", + "
" + ], + "text/plain": [ + " driver_id driver_statistics:avg_daily_trips driver_trips:trips_today\n", + "0 684814 315 None\n", + "1 60009 266 None\n", + "2 94674 80 None\n", + "3 977538 685 None\n", + "4 861519 140 None\n", + "5 85798 485 None\n", + "6 594463 315 None\n", + "7 132034 188 None\n", + "8 268644 849 None\n", + "9 369157 129 None" + ] + }, + "execution_count": 36, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ "pd.DataFrame(features)" ] }, { "cell_type": "code", - "execution_count": 40, + "execution_count": 37, "metadata": {}, "outputs": [], "source": [ + "# This will stop the streaming job\n", "job.cancel()" ] }, @@ -1293,9 +1471,9 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.7.9" + "version": "3.7.3" } }, "nbformat": 4, "nbformat_minor": 4 -} +} \ No newline at end of file diff --git a/infra/docker-compose/docker-compose.yml b/infra/docker-compose/docker-compose.yml index 7a4f62cbfc1..44f743eec8e 100644 --- a/infra/docker-compose/docker-compose.yml +++ b/infra/docker-compose/docker-compose.yml @@ -26,13 +26,14 @@ services: - core environment: FEAST_CORE_URL: core:6565 - FEAST_ONLINE_SERVING_URL: online_serving:6566 + FEAST_SERVING_URL: online_serving:6566 FEAST_SPARK_LAUNCHER: standalone FEAST_SPARK_STANDALONE_MASTER: local FEAST_SPARK_HOME: /usr/local/spark FEAST_SPARK_STAGING_LOCATION: file:///tmp/staging FEAST_HISTORICAL_FEATURE_OUTPUT_LOCATION: file:///tmp/historical_feature_output FEAST_HISTORICAL_FEATURE_OUTPUT_FORMAT: parquet + FEAST_REDIS_HOST: redis GOOGLE_APPLICATION_CREDENTIALS: /etc/gcloud/service-accounts/key.json ports: - 8888:8888