{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "## HSFS feature exploration\n", "\n", "In this notebook we are going to walk through how to use the HSFS library to explore feature groups and features in the Hopsworks Feature Store. \n", "\n", "A key component of the Hopsworks feature store is to enable sharing and re-using of features across models and use cases. As such, the HSFS libraries allows user to join features from different feature groups and use them to create training datasets.\n", "Features can be taken also from different feature stores (projects) as long as the user running the notebook has the read access to those.\n", "\n", "![Join](./images/join.svg \"Join\")\n", "\n", "As for the [feature_engineering](./feature_engineering.ipynb) notebook, the first step is to establish a connection with the feature store and retrieve the project feature store handle." ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Starting Spark application\n" ] }, { "data": { "text/html": [ "\n", "
IDYARN Application IDKindStateSpark UIDriver log
5application_1604957327609_0009pysparkidleLinkLink
" ], "text/plain": [ "" ] }, "metadata": {}, "output_type": "display_data" }, { "name": "stdout", "output_type": "stream", "text": [ "SparkSession available as 'spark'.\n", "Connected. Call `.close()` to terminate connection gracefully." ] } ], "source": [ "import hsfs\n", "# Create a connection\n", "connection = hsfs.connection()\n", "# Get the feature store handle for the project's feature store\n", "fs = connection.get_feature_store()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Explore feature groups\n", "\n", "You can interact with the feature groups as if they were Spark dataframe. A feature group object has a `show()` method, to show `n` number of lines, and a `read()` method to read the content of the feature group in a Spark dataframe.\n", "\n", "The first step to do any operation on a feature group is to get its handle from the feature store. The `get_feature_group` method accepts the name of the feature group and an optional parameter with the version you want to select. If you do not provide a version, the APIs will default to version 1" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "VersionWarning: No version provided for getting feature group `sales_fg`, defaulting to `1`." ] } ], "source": [ "sales_fg = fs.get_feature_group(\"sales_fg\")" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+----------------------------+-----+--------------------------+--------------------------+------------+----------+----------+---------------------+------------------------+------------------------------+-------------------------+----------------------+----+\n", "|sales_last_quarter_store_dep|store|sales_last_month_store_dep|sales_last_six_month_store|weekly_sales|is_holiday| date|sales_last_year_store|sales_last_quarter_store|sales_last_six_month_store_dep|sales_last_year_store_dep|sales_last_month_store|dept|\n", "+----------------------------+-----+--------------------------+--------------------------+------------+----------+----------+---------------------+------------------------+------------------------------+-------------------------+----------------------+----+\n", "| 0.0| 20| 0.0| 0.0| 32362.95| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 55|\n", "| 0.0| 20| 0.0| 0.0| 63787.83| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 94|\n", "| 0.0| 20| 0.0| 0.0| 17597.83| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 22|\n", "| 0.0| 20| 0.0| 0.0| 9488.37| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 30|\n", "| 0.0| 20| 0.0| 0.0| 85812.69| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 2|\n", "+----------------------------+-----+--------------------------+--------------------------+------------+----------+----------+---------------------+------------------------+------------------------------+-------------------------+----------------------+----+\n", "only showing top 5 rows" ] } ], "source": [ "sales_fg.show(5)" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+----------------------------+-----+--------------------------+--------------------------+------------+----------+----------+---------------------+------------------------+------------------------------+-------------------------+----------------------+----+\n", "|sales_last_quarter_store_dep|store|sales_last_month_store_dep|sales_last_six_month_store|weekly_sales|is_holiday| date|sales_last_year_store|sales_last_quarter_store|sales_last_six_month_store_dep|sales_last_year_store_dep|sales_last_month_store|dept|\n", "+----------------------------+-----+--------------------------+--------------------------+------------+----------+----------+---------------------+------------------------+------------------------------+-------------------------+----------------------+----+\n", "| 0.0| 20| 0.0| 0.0| 32362.95| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 55|\n", "| 0.0| 20| 0.0| 0.0| 63787.83| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 94|\n", "| 0.0| 20| 0.0| 0.0| 17597.83| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 22|\n", "| 0.0| 20| 0.0| 0.0| 9488.37| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 30|\n", "| 0.0| 20| 0.0| 0.0| 85812.69| false|2010-02-05| 0.0| 0.0| 0.0| 0.0| 0.0| 2|\n", "+----------------------------+-----+--------------------------+--------------------------+------------+----------+----------+---------------------+------------------------+------------------------------+-------------------------+----------------------+----+\n", "only showing top 5 rows\n", "\n", "" ] } ], "source": [ "sales_df = sales_fg.read()\n", "sales_df.filter(\"store == 20\").show(5)\n", "print(type(sales_df))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "You can also inspect the metadata of the feature group. You can, for instance, show the features the feature group is made of and if they are primary or partition keys:" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Name: sales_fg\n", "Description: Sales related features\n", "Features:\n", "sales_last_quarter_store_dep \t Primary: False \t Partition: False\n", "store \t Primary: True \t Partition: False\n", "sales_last_month_store_dep \t Primary: False \t Partition: False\n", "sales_last_six_month_store \t Primary: False \t Partition: False\n", "weekly_sales \t Primary: False \t Partition: False\n", "is_holiday \t Primary: False \t Partition: False\n", "date \t Primary: True \t Partition: False\n", "sales_last_year_store \t Primary: False \t Partition: False\n", "sales_last_quarter_store \t Primary: False \t Partition: False\n", "sales_last_six_month_store_dep \t Primary: False \t Partition: False\n", "sales_last_year_store_dep \t Primary: False \t Partition: False\n", "sales_last_month_store \t Primary: False \t Partition: False\n", "dept \t Primary: True \t Partition: False" ] } ], "source": [ "print(\"Name: {}\".format(sales_fg.name))\n", "print(\"Description: {}\".format(sales_fg.description))\n", "print(\"Features:\")\n", "features = sales_fg.features\n", "for feature in features:\n", " print(\"{:<60} \\t Primary: {} \\t Partition: {}\".format(feature.name, feature.primary, feature.partition))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "If you are interested only in a subset of features, you can use the `select()` method on the feature group object to select a list of features. The `select()` behaves like a feature group, as such, you can call the `.show()` or `.read()` methods on it." ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+----+------------+\n", "|store|dept|weekly_sales|\n", "+-----+----+------------+\n", "| 20| 55| 32362.95|\n", "| 20| 94| 63787.83|\n", "| 20| 22| 17597.83|\n", "| 20| 30| 9488.37|\n", "| 20| 2| 85812.69|\n", "+-----+----+------------+\n", "only showing top 5 rows" ] } ], "source": [ "sales_fg.select(['store', 'dept', 'weekly_sales']).show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "If your feature group is available both online and offline, you can use the `online` option of the `show()` and `read()` methods to specify if you want to read your feature group from online storage." ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+----+------------+\n", "|store|dept|weekly_sales|\n", "+-----+----+------------+\n", "| 24| 8| 51815.65|\n", "| 19| 46| 22333.69|\n", "| 19| 83| 3570.62|\n", "| 22| 10| 17241.23|\n", "| 8| 49| 2580.28|\n", "+-----+----+------------+\n", "only showing top 5 rows" ] } ], "source": [ "sales_fg_3 = fs.get_feature_group('sales_fg', 3)\n", "\n", "sales_fg_3.select(['store', 'dept', 'weekly_sales']).show(5, online=True)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Join Features and Feature Groups\n", "\n", "HSFS provides an API similar to Pandas to join feature groups together and to select features from different feature groups.\n", "The easies query you can write is by selecting all the features from a feature group and join them with all the features of another feature group.\n", "\n", "You can use the `select_all()` method of a feature group to select all its features. HSFS relies on the Hopsworks feature store to identify which features of the two feature groups to use as joining condition. \n", "If you don't specify anything, Hopsworks will use the largest matching subset of primary keys with the same name.\n", "\n", "In the example below, `sales_fg` has `store`, `dept` and `date` as composite primary key while `exogenous_fg` has only `store` and `date`. So Hopsworks will set as joining condition `store` and `date`." ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "VersionWarning: No version provided for getting feature group `sales_fg`, defaulting to `1`.\n", "VersionWarning: No version provided for getting feature group `exogenous_fg`, defaulting to `1`." ] } ], "source": [ "sales_fg = fs.get_feature_group('sales_fg')\n", "exogenous_fg = fs.get_feature_group('exogenous_fg')\n", "\n", "query = sales_fg.select_all().join(exogenous_fg.select_all())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "You can use the query object to create training datasets (see training dataset notebook). You can inspect the query generated by calling the `to_string()` method on it." ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "SELECT `fg0`.`sales_last_quarter_store_dep`, `fg0`.`store`, `fg0`.`sales_last_month_store_dep`, `fg0`.`sales_last_year_store_dep`, `fg0`.`sales_last_month_store`, `fg0`.`dept`, `fg0`.`sales_last_year_store`, `fg0`.`sales_last_quarter_store`, `fg0`.`sales_last_six_month_store_dep`, `fg0`.`sales_last_six_month_store`, `fg0`.`weekly_sales`, `fg0`.`is_holiday`, `fg0`.`date`, `fg1`.`markdown5`, `fg1`.`markdown2`, `fg1`.`fuel_price`, `fg1`.`markdown1`, `fg1`.`markdown4`, `fg1`.`cpi`, CASE WHEN `fg1`.`appended_feature` IS NULL THEN 10.0 ELSE `fg1`.`appended_feature` END `appended_feature`, `fg1`.`temperature`, `fg1`.`markdown3`, `fg1`.`unemployment`, `fg1`.`is_holiday`\n", "FROM `demo_fs_meb10000_featurestore`.`sales_fg_1` `fg0`\n", "INNER JOIN `demo_fs_meb10000_featurestore`.`exogenous_fg_1` `fg1` ON `fg0`.`date` = `fg1`.`date` AND `fg0`.`store` = `fg1`.`store`" ] } ], "source": [ "print(query.to_string())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "As for the feature groups, you can call the `show()` method to inspect the data before generating a training dataset from it. Or you can call the `read()` method to get a Spark DataFrame with the result of the query and apply additional transformations to it." ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+----------------------------+-----+--------------------------+-------------------------+----------------------+----+---------------------+------------------------+------------------------------+--------------------------+------------+----------+----------+---------+---------+----------+---------+---------+-----------+----------------+-----------+---------+------------+----------+\n", "|sales_last_quarter_store_dep|store|sales_last_month_store_dep|sales_last_year_store_dep|sales_last_month_store|dept|sales_last_year_store|sales_last_quarter_store|sales_last_six_month_store_dep|sales_last_six_month_store|weekly_sales|is_holiday| date|markdown5|markdown2|fuel_price|markdown1|markdown4| cpi|appended_feature|temperature|markdown3|unemployment|is_holiday|\n", "+----------------------------+-----+--------------------------+-------------------------+----------------------+----+---------------------+------------------------+------------------------------+--------------------------+------------+----------+----------+---------+---------+----------+---------+---------+-----------+----------------+-----------+---------+------------+----------+\n", "| 0.0| 20| 0.0| 0.0| 0.0| 55| 0.0| 0.0| 0.0| 0.0| 32362.95| false|2010-02-05| NA| NA| 2.784| NA| NA|204.2471935| 10.0| 25.92| NA| 8.187| false|\n", "| 0.0| 20| 0.0| 0.0| 0.0| 94| 0.0| 0.0| 0.0| 0.0| 63787.83| false|2010-02-05| NA| NA| 2.784| NA| NA|204.2471935| 10.0| 25.92| NA| 8.187| false|\n", "| 0.0| 20| 0.0| 0.0| 0.0| 22| 0.0| 0.0| 0.0| 0.0| 17597.83| false|2010-02-05| NA| NA| 2.784| NA| NA|204.2471935| 10.0| 25.92| NA| 8.187| false|\n", "| 0.0| 20| 0.0| 0.0| 0.0| 30| 0.0| 0.0| 0.0| 0.0| 9488.37| false|2010-02-05| NA| NA| 2.784| NA| NA|204.2471935| 10.0| 25.92| NA| 8.187| false|\n", "| 0.0| 20| 0.0| 0.0| 0.0| 2| 0.0| 0.0| 0.0| 0.0| 85812.69| false|2010-02-05| NA| NA| 2.784| NA| NA|204.2471935| 10.0| 25.92| NA| 8.187| false|\n", "+----------------------------+-----+--------------------------+-------------------------+----------------------+----+---------------------+------------------------+------------------------------+--------------------------+------------+----------+----------+---------+---------+----------+---------+---------+-----------+----------------+-----------+---------+------------+----------+\n", "only showing top 5 rows" ] } ], "source": [ "query.show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "As for the `show()` and `read()` method of the feature group, even in the case of a query you can specify against which storage to run the query." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Select only a subset of features\n", "\n", "You can replace the `select_all()` method with the `select([])` method to be able to select only a subset of features from a feature group you want to join:" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+----+------------+----------+\n", "|store|dept|weekly_sales|fuel_price|\n", "+-----+----+------------+----------+\n", "| 20| 55| 32362.95| 2.784|\n", "| 20| 94| 63787.83| 2.784|\n", "| 20| 22| 17597.83| 2.784|\n", "| 20| 30| 9488.37| 2.784|\n", "| 20| 2| 85812.69| 2.784|\n", "+-----+----+------------+----------+\n", "only showing top 5 rows" ] } ], "source": [ "query = sales_fg.select(['store', 'dept', 'weekly_sales'])\\\n", " .join(exogenous_fg.select(['fuel_price']))\n", "query.show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Overwrite the joining key\n", "\n", "If your feature groups don't have a primary key, or if they have different names or if you want to overwrite the joining key, you can pass it as a parameter of the join.\n", "\n", "As in Pandas, if the feature has the same name on both feature groups, then you can use the `on=[]` paramter. If they have different names, then you can use the `left_on=[]` and `right_on=[]` paramters:" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+----+------------+----------+\n", "|store|dept|weekly_sales|fuel_price|\n", "+-----+----+------------+----------+\n", "| 20| 55| 32362.95| 2.784|\n", "| 20| 55| 32362.95| 2.666|\n", "| 20| 55| 32362.95| 2.572|\n", "| 20| 55| 32362.95| 2.962|\n", "| 20| 55| 32362.95| 2.58|\n", "+-----+----+------------+----------+\n", "only showing top 5 rows" ] } ], "source": [ "query = sales_fg.select(['store', 'dept', 'weekly_sales'])\\\n", " .join(exogenous_fg.select(['fuel_price']), on=['date'])\n", "query.show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Overwriting the join type\n", "\n", "By default, the join type between two feature groups is `INNER JOIN`. You can overwrite this behavior by passing the `join_type` parameter to the join method. Valid types are: `INNER, LEFT, RIGHT, FULL, CROSS, LEFT_SEMI_JOIN, COMMA`" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "SELECT `fg0`.`store`, `fg0`.`dept`, `fg0`.`weekly_sales`, `fg1`.`fuel_price`\n", "FROM `demo_fs_meb10000_featurestore`.`sales_fg_1` `fg0`\n", "LEFT JOIN `demo_fs_meb10000_featurestore`.`exogenous_fg_1` `fg1` ON `fg0`.`store` = `fg1`.`store` AND `fg0`.`date` = `fg1`.`date`" ] } ], "source": [ "query = sales_fg.select(['store', 'dept', 'weekly_sales'])\\\n", " .join(exogenous_fg.select(['fuel_price']), join_type=\"left\")\n", "\n", "print(query.to_string())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Join mulitple feature groups\n", "\n", "You can concatenate as many feature gropus as you wish. In the example below the order of execution will be:\n", "\n", " (`sales_fg` <> `store_fg`) <> `exogenous_fg`\n", "\n", "The join paramers you pass in each `join()` method call apply to that specific join. This means that you can concatenate left and right joins.\n", "Please be aware that currently HSFS **does not support** nested join such as: \n", "\n", " `sales_fg` <> (`store_fg` <> `exogenous_fg`)" ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "SELECT `fg0`.`sales_last_quarter_store_dep`, `fg0`.`store`, `fg0`.`sales_last_month_store_dep`, `fg0`.`sales_last_year_store_dep`, `fg0`.`sales_last_month_store`, `fg0`.`dept`, `fg0`.`sales_last_year_store`, `fg0`.`sales_last_quarter_store`, `fg0`.`sales_last_six_month_store_dep`, `fg0`.`sales_last_six_month_store`, `fg0`.`weekly_sales`, `fg0`.`is_holiday`, `fg0`.`date`, `fg1`.`size`, `fg1`.`type`, `fg1`.`num_depts`, `fg2`.`fuel_price`, `fg2`.`unemployment`, `fg2`.`cpi`\n", "FROM `demo_fs_meb10000_featurestore`.`sales_fg_1` `fg0`\n", "INNER JOIN `demo_fs_meb10000_featurestore`.`store_fg_1` `fg1` ON `fg0`.`store` = `fg1`.`store`\n", "INNER JOIN `demo_fs_meb10000_featurestore`.`exogenous_fg_1` `fg2` ON `fg0`.`store` = `fg2`.`store` AND `fg0`.`date` = `fg2`.`date`\n", "VersionWarning: No version provided for getting feature group `store_fg`, defaulting to `1`." ] } ], "source": [ "store_fg = fs.get_feature_group(\"store_fg\")\n", "\n", "query = sales_fg.select_all()\\\n", " .join(store_fg.select_all())\\\n", " .join(exogenous_fg.select(['fuel_price', 'unemployment', 'cpi']))\n", "\n", "print(query.to_string())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Free hand query\n", "\n", "With HSFS you are free of writing skipping entirely the Hopsworks query constructor and write your own query. This functionality can be useful if you need to express more complex queries for your use case. `fs.sql` returns a Spark Dataframe." ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "DataFrame[store: int, type: string, size: int, num_depts: bigint]" ] } ], "source": [ "fs.sql(\"SELECT * FROM `store_fg_1`\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "PySpark", "language": "python", "name": "pysparkkernel" }, "language_info": { "codemirror_mode": { "name": "python", "version": 3 }, "mimetype": "text/x-python", "name": "pyspark", "pygments_lexer": "python3" } }, "nbformat": 4, "nbformat_minor": 4 }