From 3a5f168117808efbd08970ea64cfe233ee66bc3f Mon Sep 17 00:00:00 2001 From: Yuqing Wei Date: Tue, 14 Jun 2022 16:52:17 +0800 Subject: [PATCH 1/2] upgrade object id query cmd --- docs/quickstart_synapse.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/quickstart_synapse.md b/docs/quickstart_synapse.md index 20a75492f..e3aad375f 100644 --- a/docs/quickstart_synapse.md +++ b/docs/quickstart_synapse.md @@ -23,7 +23,7 @@ First step is to provision required cloud resources if you want to use Feathr. F Feathr has native cloud integration. To use Feathr on Azure, you only need three steps: -1. Get the `Principal ID` of your account by running `az ad signed-in-user show --query objectId -o tsv` in the link below (Select "Bash" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI. +1. Get the `Principal ID` of your account by running `az ad signed-in-user show --query id -o tsv` in the link below (Select "Bash" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI. [Launch Cloud Shell](https://shell.azure.com/bash) From 703f078388ddf783da698189ca454363c5a3eb5f Mon Sep 17 00:00:00 2001 From: Yuqing Wei Date: Tue, 14 Jun 2022 17:07:05 +0800 Subject: [PATCH 2/2] fix all occurance of object id query --- docs/README.md | 2 +- docs/how-to-guides/azure-deployment.md | 2 +- .../azure_resource_provision.json | 636 ++++--- docs/how-to-guides/deployment/deploy.json | 9 +- docs/how-to-guides/deployment/main.bicep | 2 +- .../nyc_driver_demo.ipynb | 1468 ++++++++-------- .../product_recommendation_demo.ipynb | 1506 ++++++++--------- 7 files changed, 1793 insertions(+), 1832 deletions(-) diff --git a/docs/README.md b/docs/README.md index 1e80d55bf..6e26207b7 100644 --- a/docs/README.md +++ b/docs/README.md @@ -33,7 +33,7 @@ Feathr automatically computes your feature values and joins them to your trainin Feathr has native cloud integration. To use Feathr on Azure, you only need three steps: -1. Get the `Principal ID` of your account by running `az ad signed-in-user show --query objectId -o tsv` in the link below (Select "Bash" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI. +1. Get the `Principal ID` of your account by running `az ad signed-in-user show --query id -o tsv` in the link below (Select "Bash" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI. [Launch Cloud Shell](https://shell.azure.com/bash) diff --git a/docs/how-to-guides/azure-deployment.md b/docs/how-to-guides/azure-deployment.md index adf5e2a48..4df308cdb 100644 --- a/docs/how-to-guides/azure-deployment.md +++ b/docs/how-to-guides/azure-deployment.md @@ -13,7 +13,7 @@ Due to the complexity of the possible cloud environment, it is almost impossible Feathr has native cloud integration and getting started with Feathr is very straightforward. You only need three steps: -1. Get the principal ID of your account by running `az ad signed-in-user show --query objectId -o tsv` in the link below (Select "Bash" if you are asked to choose one), and write down that value (will be something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`) +1. Get the principal ID of your account by running `az ad signed-in-user show --query id -o tsv` in the link below (Select "Bash" if you are asked to choose one), and write down that value (will be something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`) [Launch Cloud Shell](https://shell.azure.com/bash) diff --git a/docs/how-to-guides/azure_resource_provision.json b/docs/how-to-guides/azure_resource_provision.json index 33117f60d..ea7c0ad0e 100644 --- a/docs/how-to-guides/azure_resource_provision.json +++ b/docs/how-to-guides/azure_resource_provision.json @@ -1,355 +1,319 @@ { - "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#", - "contentVersion": "1.0.0.0", - "parameters": { - "resourcePrefix": { - "minLength": 3, - "maxLength": 15, - "type": "String", - "metadata": { - "description": "Resource prefix for all the resource provisioned. This should be an alphanumeric string." - } - }, - "principalId": { - "type": "String", - "metadata": { - "description": "Specifies the principal ID assigned to the role. You can find it by logging into 'https://shell.azure.com/bash' and run 'az ad signed-in-user show --query objectId -o tsv'" - } - }, - "allowAllConnections": { - "defaultValue": "true", - "allowedValues": [ - "true", - "false" - ], - "type": "String", - "metadata": { - "description": "Specifies whether to allow client IPs to connect to Synapse" - } - }, - "provisionPurview": { - "defaultValue": "true", - "allowedValues": [ - "true", - "false" - ], - "type": "String", - "metadata": { - "description": "Whether or not put purview in the provision script" - } - }, - "provisionEventHub": { - "defaultValue": "true", - "allowedValues": [ - "true", - "false" - ], - "type": "String", - "metadata": { - "description": "Whether or not to deploy eventhub provision script" - } - } + "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#", + "contentVersion": "1.0.0.0", + "parameters": { + "resourcePrefix": { + "minLength": 3, + "maxLength": 15, + "type": "String", + "metadata": { + "description": "Resource prefix for all the resource provisioned. This should be an alphanumeric string." + } }, - "variables": { - "location": "[resourceGroup().location]", - "tenantId": "[subscription().tenantId]", - "redisCacheName": "[concat(parameters('resourcePrefix'),'redis' )]", - "keyVaultName": "[concat(parameters('resourcePrefix'),'kv')]", - "eventhubNameSpaceName": "[concat(parameters('resourcePrefix'),'ehns')]", - "eventhubName": "[concat(parameters('resourcePrefix'),'eh')]", - "eventhubSku": "Standard", - "eventhubSkuCapacity": 1, - "keyVault": "[resourceId('Microsoft.KeyVault/vaults', variables('keyVaultName'))]", - "redisCache": "[resourceId('Microsoft.Cache/redis', variables('redisCacheName'))]", - "eventhubNameSpace": "[resourceId('Microsoft.EventHub/namespaces/', variables('eventhubNameSpaceName'))]", - "sparkPoolName": "spark31", - "workspaceName": "[toLower(concat(parameters('resourcePrefix'),'syws'))]", - "dlsName": "[toLower(concat(parameters('resourcePrefix'), 'dls'))]", - "dlsFsName": "[toLower(concat(parameters('resourcePrefix'),'fs'))]", - "dlsAccount": "[resourceId('Microsoft.Storage/storageAccounts', variables('dlsName'))]", - "purviewName": "[concat(parameters('resourcePrefix'),'purview' )]", - "roleDefinitionIdForBlobContributor": "ba92f5b4-2d11-453d-a403-e96b0029c9fe", - "roleDefinitionIdForKeyVaultSecretsUser": "4633458b-17de-408a-b874-0445c86b69e6", - "roleAssignmentNameForBlobContributor": "[guid(parameters('principalId'), variables('roleDefinitionIdForBlobContributor'), resourceGroup().id)]", - "roleAssignmentNameForKeyVaultSecretsUser": "[guid(parameters('principalId'), variables('roleDefinitionIdForKeyVaultSecretsUser'), resourceGroup().id)]" + "principalId": { + "type": "String", + "metadata": { + "description": "Specifies the principal ID assigned to the role. You can find it by logging into 'https://shell.azure.com/bash' and run 'az ad signed-in-user show --query id -o tsv'" + } }, - "functions": [], - "resources": [ - { - "type": "Microsoft.KeyVault/vaults", - "apiVersion": "2021-10-01", - "name": "[variables('keyVaultName')]", - "location": "[variables('location')]", - "properties": { - "tenantId": "[variables('tenantId')]", - "sku": { - "name": "standard", - "family": "A" - }, - "accessPolicies": [], - "enableSoftDelete": true, - "enableRbacAuthorization": true - }, - "resources": [ - { - "type": "Microsoft.KeyVault/vaults/secrets", - "apiVersion": "2021-10-01", - "name": "[concat(variables('keyVaultName'), '/FEATHR-PREFIX')]", - "location": "[resourceGroup().location]", - "dependsOn": [ - "[variables('keyVault')]" - ], - "properties": { - "value": "[parameters('resourcePrefix')]" - } - } - ] + "allowAllConnections": { + "defaultValue": "true", + "allowedValues": ["true", "false"], + "type": "String", + "metadata": { + "description": "Specifies whether to allow client IPs to connect to Synapse" + } + }, + "provisionPurview": { + "defaultValue": "true", + "allowedValues": ["true", "false"], + "type": "String", + "metadata": { + "description": "Whether or not put purview in the provision script" + } + }, + "provisionEventHub": { + "defaultValue": "true", + "allowedValues": ["true", "false"], + "type": "String", + "metadata": { + "description": "Whether or not to deploy eventhub provision script" + } + } + }, + "variables": { + "location": "[resourceGroup().location]", + "tenantId": "[subscription().tenantId]", + "redisCacheName": "[concat(parameters('resourcePrefix'),'redis' )]", + "keyVaultName": "[concat(parameters('resourcePrefix'),'kv')]", + "eventhubNameSpaceName": "[concat(parameters('resourcePrefix'),'ehns')]", + "eventhubName": "[concat(parameters('resourcePrefix'),'eh')]", + "eventhubSku": "Standard", + "eventhubSkuCapacity": 1, + "keyVault": "[resourceId('Microsoft.KeyVault/vaults', variables('keyVaultName'))]", + "redisCache": "[resourceId('Microsoft.Cache/redis', variables('redisCacheName'))]", + "eventhubNameSpace": "[resourceId('Microsoft.EventHub/namespaces/', variables('eventhubNameSpaceName'))]", + "sparkPoolName": "spark31", + "workspaceName": "[toLower(concat(parameters('resourcePrefix'),'syws'))]", + "dlsName": "[toLower(concat(parameters('resourcePrefix'), 'dls'))]", + "dlsFsName": "[toLower(concat(parameters('resourcePrefix'),'fs'))]", + "dlsAccount": "[resourceId('Microsoft.Storage/storageAccounts', variables('dlsName'))]", + "purviewName": "[concat(parameters('resourcePrefix'),'purview' )]", + "roleDefinitionIdForBlobContributor": "ba92f5b4-2d11-453d-a403-e96b0029c9fe", + "roleDefinitionIdForKeyVaultSecretsUser": "4633458b-17de-408a-b874-0445c86b69e6", + "roleAssignmentNameForBlobContributor": "[guid(parameters('principalId'), variables('roleDefinitionIdForBlobContributor'), resourceGroup().id)]", + "roleAssignmentNameForKeyVaultSecretsUser": "[guid(parameters('principalId'), variables('roleDefinitionIdForKeyVaultSecretsUser'), resourceGroup().id)]" + }, + "functions": [], + "resources": [ + { + "type": "Microsoft.KeyVault/vaults", + "apiVersion": "2021-10-01", + "name": "[variables('keyVaultName')]", + "location": "[variables('location')]", + "properties": { + "tenantId": "[variables('tenantId')]", + "sku": { + "name": "standard", + "family": "A" }, + "accessPolicies": [], + "enableSoftDelete": true, + "enableRbacAuthorization": true + }, + "resources": [ { - "type": "Microsoft.Cache/redis", - "apiVersion": "2021-06-01", - "name": "[variables('redisCacheName')]", - "location": "[resourceGroup().location]", - "tags": { - "displayName": "Feathr Online Store" - }, - "properties": { - "redisVersion": "6", - "sku": { - "name": "Basic", - "family": "C", - "capacity": 2 - } - }, - "resources": [ - { - "type": "Microsoft.KeyVault/vaults/secrets", - "apiVersion": "2021-10-01", - "name": "[concat(variables('keyVaultName'), '/FEATHR-ONLINE-STORE-CONN')]", - "location": "[resourceGroup().location]", - "dependsOn": [ - "[variables('keyVault')]", - "[variables('redisCache')]" - ], - "properties": { - "value": "[concat(variables('redisCacheName'),'.redis.cache.windows.net:6380,password=', listKeys(concat('Microsoft.Cache/redis/', variables('redisCacheName')), '2021-06-01').primaryKey, ',ssl=True')]" - } - }, - { - "type": "Microsoft.KeyVault/vaults/secrets", - "apiVersion": "2021-10-01", - "name": "[concat(variables('keyVaultName'), '/REDIS-PASSWORD')]", - "location": "[resourceGroup().location]", - "dependsOn": [ - "[variables('keyVault')]", - "[variables('redisCache')]" - ], - "properties": { - "value": "[listKeys(concat('Microsoft.Cache/redis/', variables('redisCacheName')), '2021-06-01').primaryKey]" - } - } - ] - }, + "type": "Microsoft.KeyVault/vaults/secrets", + "apiVersion": "2021-10-01", + "name": "[concat(variables('keyVaultName'), '/FEATHR-PREFIX')]", + "location": "[resourceGroup().location]", + "dependsOn": ["[variables('keyVault')]"], + "properties": { + "value": "[parameters('resourcePrefix')]" + } + } + ] + }, + { + "type": "Microsoft.Cache/redis", + "apiVersion": "2021-06-01", + "name": "[variables('redisCacheName')]", + "location": "[resourceGroup().location]", + "tags": { + "displayName": "Feathr Online Store" + }, + "properties": { + "redisVersion": "6", + "sku": { + "name": "Basic", + "family": "C", + "capacity": 2 + } + }, + "resources": [ { - "type": "Microsoft.Storage/storageAccounts", - "apiVersion": "2021-08-01", - "name": "[variables('dlsName')]", - "location": "[variables('location')]", - "sku": { - "name": "Standard_LRS", - "tier": "Standard" - }, - "kind": "StorageV2", - "properties": { - "accessTier": "Hot", - "supportsHttpsTrafficOnly": true, - "isHnsEnabled": true - }, - "resources": [ - { - "type": "blobServices/containers", - "apiVersion": "2021-08-01", - "name": "[concat('default/', variables('dlsFsName'))]", - "dependsOn": [ - "[variables('dlsName')]" - ], - "properties": { - "publicAccess": "None" - } - } - ] + "type": "Microsoft.KeyVault/vaults/secrets", + "apiVersion": "2021-10-01", + "name": "[concat(variables('keyVaultName'), '/FEATHR-ONLINE-STORE-CONN')]", + "location": "[resourceGroup().location]", + "dependsOn": ["[variables('keyVault')]", "[variables('redisCache')]"], + "properties": { + "value": "[concat(variables('redisCacheName'),'.redis.cache.windows.net:6380,password=', listKeys(concat('Microsoft.Cache/redis/', variables('redisCacheName')), '2021-06-01').primaryKey, ',ssl=True')]" + } }, { - "condition": "[equals(parameters('provisionPurview'),'true')]", - "type": "Microsoft.Purview/accounts", - "apiVersion": "2021-07-01", - "name": "[variables('purviewName')]", - "location": "[variables('location')]", - "sku": { - "name": "Standard", - "capacity": 1 - }, - "identity": { - "type": "SystemAssigned" - }, - "properties": { - "cloudConnectors": {}, - "publicNetworkAccess": "Enabled" - } - }, + "type": "Microsoft.KeyVault/vaults/secrets", + "apiVersion": "2021-10-01", + "name": "[concat(variables('keyVaultName'), '/REDIS-PASSWORD')]", + "location": "[resourceGroup().location]", + "dependsOn": ["[variables('keyVault')]", "[variables('redisCache')]"], + "properties": { + "value": "[listKeys(concat('Microsoft.Cache/redis/', variables('redisCacheName')), '2021-06-01').primaryKey]" + } + } + ] + }, + { + "type": "Microsoft.Storage/storageAccounts", + "apiVersion": "2021-08-01", + "name": "[variables('dlsName')]", + "location": "[variables('location')]", + "sku": { + "name": "Standard_LRS", + "tier": "Standard" + }, + "kind": "StorageV2", + "properties": { + "accessTier": "Hot", + "supportsHttpsTrafficOnly": true, + "isHnsEnabled": true + }, + "resources": [ { - "condition": "[equals(parameters('provisionEventHub'),'true')]", - "type": "Microsoft.EventHub/namespaces", - "apiVersion": "2021-11-01", - "name": "[variables('eventhubNameSpaceName')]", - "location": "[variables('location')]", - "sku": { - "name": "[variables('eventhubSku')]", - "tier": "[variables('eventhubSku')]", - "capacity": "[variables('eventhubSkuCapacity')]" - }, - "properties": {}, - "resources": [ - { - "condition": "[equals(parameters('provisionEventHub'),'true')]", - "type": "eventhubs", - "apiVersion": "2021-11-01", - "name": "[variables('eventhubName')]", - "dependsOn": [ - "[variables('eventhubNameSpace')]" - ], - "properties": {} - }, - { - "condition": "[equals(parameters('provisionEventHub'),'true')]", - "type": "Microsoft.KeyVault/vaults/secrets", - "apiVersion": "2021-10-01", - "name": "[concat(variables('keyVaultName'), '/EVENTHUB-POLICY-KEY')]", - "condition": "[equals(parameters('provisionEventHub'),'true')]", - "location": "[resourceGroup().location]", - "dependsOn": [ - "[variables('keyVault')]", - "[variables('eventhubNameSpace')]" - ], - "properties": { - "value": "[if(equals(parameters('provisionEventHub'),'true'),listKeys(resourceId('Microsoft.EventHub/namespaces/AuthorizationRules',variables('eventhubNameSpaceName'),'RootManageSharedAccessKey'),'2021-11-01').primaryConnectionString,'null' )]" - } - } - ] + "type": "blobServices/containers", + "apiVersion": "2021-08-01", + "name": "[concat('default/', variables('dlsFsName'))]", + "dependsOn": ["[variables('dlsName')]"], + "properties": { + "publicAccess": "None" + } + } + ] + }, + { + "condition": "[equals(parameters('provisionPurview'),'true')]", + "type": "Microsoft.Purview/accounts", + "apiVersion": "2021-07-01", + "name": "[variables('purviewName')]", + "location": "[variables('location')]", + "sku": { + "name": "Standard", + "capacity": 1 + }, + "identity": { + "type": "SystemAssigned" + }, + "properties": { + "cloudConnectors": {}, + "publicNetworkAccess": "Enabled" + } + }, + { + "condition": "[equals(parameters('provisionEventHub'),'true')]", + "type": "Microsoft.EventHub/namespaces", + "apiVersion": "2021-11-01", + "name": "[variables('eventhubNameSpaceName')]", + "location": "[variables('location')]", + "sku": { + "name": "[variables('eventhubSku')]", + "tier": "[variables('eventhubSku')]", + "capacity": "[variables('eventhubSkuCapacity')]" + }, + "properties": {}, + "resources": [ + { + "condition": "[equals(parameters('provisionEventHub'),'true')]", + "type": "eventhubs", + "apiVersion": "2021-11-01", + "name": "[variables('eventhubName')]", + "dependsOn": ["[variables('eventhubNameSpace')]"], + "properties": {} }, { - "type": "Microsoft.Synapse/workspaces", - "apiVersion": "2021-06-01", - "name": "[variables('workspaceName')]", - "location": "[variables('location')]", - "dependsOn": [ - "[variables('dlsName')]", - "[variables('dlsFsName')]" - ], - "identity": { - "type": "SystemAssigned" - }, - "properties": { - "defaultDataLakeStorage": { - "accountUrl": "[reference(variables('dlsName')).primaryEndpoints.dfs]", - "filesystem": "[variables('dlsFsName')]" - }, - "managedVirtualNetwork": "default" - }, - "resources": [ - { - "type": "firewallrules", - "apiVersion": "2021-06-01", - "name": "allowAll", - "location": "[variables('location')]", - "dependsOn": [ - "[variables('workspaceName')]" - ], - "properties": { - "startIpAddress": "0.0.0.0", - "endIpAddress": "255.255.255.255" - }, - "condition": "[equals(parameters('allowAllConnections'),'true')]" - }, - { - "type": "firewallrules", - "apiVersion": "2021-06-01", - "name": "AllowAllWindowsAzureIps", - "location": "[variables('location')]", - "dependsOn": [ - "[variables('workspaceName')]" - ], - "properties": { - "startIpAddress": "0.0.0.0", - "endIpAddress": "0.0.0.0" - } - }, - { - "type": "managedIdentitySqlControlSettings", - "apiVersion": "2021-06-01", - "name": "default", - "location": "[variables('location')]", - "dependsOn": [ - "[variables('workspaceName')]" - ], - "properties": { - "grantSqlControlToManagedIdentity": { - "desiredState": "Enabled" - } - } - } - ] + "condition": "[equals(parameters('provisionEventHub'),'true')]", + "type": "Microsoft.KeyVault/vaults/secrets", + "apiVersion": "2021-10-01", + "name": "[concat(variables('keyVaultName'), '/EVENTHUB-POLICY-KEY')]", + "condition": "[equals(parameters('provisionEventHub'),'true')]", + "location": "[resourceGroup().location]", + "dependsOn": [ + "[variables('keyVault')]", + "[variables('eventhubNameSpace')]" + ], + "properties": { + "value": "[if(equals(parameters('provisionEventHub'),'true'),listKeys(resourceId('Microsoft.EventHub/namespaces/AuthorizationRules',variables('eventhubNameSpaceName'),'RootManageSharedAccessKey'),'2021-11-01').primaryConnectionString,'null' )]" + } + } + ] + }, + { + "type": "Microsoft.Synapse/workspaces", + "apiVersion": "2021-06-01", + "name": "[variables('workspaceName')]", + "location": "[variables('location')]", + "dependsOn": ["[variables('dlsName')]", "[variables('dlsFsName')]"], + "identity": { + "type": "SystemAssigned" + }, + "properties": { + "defaultDataLakeStorage": { + "accountUrl": "[reference(variables('dlsName')).primaryEndpoints.dfs]", + "filesystem": "[variables('dlsFsName')]" }, + "managedVirtualNetwork": "default" + }, + "resources": [ { - "type": "Microsoft.Synapse/workspaces/bigDataPools", - "apiVersion": "2021-06-01", - "name": "[concat(variables('workspaceName'), '/', variables('sparkPoolName'))]", - "location": "[variables('location')]", - "dependsOn": [ - "[variables('workspaceName')]" - ], - "properties": { - "autoPause": { - "delayInMinutes": 30, - "enabled": true - }, - "autoScale": { - "enabled": true, - "minNodeCount": 1, - "maxNodeCount": 3 - }, - "sparkVersion": "3.1", - "nodeCount": 3, - "nodeSizeFamily": "MemoryOptimized", - "nodeSize": "Medium" - } + "type": "firewallrules", + "apiVersion": "2021-06-01", + "name": "allowAll", + "location": "[variables('location')]", + "dependsOn": ["[variables('workspaceName')]"], + "properties": { + "startIpAddress": "0.0.0.0", + "endIpAddress": "255.255.255.255" + }, + "condition": "[equals(parameters('allowAllConnections'),'true')]" }, { - "type": "Microsoft.Authorization/roleAssignments", - "apiVersion": "2020-10-01-preview", - "name": "[variables('roleAssignmentNameForBlobContributor')]", - "dependsOn": [ - "[variables('dlsAccount')]" - ], - "properties": { - "roleDefinitionId": "[resourceId('Microsoft.Authorization/roleDefinitions', variables('roleDefinitionIdForBlobContributor'))]", - "principalId": "[parameters('principalId')]", - "scope": "[resourceGroup().id]" - } + "type": "firewallrules", + "apiVersion": "2021-06-01", + "name": "AllowAllWindowsAzureIps", + "location": "[variables('location')]", + "dependsOn": ["[variables('workspaceName')]"], + "properties": { + "startIpAddress": "0.0.0.0", + "endIpAddress": "0.0.0.0" + } }, { - "type": "Microsoft.Authorization/roleAssignments", - "apiVersion": "2020-10-01-preview", - "name": "[variables('roleAssignmentNameForKeyVaultSecretsUser')]", - "dependsOn": [ - "[variables('keyVault')]" - ], - "properties": { - "roleDefinitionId": "[resourceId('Microsoft.Authorization/roleDefinitions', variables('roleDefinitionIdForKeyVaultSecretsUser'))]", - "principalId": "[parameters('principalId')]", - "scope": "[resourceGroup().id]" + "type": "managedIdentitySqlControlSettings", + "apiVersion": "2021-06-01", + "name": "default", + "location": "[variables('location')]", + "dependsOn": ["[variables('workspaceName')]"], + "properties": { + "grantSqlControlToManagedIdentity": { + "desiredState": "Enabled" } + } } - ], - "outputs": {} -} \ No newline at end of file + ] + }, + { + "type": "Microsoft.Synapse/workspaces/bigDataPools", + "apiVersion": "2021-06-01", + "name": "[concat(variables('workspaceName'), '/', variables('sparkPoolName'))]", + "location": "[variables('location')]", + "dependsOn": ["[variables('workspaceName')]"], + "properties": { + "autoPause": { + "delayInMinutes": 30, + "enabled": true + }, + "autoScale": { + "enabled": true, + "minNodeCount": 1, + "maxNodeCount": 3 + }, + "sparkVersion": "3.1", + "nodeCount": 3, + "nodeSizeFamily": "MemoryOptimized", + "nodeSize": "Medium" + } + }, + { + "type": "Microsoft.Authorization/roleAssignments", + "apiVersion": "2020-10-01-preview", + "name": "[variables('roleAssignmentNameForBlobContributor')]", + "dependsOn": ["[variables('dlsAccount')]"], + "properties": { + "roleDefinitionId": "[resourceId('Microsoft.Authorization/roleDefinitions', variables('roleDefinitionIdForBlobContributor'))]", + "principalId": "[parameters('principalId')]", + "scope": "[resourceGroup().id]" + } + }, + { + "type": "Microsoft.Authorization/roleAssignments", + "apiVersion": "2020-10-01-preview", + "name": "[variables('roleAssignmentNameForKeyVaultSecretsUser')]", + "dependsOn": ["[variables('keyVault')]"], + "properties": { + "roleDefinitionId": "[resourceId('Microsoft.Authorization/roleDefinitions', variables('roleDefinitionIdForKeyVaultSecretsUser'))]", + "principalId": "[parameters('principalId')]", + "scope": "[resourceGroup().id]" + } + } + ], + "outputs": {} +} diff --git a/docs/how-to-guides/deployment/deploy.json b/docs/how-to-guides/deployment/deploy.json index 0c8241eef..3e3dd40f5 100644 --- a/docs/how-to-guides/deployment/deploy.json +++ b/docs/how-to-guides/deployment/deploy.json @@ -21,16 +21,13 @@ "principalId": { "type": "string", "metadata": { - "description": "Specifies the principal ID assigned to the role. You can find it by logging into 'https://shell.azure.com/bash' and run 'az ad signed-in-user show --query objectId -o tsv' " + "description": "Specifies the principal ID assigned to the role. You can find it by logging into 'https://shell.azure.com/bash' and run 'az ad signed-in-user show --query id -o tsv' " } }, "allowAllConnections": { "type": "string", "defaultValue": "true", - "allowedValues": [ - "true", - "false" - ], + "allowedValues": ["true", "false"], "metadata": { "description": "Specifies whether to allow client IPs to connect to Synapse" } @@ -635,4 +632,4 @@ "value": "[parameters('resourcePrefix')]" } } -} \ No newline at end of file +} diff --git a/docs/how-to-guides/deployment/main.bicep b/docs/how-to-guides/deployment/main.bicep index aef63e2b4..566ff6668 100644 --- a/docs/how-to-guides/deployment/main.bicep +++ b/docs/how-to-guides/deployment/main.bicep @@ -8,7 +8,7 @@ targetScope = 'subscription' @minLength(3) param resourcePrefix string = 'feathr${take(newGuid(),5)}' -@description('Specifies the principal ID assigned to the role. You can find it by logging into \'https://shell.azure.com/bash\' and run \'az ad signed-in-user show --query objectId -o tsv\' ') +@description('Specifies the principal ID assigned to the role. You can find it by logging into \'https://shell.azure.com/bash\' and run \'az ad signed-in-user show --query id -o tsv\' ') param principalId string @description('Specifies whether to allow client IPs to connect to Synapse') diff --git a/feathr_project/feathrcli/data/feathr_user_workspace/nyc_driver_demo.ipynb b/feathr_project/feathrcli/data/feathr_user_workspace/nyc_driver_demo.ipynb index 3ccd51814..0ed133169 100644 --- a/feathr_project/feathrcli/data/feathr_user_workspace/nyc_driver_demo.ipynb +++ b/feathr_project/feathrcli/data/feathr_user_workspace/nyc_driver_demo.ipynb @@ -1,736 +1,736 @@ { - "cells": [ - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "# Feathr Feature Store on Azure Demo Notebook\n", - "\n", - "This notebook illustrates the use of Feature Store to create a model that predicts NYC Taxi fares. It includes these steps:\n", - "\n", - "\n", - "This tutorial demonstrates the key capabilities of Feathr, including:\n", - "\n", - "1. Install and set up Feathr with Azure\n", - "2. Create shareable features with Feathr feature definition configs.\n", - "3. Create a training dataset via point-in-time feature join.\n", - "4. Compute and write features.\n", - "5. Train a model using these features to predict fares.\n", - "6. Materialize feature value to online store.\n", - "7. Fetch feature value in real-time from online store for online scoring.\n", - "\n", - "In this tutorial, we use Feathr Feature Store to create a model that predicts NYC Taxi fares. The dataset comes from [here](https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page). The feature flow is as below:\n", - "\n", - "![Feature Flow](https://github.com/linkedin/feathr/blob/main/docs/images/feature_flow.png?raw=true)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Use Quick Start Template to Provision Azure Resources\n", - "\n", - "Feathr has native cloud integration. To use Feathr on Azure, you only need three steps:\n", - "\n", - "- Get the `Principal ID` of your account by running `az ad signed-in-user show --query objectId -o tsv` in the link below (Select \"Bash\" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI.\n", - "\n", - "[Launch Cloud Shell](https://shell.azure.com/bash)\n", - "\n", - "- Click the button below to deploy a minimal set of Feathr resources for demo purpose. You will need to fill in the `Principal ID` and `Resource Prefix`. You will need \"Owner\" permission of the selected subscription.\n", - "\n", - "[![Deploy to Azure](https://aka.ms/deploytoazurebutton)](https://portal.azure.com/#create/Microsoft.Template/uri/https%3A%2F%2Fraw.githubusercontent.com%2Flinkedin%2Ffeathr%2Fmain%2Fdocs%2Fhow-to-guides%2Fazure_resource_provision.json)\n", - "\n", - "- Run the cells below.\n", - "\n", - "And the architecture is as below. In the above template, we are using Synapse as Spark provider, use Azure Data Lake Gen2 as offline store, and use Redis as online store, Azure Purview (Apache Atlas compatible) as feature reigstry. \n", - "\n", - "\n", - "![Architecture](https://github.com/linkedin/feathr/blob/main/docs/images/architecture.png?raw=true)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Install Feathr \n", - "\n", - "Install Feathr using pip:\n", - "\n", - "`pip install -U feathr pandavro scikit-learn`" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Configure the required environment with Feathr Quick Start Template\n", - "\n", - "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. Run the code below to install Feathr, login to Azure to get the required credentials to access more cloud resources." - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "**REQUIRED STEP: Fill in the resource prefix when provisioning the resources**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "resource_prefix = \"feathr_resource_prefix\"" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "! pip install feathr azure-cli pandavro scikit-learn" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Login to Azure with a device code (You will see instructions in the output):" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "! az login --use-device-code" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "import glob\n", - "import os\n", - "import tempfile\n", - "from datetime import datetime, timedelta\n", - "from math import sqrt\n", - "\n", - "import pandas as pd\n", - "import pandavro as pdx\n", - "from feathr import FeathrClient\n", - "from feathr import BOOLEAN, FLOAT, INT32, ValueType\n", - "from feathr import Feature, DerivedFeature, FeatureAnchor\n", - "from feathr import BackfillTime, MaterializationSettings\n", - "from feathr import FeatureQuery, ObservationSettings\n", - "from feathr import RedisSink\n", - "from feathr import INPUT_CONTEXT, HdfsSource\n", - "from feathr import WindowAggTransformation\n", - "from feathr import TypedKey\n", - "from sklearn.metrics import mean_squared_error\n", - "from sklearn.model_selection import train_test_split\n", - "from azure.identity import DefaultAzureCredential\n", - "from azure.keyvault.secrets import SecretClient\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Get all the required credentials from Azure KeyVault" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Get all the required credentials from Azure Key Vault\n", - "key_vault_name=resource_prefix+\"kv\"\n", - "synapse_workspace_url=resource_prefix+\"syws\"\n", - "adls_account=resource_prefix+\"dls\"\n", - "adls_fs_name=resource_prefix+\"fs\"\n", - "purview_name=resource_prefix+\"purview\"\n", - "key_vault_uri = f\"https://{key_vault_name}.vault.azure.net\"\n", - "credential = DefaultAzureCredential(exclude_interactive_browser_credential=False)\n", - "client = SecretClient(vault_url=key_vault_uri, credential=credential)\n", - "secretName = \"FEATHR-ONLINE-STORE-CONN\"\n", - "retrieved_secret = client.get_secret(secretName).value\n", - "\n", - "# Get redis credentials; This is to parse Redis connection string.\n", - "redis_port=retrieved_secret.split(',')[0].split(\":\")[1]\n", - "redis_host=retrieved_secret.split(',')[0].split(\":\")[0]\n", - "redis_password=retrieved_secret.split(',')[1].split(\"password=\",1)[1]\n", - "redis_ssl=retrieved_secret.split(',')[2].split(\"ssl=\",1)[1]\n", - "\n", - "# Set the resource link\n", - "os.environ['spark_config__azure_synapse__dev_url'] = f'https://{synapse_workspace_url}.dev.azuresynapse.net'\n", - "os.environ['spark_config__azure_synapse__pool_name'] = 'spark31'\n", - "os.environ['spark_config__azure_synapse__workspace_dir'] = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_project'\n", - "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", - "os.environ['online_store__redis__host'] = redis_host\n", - "os.environ['online_store__redis__port'] = redis_port\n", - "os.environ['online_store__redis__ssl_enabled'] = redis_ssl\n", - "os.environ['REDIS_PASSWORD']=redis_password\n", - "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", - "feathr_output_path = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_output'" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Configure the required environment (Don't need to update if using the above Quick Start Template)\n", - "\n", - "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. If you use Feathr CLI to create a workspace, you should have a folder with a file called `feathr_config.yaml` in it with all the required configurations. Otherwise, update the configuration below.\n", - "\n", - "The code below will write this configuration string to a temporary location and load it to Feathr. Please still refer to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It should also have more explanations on the meaning of each variable." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "import tempfile\n", - "yaml_config = \"\"\"\n", - "# Please refer to https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml for explanations on the meaning of each field.\n", - "api_version: 1\n", - "project_config:\n", - " project_name: 'feathr_getting_started'\n", - " required_environment_variables:\n", - " - 'REDIS_PASSWORD'\n", - " - 'AZURE_CLIENT_ID'\n", - " - 'AZURE_TENANT_ID'\n", - " - 'AZURE_CLIENT_SECRET'\n", - "offline_store:\n", - " adls:\n", - " adls_enabled: true\n", - " wasb:\n", - " wasb_enabled: true\n", - " s3:\n", - " s3_enabled: false\n", - " s3_endpoint: 's3.amazonaws.com'\n", - " jdbc:\n", - " jdbc_enabled: false\n", - " jdbc_database: 'feathrtestdb'\n", - " jdbc_table: 'feathrtesttable'\n", - " snowflake:\n", - " url: \"dqllago-ol19457.snowflakecomputing.com\"\n", - " user: \"feathrintegration\"\n", - " role: \"ACCOUNTADMIN\"\n", - "spark_config:\n", - " spark_cluster: 'azure_synapse'\n", - " spark_result_output_parts: '1'\n", - " azure_synapse:\n", - " dev_url: 'https://feathrazuretest3synapse.dev.azuresynapse.net'\n", - " pool_name: 'spark3'\n", - " workspace_dir: 'abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/feathr_getting_started'\n", - " executor_size: 'Small'\n", - " executor_num: 4\n", - " feathr_runtime_location: wasbs://public@azurefeathrstorage.blob.core.windows.net/feathr-assembly-LATEST.jar\n", - " databricks:\n", - " workspace_instance_url: 'https://adb-2474129336842816.16.azuredatabricks.net'\n", - " config_template: {'run_name':'','new_cluster':{'spark_version':'9.1.x-scala2.12','node_type_id':'Standard_D3_v2','num_workers':2,'spark_conf':{}},'libraries':[{'jar':''}],'spark_jar_task':{'main_class_name':'','parameters':['']}}\n", - " work_dir: 'dbfs:/feathr_getting_started'\n", - " feathr_runtime_location: https://azurefeathrstorage.blob.core.windows.net/public/feathr-assembly-LATEST.jar\n", - "online_store:\n", - " redis:\n", - " host: 'feathrazuretest3redis.redis.cache.windows.net'\n", - " port: 6380\n", - " ssl_enabled: True\n", - "feature_registry:\n", - " purview:\n", - " type_system_initialization: true\n", - " purview_name: 'feathrazuretest3-purview1'\n", - " delimiter: '__'\n", - "\"\"\"\n", - "tmp = tempfile.NamedTemporaryFile(mode='w', delete=False)\n", - "with open(tmp.name, \"w\") as text_file:\n", - " text_file.write(yaml_config)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Setup necessary environment variables (Skip if using the above Quick Start Template)\n", - "\n", - "You should setup the environment variables in order to run this sample. More environment variables can be set by referring to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It also has more explanations on the meaning of each variable.\n", - "\n", - "To run this notebook, for Azure users, you need AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET and REDIS_PASSWORD.\n", - "To run this notebook, for Databricks useres, you need DATABRICKS_WORKSPACE_TOKEN_VALUE and REDIS_PASSWORD." - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "# Initialize Feathr Client" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client = FeathrClient(config_path=tmp.name)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## View the data\n", - "\n", - "In this tutorial, we use Feathr Feature Store to create a model that predicts NYC Taxi fares. The dataset comes from [here](https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page). The data is as below" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "import pandas as pd\n", - "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/green_tripdata_2020-04_with_index.csv\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Defining Features with Feathr\n", - "\n", - "In Feathr, a feature is viewed as a function, mapping from entity id or key, and timestamp to a feature value. For more details on feature definition, please refer to the [Feathr Feature Definition Guide](https://github.com/linkedin/feathr/blob/main/docs/concepts/feature-definition.md)\n", - "\n", - "\n", - "1. The typed key (a.k.a. entity id) identifies the subject of feature, e.g. a user id, 123.\n", - "2. The feature name is the aspect of the entity that the feature is indicating, e.g. the age of the user.\n", - "3. The feature value is the actual value of that aspect at a particular time, e.g. the value is 30 at year 2022.\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Note that, in some cases, such as features defined on top of request data, may have no entity key or timestamp.\n", - "It is merely a function/transformation executing against request data at runtime.\n", - "For example, the day of week of the request, which is calculated by converting the request UNIX timestamp.\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Define Sources Section with UDFs\n", - "A feature source is needed for anchored features that describes the raw data in which the feature values are computed from. See the python documentation to get the details on each input column.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from pyspark.sql import SparkSession, DataFrame\n", - "def feathr_udf_day_calc(df: DataFrame) -> DataFrame:\n", - " from pyspark.sql.functions import dayofweek, dayofyear, col\n", - " df = df.withColumn(\"fare_amount_cents\", col(\"fare_amount\")*100)\n", - " return df\n", - "\n", - "batch_source = HdfsSource(name=\"nycTaxiBatchSource\",\n", - " path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/green_tripdata_2020-04_with_index.csv\",\n", - " event_timestamp_column=\"lpep_dropoff_datetime\",\n", - " preprocessing=feathr_udf_day_calc,\n", - " timestamp_format=\"yyyy-MM-dd HH:mm:ss\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Define Anchors and Features\n", - "A feature is called an anchored feature when the feature is directly extracted from the source data, rather than computed on top of other features. The latter case is called derived feature." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "f_trip_distance = Feature(name=\"f_trip_distance\",\n", - " feature_type=FLOAT, transform=\"trip_distance\")\n", - "f_trip_time_duration = Feature(name=\"f_trip_time_duration\",\n", - " feature_type=INT32,\n", - " transform=\"(to_unix_timestamp(lpep_dropoff_datetime) - to_unix_timestamp(lpep_pickup_datetime))/60\")\n", - "\n", - "features = [\n", - " f_trip_distance,\n", - " f_trip_time_duration,\n", - " Feature(name=\"f_is_long_trip_distance\",\n", - " feature_type=BOOLEAN,\n", - " transform=\"cast_float(trip_distance)>30\"),\n", - " Feature(name=\"f_day_of_week\",\n", - " feature_type=INT32,\n", - " transform=\"dayofweek(lpep_dropoff_datetime)\"),\n", - "]\n", - "\n", - "request_anchor = FeatureAnchor(name=\"request_features\",\n", - " source=INPUT_CONTEXT,\n", - " features=features)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Window aggregation features\n", - "\n", - "For window aggregation features, see the supported fields below:\n", - "\n", - "Note that the `agg_func` should be any of these:\n", - "\n", - "| Aggregation Type | Input Type | Description |\n", - "| --- | --- | --- |\n", - "|SUM, COUNT, MAX, MIN, AVG\t|Numeric|Applies the the numerical operation on the numeric inputs. |\n", - "|MAX_POOLING, MIN_POOLING, AVG_POOLING\t| Numeric Vector | Applies the max/min/avg operation on a per entry bassis for a given a collection of numbers.|\n", - "|LATEST| Any |Returns the latest not-null values from within the defined time window |\n", - "\n", - "\n", - "After you have defined features and sources, bring them together to build an anchor:\n", - "\n", - "\n", - "Note that if the data source is from the observation data, the `source` section should be `INPUT_CONTEXT` to indicate the source of those defined anchors." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "location_id = TypedKey(key_column=\"DOLocationID\",\n", - " key_column_type=ValueType.INT32,\n", - " description=\"location id in NYC\",\n", - " full_name=\"nyc_taxi.location_id\")\n", - "agg_features = [Feature(name=\"f_location_avg_fare\",\n", - " key=location_id,\n", - " feature_type=FLOAT,\n", - " transform=WindowAggTransformation(agg_expr=\"cast_float(fare_amount)\",\n", - " agg_func=\"AVG\",\n", - " window=\"90d\")),\n", - " Feature(name=\"f_location_max_fare\",\n", - " key=location_id,\n", - " feature_type=FLOAT,\n", - " transform=WindowAggTransformation(agg_expr=\"cast_float(fare_amount)\",\n", - " agg_func=\"MAX\",\n", - " window=\"90d\")),\n", - " Feature(name=\"f_location_total_fare_cents\",\n", - " key=location_id,\n", - " feature_type=FLOAT,\n", - " transform=WindowAggTransformation(agg_expr=\"fare_amount_cents\",\n", - " agg_func=\"SUM\",\n", - " window=\"90d\")),\n", - " ]\n", - "\n", - "agg_anchor = FeatureAnchor(name=\"aggregationFeatures\",\n", - " source=batch_source,\n", - " features=agg_features)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Derived Features Section\n", - "Derived features are the features that are computed from other features. They could be computed from anchored features, or other derived features." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "f_trip_time_distance = DerivedFeature(name=\"f_trip_time_distance\",\n", - " feature_type=FLOAT,\n", - " input_features=[\n", - " f_trip_distance, f_trip_time_duration],\n", - " transform=\"f_trip_distance * f_trip_time_duration\")\n", - "\n", - "f_trip_time_rounded = DerivedFeature(name=\"f_trip_time_rounded\",\n", - " feature_type=INT32,\n", - " input_features=[f_trip_time_duration],\n", - " transform=\"f_trip_time_duration % 10\")\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "And then we need to build those features so that it can be consumed later. Note that we have to build both the \"anchor\" and the \"derived\" features (which is not anchored to a source)." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.build_features(anchor_list=[agg_anchor, request_anchor], derived_feature_list=[\n", - " f_trip_time_distance, f_trip_time_rounded])" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Create training data using point-in-time correct feature join\n", - "\n", - "A training dataset usually contains entity id columns, multiple feature columns, event timestamp column and label/target column. \n", - "\n", - "To create a training dataset using Feathr, one needs to provide a feature join configuration file to specify\n", - "what features and how these features should be joined to the observation data. \n", - "\n", - "To learn more on this topic, please refer to [Point-in-time Correctness](https://github.com/linkedin/feathr/blob/main/docs/concepts/point-in-time-join.md)\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "if client.spark_runtime == 'databricks':\n", - " output_path = 'dbfs:/feathrazure_test.avro'\n", - "else:\n", - " output_path = feathr_output_path\n", - "\n", - "\n", - "feature_query = FeatureQuery(\n", - " feature_list=[\"f_location_avg_fare\", \"f_trip_time_rounded\", \"f_is_long_trip_distance\", \"f_location_total_fare_cents\"], key=location_id)\n", - "settings = ObservationSettings(\n", - " observation_path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/green_tripdata_2020-04_with_index.csv\",\n", - " event_timestamp_column=\"lpep_dropoff_datetime\",\n", - " timestamp_format=\"yyyy-MM-dd HH:mm:ss\")\n", - "client.get_offline_features(observation_settings=settings,\n", - " feature_query=feature_query,\n", - " output_path=output_path)\n", - "client.wait_job_to_finish(timeout_sec=500)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Download the result and show the result\n", - "\n", - "Let's use the helper function `get_result_df` to download the result and view it:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "def get_result_df(client: FeathrClient) -> pd.DataFrame:\n", - " \"\"\"Download the job result dataset from cloud as a Pandas dataframe.\"\"\"\n", - " res_url = client.get_job_result_uri(block=True, timeout_sec=600)\n", - " tmp_dir = tempfile.TemporaryDirectory()\n", - " client.feathr_spark_laucher.download_result(result_path=res_url, local_folder=tmp_dir.name)\n", - " dataframe_list = []\n", - " # assuming the result are in avro format\n", - " for file in glob.glob(os.path.join(tmp_dir.name, '*.avro')):\n", - " dataframe_list.append(pdx.read_avro(file))\n", - " vertical_concat_df = pd.concat(dataframe_list, axis=0)\n", - " tmp_dir.cleanup()\n", - " return vertical_concat_df\n", - "\n", - "df_res = get_result_df(client)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "df_res" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Train a machine learning model\n", - "After getting all the features, let's train a machine learning model with the converted feature by Feathr:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# remove columns\n", - "from sklearn.ensemble import GradientBoostingRegressor\n", - "final_df = df_res\n", - "final_df.drop([\"lpep_pickup_datetime\", \"lpep_dropoff_datetime\",\n", - " \"store_and_fwd_flag\"], axis=1, inplace=True, errors='ignore')\n", - "final_df.fillna(0, inplace=True)\n", - "final_df['fare_amount'] = final_df['fare_amount'].astype(\"float64\")\n", - "\n", - "\n", - "train_x, test_x, train_y, test_y = train_test_split(final_df.drop([\"fare_amount\"], axis=1),\n", - " final_df[\"fare_amount\"],\n", - " test_size=0.2,\n", - " random_state=42)\n", - "model = GradientBoostingRegressor()\n", - "model.fit(train_x, train_y)\n", - "\n", - "y_predict = model.predict(test_x)\n", - "\n", - "y_actual = test_y.values.flatten().tolist()\n", - "rmse = sqrt(mean_squared_error(y_actual, y_predict))\n", - "\n", - "sum_actuals = sum_errors = 0\n", - "\n", - "for actual_val, predict_val in zip(y_actual, y_predict):\n", - " abs_error = actual_val - predict_val\n", - " if abs_error < 0:\n", - " abs_error = abs_error * -1\n", - "\n", - " sum_errors = sum_errors + abs_error\n", - " sum_actuals = sum_actuals + actual_val\n", - "\n", - "mean_abs_percent_error = sum_errors / sum_actuals\n", - "print(\"Model MAPE:\")\n", - "print(mean_abs_percent_error)\n", - "print()\n", - "print(\"Model Accuracy:\")\n", - "print(1 - mean_abs_percent_error)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Materialize feature value into offline/online storage\n", - "\n", - "While Feathr can compute the feature value from the feature definition on-the-fly at request time, it can also pre-compute\n", - "and materialize the feature value to offline and/or online storage. \n", - "\n", - "We can push the generated features to the online store like below:\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "backfill_time = BackfillTime(start=datetime(\n", - " 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1))\n", - "redisSink = RedisSink(table_name=\"nycTaxiDemoFeature\")\n", - "settings = MaterializationSettings(\"nycTaxiTable\",\n", - " backfill_time=backfill_time,\n", - " sinks=[redisSink],\n", - " feature_names=[\"f_location_avg_fare\", \"f_location_max_fare\"])\n", - "\n", - "client.materialize_features(settings)\n", - "client.wait_job_to_finish(timeout_sec=500)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "We can then get the features from the online store (Redis):\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Fetching feature value for online inference\n", - "\n", - "For features that are already materialized by the previous step, their latest value can be queried via the client's\n", - "`get_online_features` or `multi_get_online_features` API." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "res = client.get_online_features('nycTaxiDemoFeature', '265', [\n", - " 'f_location_avg_fare', 'f_location_max_fare'])" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.multi_get_online_features(\"nycTaxiDemoFeature\", [\"239\", \"265\"], [\n", - " 'f_location_avg_fare', 'f_location_max_fare'])\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Registering and Fetching features\n", - "\n", - "We can also register the features with an Apache Atlas compatible service, such as Azure Purview, and share the registered features across teams:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.register_features()\n", - "client.list_registered_features(project_name=\"feathr_getting_started\")" - ] - } - ], - "metadata": { - "interpreter": { - "hash": "b3c5d8fd79e029a19bf620c04a250a0cafa2291ba3ed87972a3e2a099b099985" - }, - "kernelspec": { - "display_name": "Python 3.9.12 ('product_env': venv)", - "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.9.12" - } - }, - "nbformat": 4, - "nbformat_minor": 2 + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Feathr Feature Store on Azure Demo Notebook\n", + "\n", + "This notebook illustrates the use of Feature Store to create a model that predicts NYC Taxi fares. It includes these steps:\n", + "\n", + "\n", + "This tutorial demonstrates the key capabilities of Feathr, including:\n", + "\n", + "1. Install and set up Feathr with Azure\n", + "2. Create shareable features with Feathr feature definition configs.\n", + "3. Create a training dataset via point-in-time feature join.\n", + "4. Compute and write features.\n", + "5. Train a model using these features to predict fares.\n", + "6. Materialize feature value to online store.\n", + "7. Fetch feature value in real-time from online store for online scoring.\n", + "\n", + "In this tutorial, we use Feathr Feature Store to create a model that predicts NYC Taxi fares. The dataset comes from [here](https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page). The feature flow is as below:\n", + "\n", + "![Feature Flow](https://github.com/linkedin/feathr/blob/main/docs/images/feature_flow.png?raw=true)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Use Quick Start Template to Provision Azure Resources\n", + "\n", + "Feathr has native cloud integration. To use Feathr on Azure, you only need three steps:\n", + "\n", + "- Get the `Principal ID` of your account by running `az ad signed-in-user show --query id -o tsv` in the link below (Select \"Bash\" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI.\n", + "\n", + "[Launch Cloud Shell](https://shell.azure.com/bash)\n", + "\n", + "- Click the button below to deploy a minimal set of Feathr resources for demo purpose. You will need to fill in the `Principal ID` and `Resource Prefix`. You will need \"Owner\" permission of the selected subscription.\n", + "\n", + "[![Deploy to Azure](https://aka.ms/deploytoazurebutton)](https://portal.azure.com/#create/Microsoft.Template/uri/https%3A%2F%2Fraw.githubusercontent.com%2Flinkedin%2Ffeathr%2Fmain%2Fdocs%2Fhow-to-guides%2Fazure_resource_provision.json)\n", + "\n", + "- Run the cells below.\n", + "\n", + "And the architecture is as below. In the above template, we are using Synapse as Spark provider, use Azure Data Lake Gen2 as offline store, and use Redis as online store, Azure Purview (Apache Atlas compatible) as feature reigstry. \n", + "\n", + "\n", + "![Architecture](https://github.com/linkedin/feathr/blob/main/docs/images/architecture.png?raw=true)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Install Feathr \n", + "\n", + "Install Feathr using pip:\n", + "\n", + "`pip install -U feathr pandavro scikit-learn`" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Configure the required environment with Feathr Quick Start Template\n", + "\n", + "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. Run the code below to install Feathr, login to Azure to get the required credentials to access more cloud resources." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "**REQUIRED STEP: Fill in the resource prefix when provisioning the resources**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "resource_prefix = \"feathr_resource_prefix\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! pip install feathr azure-cli pandavro scikit-learn" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Login to Azure with a device code (You will see instructions in the output):" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! az login --use-device-code" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import glob\n", + "import os\n", + "import tempfile\n", + "from datetime import datetime, timedelta\n", + "from math import sqrt\n", + "\n", + "import pandas as pd\n", + "import pandavro as pdx\n", + "from feathr import FeathrClient\n", + "from feathr import BOOLEAN, FLOAT, INT32, ValueType\n", + "from feathr import Feature, DerivedFeature, FeatureAnchor\n", + "from feathr import BackfillTime, MaterializationSettings\n", + "from feathr import FeatureQuery, ObservationSettings\n", + "from feathr import RedisSink\n", + "from feathr import INPUT_CONTEXT, HdfsSource\n", + "from feathr import WindowAggTransformation\n", + "from feathr import TypedKey\n", + "from sklearn.metrics import mean_squared_error\n", + "from sklearn.model_selection import train_test_split\n", + "from azure.identity import DefaultAzureCredential\n", + "from azure.keyvault.secrets import SecretClient\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Get all the required credentials from Azure KeyVault" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Get all the required credentials from Azure Key Vault\n", + "key_vault_name=resource_prefix+\"kv\"\n", + "synapse_workspace_url=resource_prefix+\"syws\"\n", + "adls_account=resource_prefix+\"dls\"\n", + "adls_fs_name=resource_prefix+\"fs\"\n", + "purview_name=resource_prefix+\"purview\"\n", + "key_vault_uri = f\"https://{key_vault_name}.vault.azure.net\"\n", + "credential = DefaultAzureCredential(exclude_interactive_browser_credential=False)\n", + "client = SecretClient(vault_url=key_vault_uri, credential=credential)\n", + "secretName = \"FEATHR-ONLINE-STORE-CONN\"\n", + "retrieved_secret = client.get_secret(secretName).value\n", + "\n", + "# Get redis credentials; This is to parse Redis connection string.\n", + "redis_port=retrieved_secret.split(',')[0].split(\":\")[1]\n", + "redis_host=retrieved_secret.split(',')[0].split(\":\")[0]\n", + "redis_password=retrieved_secret.split(',')[1].split(\"password=\",1)[1]\n", + "redis_ssl=retrieved_secret.split(',')[2].split(\"ssl=\",1)[1]\n", + "\n", + "# Set the resource link\n", + "os.environ['spark_config__azure_synapse__dev_url'] = f'https://{synapse_workspace_url}.dev.azuresynapse.net'\n", + "os.environ['spark_config__azure_synapse__pool_name'] = 'spark31'\n", + "os.environ['spark_config__azure_synapse__workspace_dir'] = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_project'\n", + "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", + "os.environ['online_store__redis__host'] = redis_host\n", + "os.environ['online_store__redis__port'] = redis_port\n", + "os.environ['online_store__redis__ssl_enabled'] = redis_ssl\n", + "os.environ['REDIS_PASSWORD']=redis_password\n", + "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", + "feathr_output_path = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_output'" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Configure the required environment (Don't need to update if using the above Quick Start Template)\n", + "\n", + "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. If you use Feathr CLI to create a workspace, you should have a folder with a file called `feathr_config.yaml` in it with all the required configurations. Otherwise, update the configuration below.\n", + "\n", + "The code below will write this configuration string to a temporary location and load it to Feathr. Please still refer to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It should also have more explanations on the meaning of each variable." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import tempfile\n", + "yaml_config = \"\"\"\n", + "# Please refer to https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml for explanations on the meaning of each field.\n", + "api_version: 1\n", + "project_config:\n", + " project_name: 'feathr_getting_started'\n", + " required_environment_variables:\n", + " - 'REDIS_PASSWORD'\n", + " - 'AZURE_CLIENT_ID'\n", + " - 'AZURE_TENANT_ID'\n", + " - 'AZURE_CLIENT_SECRET'\n", + "offline_store:\n", + " adls:\n", + " adls_enabled: true\n", + " wasb:\n", + " wasb_enabled: true\n", + " s3:\n", + " s3_enabled: false\n", + " s3_endpoint: 's3.amazonaws.com'\n", + " jdbc:\n", + " jdbc_enabled: false\n", + " jdbc_database: 'feathrtestdb'\n", + " jdbc_table: 'feathrtesttable'\n", + " snowflake:\n", + " url: \"dqllago-ol19457.snowflakecomputing.com\"\n", + " user: \"feathrintegration\"\n", + " role: \"ACCOUNTADMIN\"\n", + "spark_config:\n", + " spark_cluster: 'azure_synapse'\n", + " spark_result_output_parts: '1'\n", + " azure_synapse:\n", + " dev_url: 'https://feathrazuretest3synapse.dev.azuresynapse.net'\n", + " pool_name: 'spark3'\n", + " workspace_dir: 'abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/feathr_getting_started'\n", + " executor_size: 'Small'\n", + " executor_num: 4\n", + " feathr_runtime_location: wasbs://public@azurefeathrstorage.blob.core.windows.net/feathr-assembly-LATEST.jar\n", + " databricks:\n", + " workspace_instance_url: 'https://adb-2474129336842816.16.azuredatabricks.net'\n", + " config_template: {'run_name':'','new_cluster':{'spark_version':'9.1.x-scala2.12','node_type_id':'Standard_D3_v2','num_workers':2,'spark_conf':{}},'libraries':[{'jar':''}],'spark_jar_task':{'main_class_name':'','parameters':['']}}\n", + " work_dir: 'dbfs:/feathr_getting_started'\n", + " feathr_runtime_location: https://azurefeathrstorage.blob.core.windows.net/public/feathr-assembly-LATEST.jar\n", + "online_store:\n", + " redis:\n", + " host: 'feathrazuretest3redis.redis.cache.windows.net'\n", + " port: 6380\n", + " ssl_enabled: True\n", + "feature_registry:\n", + " purview:\n", + " type_system_initialization: true\n", + " purview_name: 'feathrazuretest3-purview1'\n", + " delimiter: '__'\n", + "\"\"\"\n", + "tmp = tempfile.NamedTemporaryFile(mode='w', delete=False)\n", + "with open(tmp.name, \"w\") as text_file:\n", + " text_file.write(yaml_config)\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Setup necessary environment variables (Skip if using the above Quick Start Template)\n", + "\n", + "You should setup the environment variables in order to run this sample. More environment variables can be set by referring to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It also has more explanations on the meaning of each variable.\n", + "\n", + "To run this notebook, for Azure users, you need AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET and REDIS_PASSWORD.\n", + "To run this notebook, for Databricks useres, you need DATABRICKS_WORKSPACE_TOKEN_VALUE and REDIS_PASSWORD." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Initialize Feathr Client" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client = FeathrClient(config_path=tmp.name)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## View the data\n", + "\n", + "In this tutorial, we use Feathr Feature Store to create a model that predicts NYC Taxi fares. The dataset comes from [here](https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page). The data is as below" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import pandas as pd\n", + "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/green_tripdata_2020-04_with_index.csv\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Defining Features with Feathr\n", + "\n", + "In Feathr, a feature is viewed as a function, mapping from entity id or key, and timestamp to a feature value. For more details on feature definition, please refer to the [Feathr Feature Definition Guide](https://github.com/linkedin/feathr/blob/main/docs/concepts/feature-definition.md)\n", + "\n", + "\n", + "1. The typed key (a.k.a. entity id) identifies the subject of feature, e.g. a user id, 123.\n", + "2. The feature name is the aspect of the entity that the feature is indicating, e.g. the age of the user.\n", + "3. The feature value is the actual value of that aspect at a particular time, e.g. the value is 30 at year 2022.\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Note that, in some cases, such as features defined on top of request data, may have no entity key or timestamp.\n", + "It is merely a function/transformation executing against request data at runtime.\n", + "For example, the day of week of the request, which is calculated by converting the request UNIX timestamp.\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Define Sources Section with UDFs\n", + "A feature source is needed for anchored features that describes the raw data in which the feature values are computed from. See the python documentation to get the details on each input column.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from pyspark.sql import SparkSession, DataFrame\n", + "def feathr_udf_day_calc(df: DataFrame) -> DataFrame:\n", + " from pyspark.sql.functions import dayofweek, dayofyear, col\n", + " df = df.withColumn(\"fare_amount_cents\", col(\"fare_amount\")*100)\n", + " return df\n", + "\n", + "batch_source = HdfsSource(name=\"nycTaxiBatchSource\",\n", + " path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/green_tripdata_2020-04_with_index.csv\",\n", + " event_timestamp_column=\"lpep_dropoff_datetime\",\n", + " preprocessing=feathr_udf_day_calc,\n", + " timestamp_format=\"yyyy-MM-dd HH:mm:ss\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Define Anchors and Features\n", + "A feature is called an anchored feature when the feature is directly extracted from the source data, rather than computed on top of other features. The latter case is called derived feature." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "f_trip_distance = Feature(name=\"f_trip_distance\",\n", + " feature_type=FLOAT, transform=\"trip_distance\")\n", + "f_trip_time_duration = Feature(name=\"f_trip_time_duration\",\n", + " feature_type=INT32,\n", + " transform=\"(to_unix_timestamp(lpep_dropoff_datetime) - to_unix_timestamp(lpep_pickup_datetime))/60\")\n", + "\n", + "features = [\n", + " f_trip_distance,\n", + " f_trip_time_duration,\n", + " Feature(name=\"f_is_long_trip_distance\",\n", + " feature_type=BOOLEAN,\n", + " transform=\"cast_float(trip_distance)>30\"),\n", + " Feature(name=\"f_day_of_week\",\n", + " feature_type=INT32,\n", + " transform=\"dayofweek(lpep_dropoff_datetime)\"),\n", + "]\n", + "\n", + "request_anchor = FeatureAnchor(name=\"request_features\",\n", + " source=INPUT_CONTEXT,\n", + " features=features)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Window aggregation features\n", + "\n", + "For window aggregation features, see the supported fields below:\n", + "\n", + "Note that the `agg_func` should be any of these:\n", + "\n", + "| Aggregation Type | Input Type | Description |\n", + "| --- | --- | --- |\n", + "|SUM, COUNT, MAX, MIN, AVG\t|Numeric|Applies the the numerical operation on the numeric inputs. |\n", + "|MAX_POOLING, MIN_POOLING, AVG_POOLING\t| Numeric Vector | Applies the max/min/avg operation on a per entry bassis for a given a collection of numbers.|\n", + "|LATEST| Any |Returns the latest not-null values from within the defined time window |\n", + "\n", + "\n", + "After you have defined features and sources, bring them together to build an anchor:\n", + "\n", + "\n", + "Note that if the data source is from the observation data, the `source` section should be `INPUT_CONTEXT` to indicate the source of those defined anchors." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "location_id = TypedKey(key_column=\"DOLocationID\",\n", + " key_column_type=ValueType.INT32,\n", + " description=\"location id in NYC\",\n", + " full_name=\"nyc_taxi.location_id\")\n", + "agg_features = [Feature(name=\"f_location_avg_fare\",\n", + " key=location_id,\n", + " feature_type=FLOAT,\n", + " transform=WindowAggTransformation(agg_expr=\"cast_float(fare_amount)\",\n", + " agg_func=\"AVG\",\n", + " window=\"90d\")),\n", + " Feature(name=\"f_location_max_fare\",\n", + " key=location_id,\n", + " feature_type=FLOAT,\n", + " transform=WindowAggTransformation(agg_expr=\"cast_float(fare_amount)\",\n", + " agg_func=\"MAX\",\n", + " window=\"90d\")),\n", + " Feature(name=\"f_location_total_fare_cents\",\n", + " key=location_id,\n", + " feature_type=FLOAT,\n", + " transform=WindowAggTransformation(agg_expr=\"fare_amount_cents\",\n", + " agg_func=\"SUM\",\n", + " window=\"90d\")),\n", + " ]\n", + "\n", + "agg_anchor = FeatureAnchor(name=\"aggregationFeatures\",\n", + " source=batch_source,\n", + " features=agg_features)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Derived Features Section\n", + "Derived features are the features that are computed from other features. They could be computed from anchored features, or other derived features." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "f_trip_time_distance = DerivedFeature(name=\"f_trip_time_distance\",\n", + " feature_type=FLOAT,\n", + " input_features=[\n", + " f_trip_distance, f_trip_time_duration],\n", + " transform=\"f_trip_distance * f_trip_time_duration\")\n", + "\n", + "f_trip_time_rounded = DerivedFeature(name=\"f_trip_time_rounded\",\n", + " feature_type=INT32,\n", + " input_features=[f_trip_time_duration],\n", + " transform=\"f_trip_time_duration % 10\")\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "And then we need to build those features so that it can be consumed later. Note that we have to build both the \"anchor\" and the \"derived\" features (which is not anchored to a source)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.build_features(anchor_list=[agg_anchor, request_anchor], derived_feature_list=[\n", + " f_trip_time_distance, f_trip_time_rounded])" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Create training data using point-in-time correct feature join\n", + "\n", + "A training dataset usually contains entity id columns, multiple feature columns, event timestamp column and label/target column. \n", + "\n", + "To create a training dataset using Feathr, one needs to provide a feature join configuration file to specify\n", + "what features and how these features should be joined to the observation data. \n", + "\n", + "To learn more on this topic, please refer to [Point-in-time Correctness](https://github.com/linkedin/feathr/blob/main/docs/concepts/point-in-time-join.md)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "if client.spark_runtime == 'databricks':\n", + " output_path = 'dbfs:/feathrazure_test.avro'\n", + "else:\n", + " output_path = feathr_output_path\n", + "\n", + "\n", + "feature_query = FeatureQuery(\n", + " feature_list=[\"f_location_avg_fare\", \"f_trip_time_rounded\", \"f_is_long_trip_distance\", \"f_location_total_fare_cents\"], key=location_id)\n", + "settings = ObservationSettings(\n", + " observation_path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/green_tripdata_2020-04_with_index.csv\",\n", + " event_timestamp_column=\"lpep_dropoff_datetime\",\n", + " timestamp_format=\"yyyy-MM-dd HH:mm:ss\")\n", + "client.get_offline_features(observation_settings=settings,\n", + " feature_query=feature_query,\n", + " output_path=output_path)\n", + "client.wait_job_to_finish(timeout_sec=500)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Download the result and show the result\n", + "\n", + "Let's use the helper function `get_result_df` to download the result and view it:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "def get_result_df(client: FeathrClient) -> pd.DataFrame:\n", + " \"\"\"Download the job result dataset from cloud as a Pandas dataframe.\"\"\"\n", + " res_url = client.get_job_result_uri(block=True, timeout_sec=600)\n", + " tmp_dir = tempfile.TemporaryDirectory()\n", + " client.feathr_spark_laucher.download_result(result_path=res_url, local_folder=tmp_dir.name)\n", + " dataframe_list = []\n", + " # assuming the result are in avro format\n", + " for file in glob.glob(os.path.join(tmp_dir.name, '*.avro')):\n", + " dataframe_list.append(pdx.read_avro(file))\n", + " vertical_concat_df = pd.concat(dataframe_list, axis=0)\n", + " tmp_dir.cleanup()\n", + " return vertical_concat_df\n", + "\n", + "df_res = get_result_df(client)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "df_res" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Train a machine learning model\n", + "After getting all the features, let's train a machine learning model with the converted feature by Feathr:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# remove columns\n", + "from sklearn.ensemble import GradientBoostingRegressor\n", + "final_df = df_res\n", + "final_df.drop([\"lpep_pickup_datetime\", \"lpep_dropoff_datetime\",\n", + " \"store_and_fwd_flag\"], axis=1, inplace=True, errors='ignore')\n", + "final_df.fillna(0, inplace=True)\n", + "final_df['fare_amount'] = final_df['fare_amount'].astype(\"float64\")\n", + "\n", + "\n", + "train_x, test_x, train_y, test_y = train_test_split(final_df.drop([\"fare_amount\"], axis=1),\n", + " final_df[\"fare_amount\"],\n", + " test_size=0.2,\n", + " random_state=42)\n", + "model = GradientBoostingRegressor()\n", + "model.fit(train_x, train_y)\n", + "\n", + "y_predict = model.predict(test_x)\n", + "\n", + "y_actual = test_y.values.flatten().tolist()\n", + "rmse = sqrt(mean_squared_error(y_actual, y_predict))\n", + "\n", + "sum_actuals = sum_errors = 0\n", + "\n", + "for actual_val, predict_val in zip(y_actual, y_predict):\n", + " abs_error = actual_val - predict_val\n", + " if abs_error < 0:\n", + " abs_error = abs_error * -1\n", + "\n", + " sum_errors = sum_errors + abs_error\n", + " sum_actuals = sum_actuals + actual_val\n", + "\n", + "mean_abs_percent_error = sum_errors / sum_actuals\n", + "print(\"Model MAPE:\")\n", + "print(mean_abs_percent_error)\n", + "print()\n", + "print(\"Model Accuracy:\")\n", + "print(1 - mean_abs_percent_error)\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Materialize feature value into offline/online storage\n", + "\n", + "While Feathr can compute the feature value from the feature definition on-the-fly at request time, it can also pre-compute\n", + "and materialize the feature value to offline and/or online storage. \n", + "\n", + "We can push the generated features to the online store like below:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "backfill_time = BackfillTime(start=datetime(\n", + " 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1))\n", + "redisSink = RedisSink(table_name=\"nycTaxiDemoFeature\")\n", + "settings = MaterializationSettings(\"nycTaxiTable\",\n", + " backfill_time=backfill_time,\n", + " sinks=[redisSink],\n", + " feature_names=[\"f_location_avg_fare\", \"f_location_max_fare\"])\n", + "\n", + "client.materialize_features(settings)\n", + "client.wait_job_to_finish(timeout_sec=500)\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "We can then get the features from the online store (Redis):\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Fetching feature value for online inference\n", + "\n", + "For features that are already materialized by the previous step, their latest value can be queried via the client's\n", + "`get_online_features` or `multi_get_online_features` API." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "res = client.get_online_features('nycTaxiDemoFeature', '265', [\n", + " 'f_location_avg_fare', 'f_location_max_fare'])" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.multi_get_online_features(\"nycTaxiDemoFeature\", [\"239\", \"265\"], [\n", + " 'f_location_avg_fare', 'f_location_max_fare'])\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Registering and Fetching features\n", + "\n", + "We can also register the features with an Apache Atlas compatible service, such as Azure Purview, and share the registered features across teams:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.register_features()\n", + "client.list_registered_features(project_name=\"feathr_getting_started\")" + ] + } + ], + "metadata": { + "interpreter": { + "hash": "b3c5d8fd79e029a19bf620c04a250a0cafa2291ba3ed87972a3e2a099b099985" + }, + "kernelspec": { + "display_name": "Python 3.9.12 ('product_env': venv)", + "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.9.12" + } + }, + "nbformat": 4, + "nbformat_minor": 2 } diff --git a/feathr_project/feathrcli/data/feathr_user_workspace/product_recommendation_demo.ipynb b/feathr_project/feathrcli/data/feathr_user_workspace/product_recommendation_demo.ipynb index 99bf54254..d3c297eae 100644 --- a/feathr_project/feathrcli/data/feathr_user_workspace/product_recommendation_demo.ipynb +++ b/feathr_project/feathrcli/data/feathr_user_workspace/product_recommendation_demo.ipynb @@ -1,755 +1,755 @@ { - "cells": [ - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "# Feathr Feature Store on Azure Demo Notebook\n", - "\n", - "This notebook illustrates the use of Feature Store to create a model that predicts product ratings. It includes these steps:\n", - "\n", - "\n", - "This tutorial demonstrates the key capabilities of Feathr, including:\n", - "\n", - "1. Install and set up Feathr with Azure\n", - "2. Create shareable features with Feathr feature definition configs.\n", - "3. Create a training dataset via point-in-time feature join.\n", - "4. Compute and write features.\n", - "5. Train a model using these features to predict fares.\n", - "6. Materialize feature value to online store.\n", - "7. Fetch feature value in real-time from online store for online scoring.\n", - "\n", - "In this tutorial, we use Feathr Feature Store to create a model that predict users rating for a product. The feature flow is as below:\n", - "![Feature Flow](https://github.com/linkedin/feathr/blob/main/docs/images/product_recommendation_overview.png?raw=true)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Use Quick Start Template to Provision Azure Resources\n", - "\n", - "Feathr has native cloud integration. To use Feathr on Azure, you only need three steps:\n", - "\n", - "- Get the `Principal ID` of your account by running `az ad signed-in-user show --query objectId -o tsv` in the link below (Select \"Bash\" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI.\n", - "\n", - "[Launch Cloud Shell](https://shell.azure.com/bash)\n", - "\n", - "- Click the button below to deploy a minimal set of Feathr resources for demo purpose. You will need to fill in the `Principal ID` and `Resource Prefix`. You will need \"Owner\" permission of the selected subscription.\n", - "\n", - "[![Deploy to Azure](https://aka.ms/deploytoazurebutton)](https://portal.azure.com/#create/Microsoft.Template/uri/https%3A%2F%2Fraw.githubusercontent.com%2Flinkedin%2Ffeathr%2Fmain%2Fdocs%2Fhow-to-guides%2Fazure_resource_provision.json)\n", - "\n", - "- Run the cells below.\n", - "\n", - "And the architecture is as below. In the above template, we are using Synapse as Spark provider, use Azure Data Lake Gen2 as offline store, and use Redis as online store, Azure Purview (Apache Atlas compatible) as feature reigstry. \n", - "\n", - "\n", - "![Architecture](https://github.com/linkedin/feathr/blob/main/docs/images/architecture.png?raw=true)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Install Feathr \n", - "\n", - "Install Feathr using pip:\n", - "\n", - "`pip install -U feathr pandavro scikit-learn`" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Configure the required environment with Feathr Quick Start Template\n", - "\n", - "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. Run the code below to install Feathr, login to Azure to get the required credentials to access more cloud resources." - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "**REQUIRED STEP: Fill in the resource prefix when provisioning the resources**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "resource_prefix = \"feathr_resource_prefix\"" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "! pip install feathr azure-cli pandavro scikit-learn" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Login to Azure with a device code (You will see instructions in the output):" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "! az login --use-device-code" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "import glob\n", - "import os\n", - "import tempfile\n", - "from datetime import datetime, timedelta\n", - "from math import sqrt\n", - "\n", - "import pandas as pd\n", - "import pandavro as pdx\n", - "from feathr import FeathrClient\n", - "from feathr import BOOLEAN, FLOAT, INT32, ValueType\n", - "from feathr import Feature, DerivedFeature, FeatureAnchor\n", - "from feathr import BackfillTime, MaterializationSettings\n", - "from feathr import FeatureQuery, ObservationSettings\n", - "from feathr import RedisSink\n", - "from feathr import INPUT_CONTEXT, HdfsSource\n", - "from feathr import WindowAggTransformation\n", - "from feathr import TypedKey\n", - "from sklearn.metrics import mean_squared_error\n", - "from sklearn.model_selection import train_test_split\n", - "from azure.identity import DefaultAzureCredential\n", - "from azure.keyvault.secrets import SecretClient\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Get all the required credentials from Azure KeyVault" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Get all the required credentials from Azure Key Vault\n", - "key_vault_name=resource_prefix+\"kv\"\n", - "synapse_workspace_url=resource_prefix+\"syws\"\n", - "adls_account=resource_prefix+\"dls\"\n", - "adls_fs_name=resource_prefix+\"fs\"\n", - "purview_name=resource_prefix+\"purview\"\n", - "key_vault_uri = f\"https://{key_vault_name}.vault.azure.net\"\n", - "credential = DefaultAzureCredential(exclude_interactive_browser_credential=False)\n", - "client = SecretClient(vault_url=key_vault_uri, credential=credential)\n", - "secretName = \"FEATHR-ONLINE-STORE-CONN\"\n", - "retrieved_secret = client.get_secret(secretName).value\n", - "\n", - "# Get redis credentials; This is to parse Redis connection string.\n", - "redis_port=retrieved_secret.split(',')[0].split(\":\")[1]\n", - "redis_host=retrieved_secret.split(',')[0].split(\":\")[0]\n", - "redis_password=retrieved_secret.split(',')[1].split(\"password=\",1)[1]\n", - "redis_ssl=retrieved_secret.split(',')[2].split(\"ssl=\",1)[1]\n", - "\n", - "# Set the resource link\n", - "os.environ['spark_config__azure_synapse__dev_url'] = f'https://{synapse_workspace_url}.dev.azuresynapse.net'\n", - "os.environ['spark_config__azure_synapse__pool_name'] = 'spark31'\n", - "os.environ['spark_config__azure_synapse__workspace_dir'] = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_project'\n", - "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", - "os.environ['online_store__redis__host'] = redis_host\n", - "os.environ['online_store__redis__port'] = redis_port\n", - "os.environ['online_store__redis__ssl_enabled'] = redis_ssl\n", - "os.environ['REDIS_PASSWORD']=redis_password\n", - "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", - "feathr_output_path = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_output'" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Prerequisite: Configure the required environment (Don't need to update if using the above Quick Start Template)\n", - "\n", - "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. If you use Feathr CLI to create a workspace, you should have a folder with a file called `feathr_config.yaml` in it with all the required configurations. Otherwise, update the configuration below.\n", - "\n", - "The code below will write this configuration string to a temporary location and load it to Feathr. Please still refer to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It should also have more explanations on the meaning of each variable." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "import tempfile\n", - "yaml_config = \"\"\"\n", - "# Please refer to https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml for explanations on the meaning of each field.\n", - "api_version: 1\n", - "project_config:\n", - " project_name: 'feathr_getting_started'\n", - " required_environment_variables:\n", - " - 'REDIS_PASSWORD'\n", - " - 'AZURE_CLIENT_ID'\n", - " - 'AZURE_TENANT_ID'\n", - " - 'AZURE_CLIENT_SECRET'\n", - "offline_store:\n", - " adls:\n", - " adls_enabled: true\n", - " wasb:\n", - " wasb_enabled: true\n", - " s3:\n", - " s3_enabled: false\n", - " s3_endpoint: 's3.amazonaws.com'\n", - " jdbc:\n", - " jdbc_enabled: false\n", - " jdbc_database: 'feathrtestdb'\n", - " jdbc_table: 'feathrtesttable'\n", - " snowflake:\n", - " url: \"dqllago-ol19457.snowflakecomputing.com\"\n", - " user: \"feathrintegration\"\n", - " role: \"ACCOUNTADMIN\"\n", - "spark_config:\n", - " spark_cluster: 'azure_synapse'\n", - " spark_result_output_parts: '1'\n", - " azure_synapse:\n", - " dev_url: 'https://feathrazuretest3synapse.dev.azuresynapse.net'\n", - " pool_name: 'spark3'\n", - " workspace_dir: 'abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/feathr_getting_started'\n", - " executor_size: 'Small'\n", - " executor_num: 4\n", - " feathr_runtime_location: wasbs://public@azurefeathrstorage.blob.core.windows.net/feathr-assembly-LATEST.jar\n", - " databricks:\n", - " workspace_instance_url: 'https://adb-2474129336842816.16.azuredatabricks.net'\n", - " config_template: {'run_name':'','new_cluster':{'spark_version':'9.1.x-scala2.12','node_type_id':'Standard_D3_v2','num_workers':2,'spark_conf':{}},'libraries':[{'jar':''}],'spark_jar_task':{'main_class_name':'','parameters':['']}}\n", - " work_dir: 'dbfs:/feathr_getting_started'\n", - " feathr_runtime_location: https://azurefeathrstorage.blob.core.windows.net/public/feathr-assembly-LATEST.jar\n", - "online_store:\n", - " redis:\n", - " host: 'feathrazuretest3redis.redis.cache.windows.net'\n", - " port: 6380\n", - " ssl_enabled: True\n", - "feature_registry:\n", - " purview:\n", - " type_system_initialization: true\n", - " purview_name: 'feathrazuretest3-purview1'\n", - " delimiter: '__'\n", - "\"\"\"\n", - "tmp = tempfile.NamedTemporaryFile(mode='w', delete=False)\n", - "with open(tmp.name, \"w\") as text_file:\n", - " text_file.write(yaml_config)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Setup necessary environment variables (Skip if using the above Quick Start Template)\n", - "\n", - "You should setup the environment variables in order to run this sample. More environment variables can be set by referring to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It also has more explanations on the meaning of each variable.\n", - "\n", - "To run this notebook, for Azure users, you need AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET and REDIS_PASSWORD.\n", - "To run this notebook, for Databricks useres, you need DATABRICKS_WORKSPACE_TOKEN_VALUE and REDIS_PASSWORD." - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "# Initialize Feathr Client" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client = FeathrClient(config_path=tmp.name)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## View the data\n", - "\n", - "In this tutorial, we use Feathr Feature Store to create a model that predicts users product rating. To make it simple, let's just predict users' rating for one product. (We will expand the exmaple to predict ratings for arbitrary product in future tutorials.)\n", - "\n", - "We have 3 datasets to work with: one observation dataset(a.k.a. label dataset) and two raw datasets to generate features." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Observation dataset(a.k.a. label dataset)\n", - "# Observation dataset usually comes with a event_timestamp to denote when the observation happened.\n", - "# The label here is product_rating. Our model objective is to predict a user's rating for this product.\n", - "import pandas as pd\n", - "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/product_recommendation_sample/user_observation_mock_data.csv\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# User profile dataset\n", - "# Used to generate user features\n", - "import pandas as pd\n", - "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/product_recommendation_sample/user_profile_mock_data.csv\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# User purchase history dataset.\n", - "# Used to generate user features. This is activity type data, so we need to use aggregation to genearte features.\n", - "import pandas as pd\n", - "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/product_recommendation_sample/user_purchase_history_mock_data.csv\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Defining Features with Feathr\n", - "\n", - "In Feathr, a feature is viewed as a function, mapping from entity id or key, and timestamp to a feature value. For more details on feature definition, please refer to the [Feathr Feature Definition Guide](https://github.com/linkedin/feathr/blob/main/docs/concepts/feature-definition.md)\n", - "\n", - "\n", - "1. The typed key (a.k.a. entity id) identifies the subject of feature, e.g. a user id, 123.\n", - "2. The feature name is the aspect of the entity that the feature is indicating, e.g. the age of the user.\n", - "3. The feature value is the actual value of that aspect at a particular time, e.g. the value is 30 at year 2022.\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Note that, in some cases, such as features defined on top of request data, may have no entity key or timestamp.\n", - "It is merely a function/transformation executing against request data at runtime.\n", - "For example, the day of week of the request, which is calculated by converting the request UNIX timestamp.\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Define Sources Section with UDFs\n", - "A feature source is needed for anchored features that describes the raw data in which the feature values are computed from. See the python documentation to get the details on each input column.\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from pyspark.sql import SparkSession, DataFrame\n", - "def feathr_udf_preprocessing(df: DataFrame) -> DataFrame:\n", - " from pyspark.sql.functions import col\n", - " df = df.withColumn(\"tax_rate_decimal\", col(\"tax_rate\")/100)\n", - " df.show(10)\n", - " return df\n", - "\n", - "batch_source = HdfsSource(name=\"userProfileData\",\n", - " path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/product_recommendation_sample/user_profile_mock_data.csv\",\n", - " preprocessing=feathr_udf_preprocessing)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Define Anchors and Features\n", - "A feature is called an anchored feature when the feature is directly extracted from the source data, rather than computed on top of other features. The latter case is called derived feature." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "user_id = TypedKey(key_column=\"user_id\",\n", - " key_column_type=ValueType.INT32,\n", - " description=\"user id\",\n", - " full_name=\"product_recommendation.user_id\")\n", - "\n", - "feature_user_age = Feature(name=\"feature_user_age\",\n", - " key=user_id,\n", - " feature_type=INT32, transform=\"age\")\n", - "feature_user_tax_rate = Feature(name=\"feature_user_tax_rate\",\n", - " key=user_id,\n", - " feature_type=FLOAT,\n", - " transform=\"tax_rate_decimal\")\n", - "feature_user_gift_card_balance = Feature(name=\"feature_user_gift_card_balance\",\n", - " key=user_id,\n", - " feature_type=FLOAT,\n", - " transform=\"gift_card_balance\")\n", - "feature_user_has_valid_credit_card = Feature(name=\"feature_user_has_valid_credit_card\",\n", - " key=user_id,\n", - " feature_type=BOOLEAN,\n", - " transform=\"number_of_credit_cards > 0\")\n", - " \n", - "features = [\n", - " feature_user_age,\n", - " feature_user_tax_rate,\n", - " feature_user_gift_card_balance,\n", - " feature_user_has_valid_credit_card\n", - "]\n", - "\n", - "request_anchor = FeatureAnchor(name=\"anchored_features\",\n", - " source=batch_source,\n", - " features=features)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Window aggregation features\n", - "\n", - "For window aggregation features, see the supported fields below:\n", - "\n", - "Note that the `agg_func` should be any of these:\n", - "\n", - "| Aggregation Type | Input Type | Description |\n", - "| --- | --- | --- |\n", - "|SUM, COUNT, MAX, MIN, AVG\t|Numeric|Applies the the numerical operation on the numeric inputs. |\n", - "|MAX_POOLING, MIN_POOLING, AVG_POOLING\t| Numeric Vector | Applies the max/min/avg operation on a per entry bassis for a given a collection of numbers.|\n", - "|LATEST| Any |Returns the latest not-null values from within the defined time window |\n", - "\n", - "\n", - "After you have defined features and sources, bring them together to build an anchor:\n", - "\n", - "\n", - "Note that if the data source is from the observation data, the `source` section should be `INPUT_CONTEXT` to indicate the source of those defined anchors." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "purchase_history_data = HdfsSource(name=\"purchase_history_data\",\n", - " path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/product_recommendation_sample/user_purchase_history_mock_data.csv\",\n", - " event_timestamp_column=\"purchase_date\",\n", - " timestamp_format=\"yyyy-MM-dd\")\n", - " \n", - "agg_features = [Feature(name=\"feature_user_totla_purchase_in_90days\",\n", - " key=user_id,\n", - " feature_type=FLOAT,\n", - " transform=WindowAggTransformation(agg_expr=\"cast_float(purchase_amount)\",\n", - " agg_func=\"AVG\",\n", - " window=\"90d\"))\n", - " ]\n", - "\n", - "agg_anchor = FeatureAnchor(name=\"aggregationFeatures\",\n", - " source=purchase_history_data,\n", - " features=agg_features)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Derived Features Section\n", - "Derived features are the features that are computed from other features. They could be computed from anchored features, or other derived features." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "feature_user_purchasing_power = DerivedFeature(name=\"feature_user_purchasing_power\",\n", - " key=user_id,\n", - " feature_type=FLOAT,\n", - " input_features=[\n", - " feature_user_gift_card_balance, feature_user_has_valid_credit_card],\n", - " transform=\"feature_user_gift_card_balance + if_else(feature_user_has_valid_credit_card, 100, 0)\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "And then we need to build those features so that it can be consumed later. Note that we have to build both the \"anchor\" and the \"derived\" features (which is not anchored to a source)." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.build_features(anchor_list=[agg_anchor, request_anchor], derived_feature_list=[\n", - " feature_user_purchasing_power])" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Create training data using point-in-time correct feature join\n", - "\n", - "A training dataset usually contains entity id columns, multiple feature columns, event timestamp column and label/target column. \n", - "\n", - "To create a training dataset using Feathr, one needs to provide a feature join configuration file to specify\n", - "what features and how these features should be joined to the observation data. \n", - "\n", - "To learn more on this topic, please refer to [Point-in-time Correctness](https://github.com/linkedin/feathr/blob/main/docs/concepts/point-in-time-join.md)\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "if client.spark_runtime == 'databricks':\n", - " output_path = 'dbfs:/feathrazure_test.avro'\n", - "else:\n", - " output_path = feathr_output_path\n", - "\n", - "\n", - "feature_query = FeatureQuery(\n", - " feature_list=[\"feature_user_age\", \n", - " \"feature_user_tax_rate\", \n", - " \"feature_user_gift_card_balance\", \n", - " \"feature_user_has_valid_credit_card\", \n", - " \"feature_user_totla_purchase_in_90days\",\n", - " \"feature_user_purchasing_power\"], key=user_id)\n", - "settings = ObservationSettings(\n", - " observation_path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/product_recommendation_sample/user_observation_mock_data.csv\",\n", - " event_timestamp_column=\"event_timestamp\",\n", - " timestamp_format=\"yyyy-MM-dd\")\n", - "client.get_offline_features(observation_settings=settings,\n", - " feature_query=feature_query,\n", - " output_path=output_path)\n", - "client.wait_job_to_finish(timeout_sec=500)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Download the result and show the result\n", - "\n", - "Let's use the helper function `get_result_df` to download the result and view it:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "def get_result_df(client: FeathrClient) -> pd.DataFrame:\n", - " \"\"\"Download the job result dataset from cloud as a Pandas dataframe.\"\"\"\n", - " res_url = client.get_job_result_uri(block=True, timeout_sec=600)\n", - " tmp_dir = tempfile.TemporaryDirectory()\n", - " client.feathr_spark_laucher.download_result(result_path=res_url, local_folder=tmp_dir.name)\n", - " dataframe_list = []\n", - " # assuming the result are in avro format\n", - " for file in glob.glob(os.path.join(tmp_dir.name, '*.avro')):\n", - " dataframe_list.append(pdx.read_avro(file))\n", - " vertical_concat_df = pd.concat(dataframe_list, axis=0)\n", - " tmp_dir.cleanup()\n", - " return vertical_concat_df\n", - "\n", - "df_res = get_result_df(client)\n", - "\n", - "df_res" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Train a machine learning model\n", - "After getting all the features, let's train a machine learning model with the converted feature by Feathr:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# drop non-feature columns\n", - "from sklearn.ensemble import GradientBoostingRegressor\n", - "final_df = df_res\n", - "final_df.drop([\"event_timestamp\"], axis=1, inplace=True, errors='ignore')\n", - "final_df.fillna(0, inplace=True)\n", - "final_df['product_rating'] = final_df['product_rating'].astype(\"float64\")\n", - "\n", - "train_x, test_x, train_y, test_y = train_test_split(final_df.drop([\"product_rating\"], axis=1),\n", - " final_df[\"product_rating\"],\n", - " test_size=0.2,\n", - " random_state=42)\n", - "model = GradientBoostingRegressor()\n", - "model.fit(train_x, train_y)\n", - "\n", - "y_predict = model.predict(test_x)\n", - "\n", - "y_actual = test_y.values.flatten().tolist()\n", - "rmse = sqrt(mean_squared_error(y_actual, y_predict))\n", - "\n", - "sum_actuals = sum_errors = 0\n", - "\n", - "for actual_val, predict_val in zip(y_actual, y_predict):\n", - " abs_error = actual_val - predict_val\n", - " if abs_error < 0:\n", - " abs_error = abs_error * -1\n", - "\n", - " sum_errors = sum_errors + abs_error\n", - " sum_actuals = sum_actuals + actual_val\n", - "\n", - "mean_abs_percent_error = sum_errors / sum_actuals\n", - "print(\"Model MAPE:\")\n", - "print(mean_abs_percent_error)\n", - "print()\n", - "print(\"Model Accuracy:\")\n", - "print(1 - mean_abs_percent_error)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Materialize feature value into offline/online storage\n", - "\n", - "While Feathr can compute the feature value from the feature definition on-the-fly at request time, it can also pre-compute\n", - "and materialize the feature value to offline and/or online storage. \n", - "\n", - "We can push the generated features to the online store like below:\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "backfill_time = BackfillTime(start=datetime(\n", - " 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1))\n", - "redisSink = RedisSink(table_name=\"productRecommendationDemoFeature\")\n", - "settings = MaterializationSettings(\"productRecommendationFeatureSetting\",\n", - " backfill_time=backfill_time,\n", - " sinks=[redisSink],\n", - " feature_names=[\"feature_user_age\", \"feature_user_gift_card_balance\"])\n", - "\n", - "client.materialize_features(settings)\n", - "client.wait_job_to_finish(timeout_sec=500)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "We can then get the features from the online store (Redis):\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Fetching feature value for online inference\n", - "\n", - "For features that are already materialized by the previous step, their latest value can be queried via the client's\n", - "`get_online_features` or `multi_get_online_features` API." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.get_online_features('productRecommendationDemoFeature', '2', [\n", - " 'feature_user_age', 'feature_user_gift_card_balance'])" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.multi_get_online_features('productRecommendationDemoFeature', ['1', '2'], [\n", - " 'feature_user_age', 'feature_user_gift_card_balance'])\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "### Registering and Fetching features\n", - "\n", - "We can also register the features with an Apache Atlas compatible service, such as Azure Purview, and share the registered features across teams:" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "client.register_features()\n", - "client.list_registered_features(project_name=\"feathr_getting_started\")" - ] - } - ], - "metadata": { - "interpreter": { - "hash": "b3c5d8fd79e029a19bf620c04a250a0cafa2291ba3ed87972a3e2a099b099985" - }, - "kernelspec": { - "display_name": "Python 3.9.12 ('product_env': venv)", - "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.9.12" - } - }, - "nbformat": 4, - "nbformat_minor": 2 + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Feathr Feature Store on Azure Demo Notebook\n", + "\n", + "This notebook illustrates the use of Feature Store to create a model that predicts product ratings. It includes these steps:\n", + "\n", + "\n", + "This tutorial demonstrates the key capabilities of Feathr, including:\n", + "\n", + "1. Install and set up Feathr with Azure\n", + "2. Create shareable features with Feathr feature definition configs.\n", + "3. Create a training dataset via point-in-time feature join.\n", + "4. Compute and write features.\n", + "5. Train a model using these features to predict fares.\n", + "6. Materialize feature value to online store.\n", + "7. Fetch feature value in real-time from online store for online scoring.\n", + "\n", + "In this tutorial, we use Feathr Feature Store to create a model that predict users rating for a product. The feature flow is as below:\n", + "![Feature Flow](https://github.com/linkedin/feathr/blob/main/docs/images/product_recommendation_overview.png?raw=true)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Use Quick Start Template to Provision Azure Resources\n", + "\n", + "Feathr has native cloud integration. To use Feathr on Azure, you only need three steps:\n", + "\n", + "- Get the `Principal ID` of your account by running `az ad signed-in-user show --query id -o tsv` in the link below (Select \"Bash\" if asked), and write down that value (something like `b65ef2e0-42b8-44a7-9b55-abbccddeefff`). Think this ID as something representing you when accessing Azure, and it will be used to grant permissions in the next step in the UI.\n", + "\n", + "[Launch Cloud Shell](https://shell.azure.com/bash)\n", + "\n", + "- Click the button below to deploy a minimal set of Feathr resources for demo purpose. You will need to fill in the `Principal ID` and `Resource Prefix`. You will need \"Owner\" permission of the selected subscription.\n", + "\n", + "[![Deploy to Azure](https://aka.ms/deploytoazurebutton)](https://portal.azure.com/#create/Microsoft.Template/uri/https%3A%2F%2Fraw.githubusercontent.com%2Flinkedin%2Ffeathr%2Fmain%2Fdocs%2Fhow-to-guides%2Fazure_resource_provision.json)\n", + "\n", + "- Run the cells below.\n", + "\n", + "And the architecture is as below. In the above template, we are using Synapse as Spark provider, use Azure Data Lake Gen2 as offline store, and use Redis as online store, Azure Purview (Apache Atlas compatible) as feature reigstry. \n", + "\n", + "\n", + "![Architecture](https://github.com/linkedin/feathr/blob/main/docs/images/architecture.png?raw=true)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Install Feathr \n", + "\n", + "Install Feathr using pip:\n", + "\n", + "`pip install -U feathr pandavro scikit-learn`" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Configure the required environment with Feathr Quick Start Template\n", + "\n", + "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. Run the code below to install Feathr, login to Azure to get the required credentials to access more cloud resources." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "**REQUIRED STEP: Fill in the resource prefix when provisioning the resources**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "resource_prefix = \"feathr_resource_prefix\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! pip install feathr azure-cli pandavro scikit-learn" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Login to Azure with a device code (You will see instructions in the output):" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! az login --use-device-code" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import glob\n", + "import os\n", + "import tempfile\n", + "from datetime import datetime, timedelta\n", + "from math import sqrt\n", + "\n", + "import pandas as pd\n", + "import pandavro as pdx\n", + "from feathr import FeathrClient\n", + "from feathr import BOOLEAN, FLOAT, INT32, ValueType\n", + "from feathr import Feature, DerivedFeature, FeatureAnchor\n", + "from feathr import BackfillTime, MaterializationSettings\n", + "from feathr import FeatureQuery, ObservationSettings\n", + "from feathr import RedisSink\n", + "from feathr import INPUT_CONTEXT, HdfsSource\n", + "from feathr import WindowAggTransformation\n", + "from feathr import TypedKey\n", + "from sklearn.metrics import mean_squared_error\n", + "from sklearn.model_selection import train_test_split\n", + "from azure.identity import DefaultAzureCredential\n", + "from azure.keyvault.secrets import SecretClient\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Get all the required credentials from Azure KeyVault" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Get all the required credentials from Azure Key Vault\n", + "key_vault_name=resource_prefix+\"kv\"\n", + "synapse_workspace_url=resource_prefix+\"syws\"\n", + "adls_account=resource_prefix+\"dls\"\n", + "adls_fs_name=resource_prefix+\"fs\"\n", + "purview_name=resource_prefix+\"purview\"\n", + "key_vault_uri = f\"https://{key_vault_name}.vault.azure.net\"\n", + "credential = DefaultAzureCredential(exclude_interactive_browser_credential=False)\n", + "client = SecretClient(vault_url=key_vault_uri, credential=credential)\n", + "secretName = \"FEATHR-ONLINE-STORE-CONN\"\n", + "retrieved_secret = client.get_secret(secretName).value\n", + "\n", + "# Get redis credentials; This is to parse Redis connection string.\n", + "redis_port=retrieved_secret.split(',')[0].split(\":\")[1]\n", + "redis_host=retrieved_secret.split(',')[0].split(\":\")[0]\n", + "redis_password=retrieved_secret.split(',')[1].split(\"password=\",1)[1]\n", + "redis_ssl=retrieved_secret.split(',')[2].split(\"ssl=\",1)[1]\n", + "\n", + "# Set the resource link\n", + "os.environ['spark_config__azure_synapse__dev_url'] = f'https://{synapse_workspace_url}.dev.azuresynapse.net'\n", + "os.environ['spark_config__azure_synapse__pool_name'] = 'spark31'\n", + "os.environ['spark_config__azure_synapse__workspace_dir'] = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_project'\n", + "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", + "os.environ['online_store__redis__host'] = redis_host\n", + "os.environ['online_store__redis__port'] = redis_port\n", + "os.environ['online_store__redis__ssl_enabled'] = redis_ssl\n", + "os.environ['REDIS_PASSWORD']=redis_password\n", + "os.environ['feature_registry__purview__purview_name'] = f'{purview_name}'\n", + "feathr_output_path = f'abfss://{adls_fs_name}@{adls_account}.dfs.core.windows.net/feathr_output'" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Prerequisite: Configure the required environment (Don't need to update if using the above Quick Start Template)\n", + "\n", + "In the first step (Provision cloud resources), you should have provisioned all the required cloud resources. If you use Feathr CLI to create a workspace, you should have a folder with a file called `feathr_config.yaml` in it with all the required configurations. Otherwise, update the configuration below.\n", + "\n", + "The code below will write this configuration string to a temporary location and load it to Feathr. Please still refer to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It should also have more explanations on the meaning of each variable." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import tempfile\n", + "yaml_config = \"\"\"\n", + "# Please refer to https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml for explanations on the meaning of each field.\n", + "api_version: 1\n", + "project_config:\n", + " project_name: 'feathr_getting_started'\n", + " required_environment_variables:\n", + " - 'REDIS_PASSWORD'\n", + " - 'AZURE_CLIENT_ID'\n", + " - 'AZURE_TENANT_ID'\n", + " - 'AZURE_CLIENT_SECRET'\n", + "offline_store:\n", + " adls:\n", + " adls_enabled: true\n", + " wasb:\n", + " wasb_enabled: true\n", + " s3:\n", + " s3_enabled: false\n", + " s3_endpoint: 's3.amazonaws.com'\n", + " jdbc:\n", + " jdbc_enabled: false\n", + " jdbc_database: 'feathrtestdb'\n", + " jdbc_table: 'feathrtesttable'\n", + " snowflake:\n", + " url: \"dqllago-ol19457.snowflakecomputing.com\"\n", + " user: \"feathrintegration\"\n", + " role: \"ACCOUNTADMIN\"\n", + "spark_config:\n", + " spark_cluster: 'azure_synapse'\n", + " spark_result_output_parts: '1'\n", + " azure_synapse:\n", + " dev_url: 'https://feathrazuretest3synapse.dev.azuresynapse.net'\n", + " pool_name: 'spark3'\n", + " workspace_dir: 'abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/feathr_getting_started'\n", + " executor_size: 'Small'\n", + " executor_num: 4\n", + " feathr_runtime_location: wasbs://public@azurefeathrstorage.blob.core.windows.net/feathr-assembly-LATEST.jar\n", + " databricks:\n", + " workspace_instance_url: 'https://adb-2474129336842816.16.azuredatabricks.net'\n", + " config_template: {'run_name':'','new_cluster':{'spark_version':'9.1.x-scala2.12','node_type_id':'Standard_D3_v2','num_workers':2,'spark_conf':{}},'libraries':[{'jar':''}],'spark_jar_task':{'main_class_name':'','parameters':['']}}\n", + " work_dir: 'dbfs:/feathr_getting_started'\n", + " feathr_runtime_location: https://azurefeathrstorage.blob.core.windows.net/public/feathr-assembly-LATEST.jar\n", + "online_store:\n", + " redis:\n", + " host: 'feathrazuretest3redis.redis.cache.windows.net'\n", + " port: 6380\n", + " ssl_enabled: True\n", + "feature_registry:\n", + " purview:\n", + " type_system_initialization: true\n", + " purview_name: 'feathrazuretest3-purview1'\n", + " delimiter: '__'\n", + "\"\"\"\n", + "tmp = tempfile.NamedTemporaryFile(mode='w', delete=False)\n", + "with open(tmp.name, \"w\") as text_file:\n", + " text_file.write(yaml_config)\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Setup necessary environment variables (Skip if using the above Quick Start Template)\n", + "\n", + "You should setup the environment variables in order to run this sample. More environment variables can be set by referring to [feathr_config.yaml](https://github.com/linkedin/feathr/blob/main/feathr_project/feathrcli/data/feathr_user_workspace/feathr_config.yaml) and use that as the source of truth. It also has more explanations on the meaning of each variable.\n", + "\n", + "To run this notebook, for Azure users, you need AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET and REDIS_PASSWORD.\n", + "To run this notebook, for Databricks useres, you need DATABRICKS_WORKSPACE_TOKEN_VALUE and REDIS_PASSWORD." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Initialize Feathr Client" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client = FeathrClient(config_path=tmp.name)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## View the data\n", + "\n", + "In this tutorial, we use Feathr Feature Store to create a model that predicts users product rating. To make it simple, let's just predict users' rating for one product. (We will expand the exmaple to predict ratings for arbitrary product in future tutorials.)\n", + "\n", + "We have 3 datasets to work with: one observation dataset(a.k.a. label dataset) and two raw datasets to generate features." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Observation dataset(a.k.a. label dataset)\n", + "# Observation dataset usually comes with a event_timestamp to denote when the observation happened.\n", + "# The label here is product_rating. Our model objective is to predict a user's rating for this product.\n", + "import pandas as pd\n", + "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/product_recommendation_sample/user_observation_mock_data.csv\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# User profile dataset\n", + "# Used to generate user features\n", + "import pandas as pd\n", + "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/product_recommendation_sample/user_profile_mock_data.csv\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# User purchase history dataset.\n", + "# Used to generate user features. This is activity type data, so we need to use aggregation to genearte features.\n", + "import pandas as pd\n", + "pd.read_csv(\"https://azurefeathrstorage.blob.core.windows.net/public/sample_data/product_recommendation_sample/user_purchase_history_mock_data.csv\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Defining Features with Feathr\n", + "\n", + "In Feathr, a feature is viewed as a function, mapping from entity id or key, and timestamp to a feature value. For more details on feature definition, please refer to the [Feathr Feature Definition Guide](https://github.com/linkedin/feathr/blob/main/docs/concepts/feature-definition.md)\n", + "\n", + "\n", + "1. The typed key (a.k.a. entity id) identifies the subject of feature, e.g. a user id, 123.\n", + "2. The feature name is the aspect of the entity that the feature is indicating, e.g. the age of the user.\n", + "3. The feature value is the actual value of that aspect at a particular time, e.g. the value is 30 at year 2022.\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Note that, in some cases, such as features defined on top of request data, may have no entity key or timestamp.\n", + "It is merely a function/transformation executing against request data at runtime.\n", + "For example, the day of week of the request, which is calculated by converting the request UNIX timestamp.\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Define Sources Section with UDFs\n", + "A feature source is needed for anchored features that describes the raw data in which the feature values are computed from. See the python documentation to get the details on each input column.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from pyspark.sql import SparkSession, DataFrame\n", + "def feathr_udf_preprocessing(df: DataFrame) -> DataFrame:\n", + " from pyspark.sql.functions import col\n", + " df = df.withColumn(\"tax_rate_decimal\", col(\"tax_rate\")/100)\n", + " df.show(10)\n", + " return df\n", + "\n", + "batch_source = HdfsSource(name=\"userProfileData\",\n", + " path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/product_recommendation_sample/user_profile_mock_data.csv\",\n", + " preprocessing=feathr_udf_preprocessing)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Define Anchors and Features\n", + "A feature is called an anchored feature when the feature is directly extracted from the source data, rather than computed on top of other features. The latter case is called derived feature." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "user_id = TypedKey(key_column=\"user_id\",\n", + " key_column_type=ValueType.INT32,\n", + " description=\"user id\",\n", + " full_name=\"product_recommendation.user_id\")\n", + "\n", + "feature_user_age = Feature(name=\"feature_user_age\",\n", + " key=user_id,\n", + " feature_type=INT32, transform=\"age\")\n", + "feature_user_tax_rate = Feature(name=\"feature_user_tax_rate\",\n", + " key=user_id,\n", + " feature_type=FLOAT,\n", + " transform=\"tax_rate_decimal\")\n", + "feature_user_gift_card_balance = Feature(name=\"feature_user_gift_card_balance\",\n", + " key=user_id,\n", + " feature_type=FLOAT,\n", + " transform=\"gift_card_balance\")\n", + "feature_user_has_valid_credit_card = Feature(name=\"feature_user_has_valid_credit_card\",\n", + " key=user_id,\n", + " feature_type=BOOLEAN,\n", + " transform=\"number_of_credit_cards > 0\")\n", + " \n", + "features = [\n", + " feature_user_age,\n", + " feature_user_tax_rate,\n", + " feature_user_gift_card_balance,\n", + " feature_user_has_valid_credit_card\n", + "]\n", + "\n", + "request_anchor = FeatureAnchor(name=\"anchored_features\",\n", + " source=batch_source,\n", + " features=features)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Window aggregation features\n", + "\n", + "For window aggregation features, see the supported fields below:\n", + "\n", + "Note that the `agg_func` should be any of these:\n", + "\n", + "| Aggregation Type | Input Type | Description |\n", + "| --- | --- | --- |\n", + "|SUM, COUNT, MAX, MIN, AVG\t|Numeric|Applies the the numerical operation on the numeric inputs. |\n", + "|MAX_POOLING, MIN_POOLING, AVG_POOLING\t| Numeric Vector | Applies the max/min/avg operation on a per entry bassis for a given a collection of numbers.|\n", + "|LATEST| Any |Returns the latest not-null values from within the defined time window |\n", + "\n", + "\n", + "After you have defined features and sources, bring them together to build an anchor:\n", + "\n", + "\n", + "Note that if the data source is from the observation data, the `source` section should be `INPUT_CONTEXT` to indicate the source of those defined anchors." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "purchase_history_data = HdfsSource(name=\"purchase_history_data\",\n", + " path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/product_recommendation_sample/user_purchase_history_mock_data.csv\",\n", + " event_timestamp_column=\"purchase_date\",\n", + " timestamp_format=\"yyyy-MM-dd\")\n", + " \n", + "agg_features = [Feature(name=\"feature_user_totla_purchase_in_90days\",\n", + " key=user_id,\n", + " feature_type=FLOAT,\n", + " transform=WindowAggTransformation(agg_expr=\"cast_float(purchase_amount)\",\n", + " agg_func=\"AVG\",\n", + " window=\"90d\"))\n", + " ]\n", + "\n", + "agg_anchor = FeatureAnchor(name=\"aggregationFeatures\",\n", + " source=purchase_history_data,\n", + " features=agg_features)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Derived Features Section\n", + "Derived features are the features that are computed from other features. They could be computed from anchored features, or other derived features." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "feature_user_purchasing_power = DerivedFeature(name=\"feature_user_purchasing_power\",\n", + " key=user_id,\n", + " feature_type=FLOAT,\n", + " input_features=[\n", + " feature_user_gift_card_balance, feature_user_has_valid_credit_card],\n", + " transform=\"feature_user_gift_card_balance + if_else(feature_user_has_valid_credit_card, 100, 0)\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "And then we need to build those features so that it can be consumed later. Note that we have to build both the \"anchor\" and the \"derived\" features (which is not anchored to a source)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.build_features(anchor_list=[agg_anchor, request_anchor], derived_feature_list=[\n", + " feature_user_purchasing_power])" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Create training data using point-in-time correct feature join\n", + "\n", + "A training dataset usually contains entity id columns, multiple feature columns, event timestamp column and label/target column. \n", + "\n", + "To create a training dataset using Feathr, one needs to provide a feature join configuration file to specify\n", + "what features and how these features should be joined to the observation data. \n", + "\n", + "To learn more on this topic, please refer to [Point-in-time Correctness](https://github.com/linkedin/feathr/blob/main/docs/concepts/point-in-time-join.md)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "if client.spark_runtime == 'databricks':\n", + " output_path = 'dbfs:/feathrazure_test.avro'\n", + "else:\n", + " output_path = feathr_output_path\n", + "\n", + "\n", + "feature_query = FeatureQuery(\n", + " feature_list=[\"feature_user_age\", \n", + " \"feature_user_tax_rate\", \n", + " \"feature_user_gift_card_balance\", \n", + " \"feature_user_has_valid_credit_card\", \n", + " \"feature_user_totla_purchase_in_90days\",\n", + " \"feature_user_purchasing_power\"], key=user_id)\n", + "settings = ObservationSettings(\n", + " observation_path=\"wasbs://public@azurefeathrstorage.blob.core.windows.net/sample_data/product_recommendation_sample/user_observation_mock_data.csv\",\n", + " event_timestamp_column=\"event_timestamp\",\n", + " timestamp_format=\"yyyy-MM-dd\")\n", + "client.get_offline_features(observation_settings=settings,\n", + " feature_query=feature_query,\n", + " output_path=output_path)\n", + "client.wait_job_to_finish(timeout_sec=500)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Download the result and show the result\n", + "\n", + "Let's use the helper function `get_result_df` to download the result and view it:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "def get_result_df(client: FeathrClient) -> pd.DataFrame:\n", + " \"\"\"Download the job result dataset from cloud as a Pandas dataframe.\"\"\"\n", + " res_url = client.get_job_result_uri(block=True, timeout_sec=600)\n", + " tmp_dir = tempfile.TemporaryDirectory()\n", + " client.feathr_spark_laucher.download_result(result_path=res_url, local_folder=tmp_dir.name)\n", + " dataframe_list = []\n", + " # assuming the result are in avro format\n", + " for file in glob.glob(os.path.join(tmp_dir.name, '*.avro')):\n", + " dataframe_list.append(pdx.read_avro(file))\n", + " vertical_concat_df = pd.concat(dataframe_list, axis=0)\n", + " tmp_dir.cleanup()\n", + " return vertical_concat_df\n", + "\n", + "df_res = get_result_df(client)\n", + "\n", + "df_res" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Train a machine learning model\n", + "After getting all the features, let's train a machine learning model with the converted feature by Feathr:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# drop non-feature columns\n", + "from sklearn.ensemble import GradientBoostingRegressor\n", + "final_df = df_res\n", + "final_df.drop([\"event_timestamp\"], axis=1, inplace=True, errors='ignore')\n", + "final_df.fillna(0, inplace=True)\n", + "final_df['product_rating'] = final_df['product_rating'].astype(\"float64\")\n", + "\n", + "train_x, test_x, train_y, test_y = train_test_split(final_df.drop([\"product_rating\"], axis=1),\n", + " final_df[\"product_rating\"],\n", + " test_size=0.2,\n", + " random_state=42)\n", + "model = GradientBoostingRegressor()\n", + "model.fit(train_x, train_y)\n", + "\n", + "y_predict = model.predict(test_x)\n", + "\n", + "y_actual = test_y.values.flatten().tolist()\n", + "rmse = sqrt(mean_squared_error(y_actual, y_predict))\n", + "\n", + "sum_actuals = sum_errors = 0\n", + "\n", + "for actual_val, predict_val in zip(y_actual, y_predict):\n", + " abs_error = actual_val - predict_val\n", + " if abs_error < 0:\n", + " abs_error = abs_error * -1\n", + "\n", + " sum_errors = sum_errors + abs_error\n", + " sum_actuals = sum_actuals + actual_val\n", + "\n", + "mean_abs_percent_error = sum_errors / sum_actuals\n", + "print(\"Model MAPE:\")\n", + "print(mean_abs_percent_error)\n", + "print()\n", + "print(\"Model Accuracy:\")\n", + "print(1 - mean_abs_percent_error)\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Materialize feature value into offline/online storage\n", + "\n", + "While Feathr can compute the feature value from the feature definition on-the-fly at request time, it can also pre-compute\n", + "and materialize the feature value to offline and/or online storage. \n", + "\n", + "We can push the generated features to the online store like below:\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "backfill_time = BackfillTime(start=datetime(\n", + " 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1))\n", + "redisSink = RedisSink(table_name=\"productRecommendationDemoFeature\")\n", + "settings = MaterializationSettings(\"productRecommendationFeatureSetting\",\n", + " backfill_time=backfill_time,\n", + " sinks=[redisSink],\n", + " feature_names=[\"feature_user_age\", \"feature_user_gift_card_balance\"])\n", + "\n", + "client.materialize_features(settings)\n", + "client.wait_job_to_finish(timeout_sec=500)\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "We can then get the features from the online store (Redis):\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Fetching feature value for online inference\n", + "\n", + "For features that are already materialized by the previous step, their latest value can be queried via the client's\n", + "`get_online_features` or `multi_get_online_features` API." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.get_online_features('productRecommendationDemoFeature', '2', [\n", + " 'feature_user_age', 'feature_user_gift_card_balance'])" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.multi_get_online_features('productRecommendationDemoFeature', ['1', '2'], [\n", + " 'feature_user_age', 'feature_user_gift_card_balance'])\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Registering and Fetching features\n", + "\n", + "We can also register the features with an Apache Atlas compatible service, such as Azure Purview, and share the registered features across teams:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "client.register_features()\n", + "client.list_registered_features(project_name=\"feathr_getting_started\")" + ] + } + ], + "metadata": { + "interpreter": { + "hash": "b3c5d8fd79e029a19bf620c04a250a0cafa2291ba3ed87972a3e2a099b099985" + }, + "kernelspec": { + "display_name": "Python 3.9.12 ('product_env': venv)", + "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.9.12" + } + }, + "nbformat": 4, + "nbformat_minor": 2 }