{ "cells": [ { "cell_type": "markdown", "metadata": { "toc": true }, "source": [ "

Table of Contents

\n", "
" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Feature Store with Aerospike\n", "This notebook is the first in the series of notebooks that show how Aerospike can be used as a feature store.\n", "\n", "This notebook requires the Aerospike Database and Spark running locally with Aerospike Spark Connector. To create a Docker container that satisfies the requirements and holds a copy of Aerospike notebooks, visit the [Aerospike Notebooks Repo](https://github.com/aerospike-examples/interactive-notebooks)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Introduction\n", "This notebook demonstrates how Aerospike can be used as a Feature Store for Machine Learning applications on Spark using the Aerospike Spark Connector. It is part of the Feature Store series of notebooks, and focuses on data model and Feature Engineering aspects concerning a Feature Store. The subsequent notebook(s) will describe the Model Traiing and Model Serving aspects of an ML application that deal with a Feature Store." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "![Reference Architecture](resources/fs-arch.jpg)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "This notebook is organized in two parts: \n", "1. The first part explains the key objects, data model, and operations of a Feature Store, and provides a simple implementation on top of the Aerospike Database. \n", "2. The second part shows with two examples how the model developed in the first part can be used for Feature Engineering and storing features in the Feature Store." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Prerequisites\n", "This tutorial assumes familiarity with the following topics:\n", "\n", "- [Aerospike Notebooks - Readme and Tips](../readme_tips.ipynb)\n", "- [Hello World](../python/hello_world.ipynb)\n", "- [Aerospike Connect for Spark Tutorial for Python](AerospikeSparkPython.ipynb)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Setup\n", "Set up Aerospike Server. Spark Server, and Spark Connector." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Ensure Database Is Running\n", "This notebook requires that Aerospike datbase is running." ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Aerospike database is running!\r\n" ] } ], "source": [ "!asd >& /dev/null\n", "!pgrep -x asd >/dev/null && echo \"Aerospike database is running!\" || echo \"**Aerospike database is not running!**\"" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Initialize Spark\n", "We will be using Spark functionality in this notebook." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "#### Initialize Paths and Env Variables" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "# directory where spark notebook requisites are installed\n", "SPARK_NB_DIR = '/opt/spark-nb'\n", "SPARK_DIR = 'spark-dir-link'\n", "SPARK_HOME = SPARK_NB_DIR + '/' + SPARK_DIR\n", "AEROSPIKE_JAR = 'aerospike-jar-link'\n", "AEROSPIKE_JAR_PATH = SPARK_NB_DIR + '/' + AEROSPIKE_JAR" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [], "source": [ "# IP Address or DNS name for one host in your Aerospike cluster\n", "AS_HOST =\"localhost\"\n", "# Name of one of your namespaces. Type 'show namespaces' at the aql prompt if you are not sure\n", "AS_NAMESPACE = \"test\" \n", "AS_PORT = 3000 # Usually 3000, but change here if not\n", "AS_CONNECTION_STRING = AS_HOST + \":\"+ str(AS_PORT)" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [], "source": [ "# Locate the Spark installation using the SPARK_HOME parameter.\n", "import findspark\n", "findspark.init(SPARK_HOME)" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "# Specify the Aerospike Spark Connector jar in the command used to interact with Aerospike.\n", "import os \n", "os.environ[\"PYSPARK_SUBMIT_ARGS\"] = '--jars ' + AEROSPIKE_JAR_PATH + ' pyspark-shell'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "#### Configure Spark Session\n", "Please visit [Configuring Aerospike Connect for Spark](https://docs.aerospike.com/docs/connect/processing/spark/configuration.html) for more information about the properties used on this page." ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "# imports\n", "import pyspark\n", "from pyspark.context import SparkContext\n", "from pyspark.sql.context import SQLContext\n", "from pyspark.sql.session import SparkSession\n", "from pyspark.sql.types import StringType, StructField, StructType, ArrayType, IntegerType, MapType, LongType, DoubleType" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "sc = SparkContext.getOrCreate()\n", "conf=sc._conf.setAll([(\"aerospike.namespace\",AS_NAMESPACE),(\"aerospike.seedhost\",AS_CONNECTION_STRING)])\n", "sc.stop()\n", "sc = pyspark.SparkContext(conf=conf)\n", "spark = SparkSession(sc)\n", "sqlContext = SQLContext(sc)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Access Shell Commands\n", "You may execute shell commands including Aerospike tools like [aql](https://docs.aerospike.com/docs/tools/aql/index.html) and [asadm](https://docs.aerospike.com/docs/tools/asadm/index.html) in the terminal tab throughout this tutorial. Open a terminal tab by selecting File->Open from the notebook menu, and then New->Terminal." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Online and Offline Feature Stores\n", "Aerospike can be used as a Feature Store in cases where broader capabilities of a commercial Feature Store are not warranted. Instead, when it is critical to have:\n", "- Sub-millisecond online access to features\n", "- Scaling to a very large number of entities and features\n", "- Convenient and reliable distributed deployment \n", "- Small server footprint for a low cost of operation" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "An Online Feature Store provides access to features in real time, and as such requires speed and, for many deployments, scalability. The Online Store only stores the latest version of feature values.\n", "\n", "Aerospike is purpose built for the performance and scale requirements of an Online Store. \n", "\n", "An Offline Store requires historic data to be maintained for “time travel” use cases, that is, to create datasets at a particular point in time for training and other needs. A time series of feature values can be maintained in Aerospike as a map (timestamp->value) for each feature. Aerospike's native APIs allow efficient implementation of an Offline Store.\n", "\n", "## Offline-Online Consistency\n", "\n", "An essential requirement for a Feature Store is to maintain Online-Offline consistency to ensure that the models that were trained with the Offline (historic) data remain valid in the production (Online) environment. Typically this means convenient, efficient, and timely sync of data from Offline to Online Store. Sometimes a reverse sync from Online to Offline Store may be requred if only the Online Store may be updated. \n", "\n", "Synchronization in either direction between Online and Offline Stores can be achieved through various streaming connectors that Aerospike provides (such as for Kafka and Pulsar).\n", "\n", "What is also necessary is to be able to access a specific version “as of time” (a special case being the latest version) for various use cases using the appropriate client library or connector. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Our focus in this Feature Store series will be primarily on the Online Store." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Design Considerations\n", "The key design criteria for a Feature Store with Aerospike are access and storage efficiency. (The other key criterion, Offline - Online synchronization, will not be discussed further here. For our purpose here, we assume that can be handled with appropriate change notification and update setup between Offline and Online Stores.)\n", "\n", "We will assume the following object model which is typical in a Feature Store:\n", "- Precomputed `Feature Values` are stored for `Entities` such as users, credit card transactions, and sensors.\n", "- `Features` are organized in `Feature Groups`. A `Feature Group` signifies the set of features that share common data sources, pipeline, and are created and updated together.\n", "- A `Dataset` is a snapshot of specified `Features` across a set of `Entity` instances. The snapshot is stored outside the Feature Store, whereas the Feature Store holds the metadata such as the snapshot query and location.\n", "- The Online Store keeps only the latest version of Feature Values, whereas the Offline store maintains a time series of Feature Values. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Online Store\n", "There are multiple ways to store the latest Feature Values, each with its trade-offs. Here are two possible options:\n", "1. One `Entity` instance record holds all `Feature Values` across multiple `Feature Groups`.\n", " - Stored in a set named after the `Entity` type (e.g., \"user-features\").\n", " - Read and write for a `Feature Group`'s `Features` entails listing the `Feature` bins. Accessing a `Feature` vector at Model Serving time entails enumerating the model's `Feature` bins across `Feature Groups`. To avoid `Feature` name collisions acroos `Feature Groups`, a bin name needs to include both the `Feature Group` and `Feature` name such as `Feature-Group:Feature`. \n", " - Pros: \n", " - Fast and convenient read and write access to all features of an entity.\n", " - Cons:\n", " - A single last-update-time must be shared among all `Feature Groups` without additional mechanisms.\n", "\n", "2. Multiple `Entity` instance records, one for each `Feature Group`. \n", " - Each `Feature Group`'s values are stored in its own set named in form `Entity:FeatureGroup-features` such as \"user:fg1-features\".\n", " - Read and write for a `Feature Group`'s `Features` entail listing the `Feature` bins. Accessing a `Feature` vector at Model Serving time entails multiple requests over `Feature Groups`. Bin names are same as `Feature` names since they are unique within a `Feature Group`. \n", " - Pros:\n", " - Provides a better organization.\n", " - Allows for a large number of `Features` if they may exceed the record size limit.\n", " - Allows a separate timestamp for each `Feature Group` since each is updated separately.\n", " - Cons:\n", " - Multiple requests must be issued to access `Features` across `Feature Groups`." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Offline Store\n", "The design requires multiple versions of Feature Values to be stored, each with its timestamp. One way of modeling such a time series of feature values in Aerospike is to store each feature as a map `timestamp->value`.\n", "\n", "#### Modeling for Size\n", "With historic versions stored in it, the Offline Store has to account for the data size problem: both in terms of individual record size as well as overall database size.\n", "- Record size\n", " - Limit the number of versions in each record by designating a record by its time duration. So Feature Values for an Entity instance will have multiple records by hour, day, week, month, or year, etc, depending on the frequency of updates. The record will have a compound key of the form `entity-key:duration-id` like \"123:202101\", for example, for a record holding feature values during January 2021 for a user with id 123.\n", " - As discussed above, the record size can be kept small by storing each Feature Group's Feature Values in its own set.\n", "- Database size\n", " - To limit the size of the Offline Store, older data can be archived in a batch process. An indexed bin in a record holds the record's duration specifier allowing efficient access to old records for archival. For example, weekly records wiil have an indexed bin with a year and week values concatenated like \"202140\"." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Improving Performance and Scalability\n", "Before we dive into the implementation, here are some relevant performance and scalability aspects in use of the Spark Connector.\n", "### Random access with primary key \n", "Random access using the primary key must use the \\__key bin for direct access through the primary index. Many times the unique user key is duplicated for convenience in another bin, but the equality predicate using another bin will result in a scan. This is shown below in single object `load` methods.\n", "### Efficient scan with set indexes\n", "A set index should be enabled on a set for faster scans over the set. Data exploration as well as dataset creation operations will benefit by not having to scan the entire namespace instead of a fraction of records in a set, resulting in significant speedup as the namespace can be very large as compared to, say, the metadata sets. In the following implementation, we enable set indexes on all sets.\n", "### Aerospike Expression pushdown\n", "To minimize the amount of data retrieved from the database, query predicates must be \"pushed down\", or in other words, processed in the database and not on Spark. We will illustrate how expression pushdown is implemented with examples in the Model Training notebook.\n", "### Performance tuning parameters \n", "Several tuning parameters such as `partition factor` and `compression` are available for optimizing perfomrance with the Spark Connector. The implementation here does not make use of them, but you are encouraged to explire them elsewhere.\n", "### Use of secondary indexes\n", "While Aerospike provides secondary indexes, the Spark Connector currently does not leverage them. This performance enhancement is planned in the future. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Defining a Simple Feature Store\n", "To make it concrete, let us define and implement a simple Feature Store.\n", "\n", "## Main Objects\n", "The main objects in a Feature Store as discussed above are Feature Group, Feature, Entiry, and Dataset. These are explained below.\n", "### Features Group\n", "Machine Learning features are grouped by the source of data and processing pipeline. A Feature Group includes many features, and has the following attributes that are useful during feature exploration:\n", "- name: a unique name\n", "- description: human readable description\n", "- source: source of fact data \n", "- attrs: other metadata\n", "- tags: associated tags\n", "\n", "A feature group is stored in the set \"fg-metadata\".\n", "\n", "### Feature\n", "A Feature consists of the following attributes:\n", "- fid: the record's primary key consists of the string concatenation of feature group name and feature name in the form: fgname_fname\n", "- fgname: the name of the feature group that the feature belongs to\n", "- name: feature name that is unique within the feature group\n", "- type: feature type (integer, double, string, and possibly others)\n", "- description: human readable description\n", "- attrs: various performance stats and other metadata\n", "- tags: associated tags\n", "\n", "A feature is stored in the set \"feature-metadata\".\n", "\n", "### Entity\n", "Features are computed and stored for an Entity such as a user, credit card, or sensor. Feature values are stored per Entity instance. Features in multiple feature groups for an Entity type are combined in one Entity record. A feature values record for an Entity has these attributes:\n", "- id_col: the column containing the id of the entity instance that serves as the record key\n", "- feature specific bins: each feature is stored in a bin named after the feature in the format fgname_fname.\n", "- timestamp: update timestamp \n", "\n", "Entity records for an entity type are stored in an entity-type specific set \"entitytype-features\". For example, the set \"user-features\" holds features for user instances, and \"cc-features\" holds features for credit card instances. \n", " \n", "### Dataset\n", "A Dataset is a subset of features and entities selected to train an ML model. A Dataset object holds the selected features and entity instance definitions. The actual copy of entities is stored outside the feature store (for instance, in a file system). A dataset record has the following attributes.\n", "- name: name of the data set, serves as the primary key for the record\n", "- description: human readable description\n", "- features: a list of the dataset features\n", "- predicate: query predicate to enumerate the entity instances in the dataset\n", "- location: external location where the dataset is stored\n", "- attrs: other metadata\n", "- tags: associated tags\n", "\n", "Datasets are stored in the set \"dataset-metadata\".\n", "\n", "## Operations\n", "The following operations are implemented for the different use scenarios:\n", "- Feature Engineering\n", " - Feature Group\n", " - create or update (save)\n", " - load (get)\n", " - Feature\n", " - create or update\n", " - load\n", " - Entity\n", " - create or update using a dataframe\n", " - load \n", "- Model Training\n", " - Feature Group\n", " - load\n", " - query by various attributes\n", " - Feature\n", " - query\n", " - Entity\n", " - query\n", " - Dataset\n", " - create\n", " - load\n", " - query\n", "\n", "- Model Serving\n", " - Entity\n", " - get a specific feature vector for an entity instance" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Example Implementation\n", "The following code is a simple implementation of the above operations. These operations will be illustrated in this and follow-up notebooks." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Feature Group Implementation" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [], "source": [ "import copy\n", "\n", "class FeatureGroup:\n", " schema = StructType([StructField(\"name\", StringType(), False),\n", " StructField(\"description\", StringType(), True),\n", " StructField(\"source\", StringType(), True),\n", " StructField(\"attrs\", MapType(StringType(), StringType()), True),\n", " StructField(\"tags\", ArrayType(StringType()), True)])\n", " \n", " def __init__(self, name, description, source, attrs, tags):\n", " self.name = name\n", " self.description = description\n", " self.source = source\n", " self.attrs = attrs \n", " self.tags = tags \n", " return\n", "\n", " def __str__(self):\n", " return str(self.__class__) + \": \" + str(self.__dict__)\n", "\n", "\n", " def save(self):\n", " inputBuf = [(self.name, self.description, self.source, self.attrs, self.tags)]\n", " inputRDD = spark.sparkContext.parallelize(inputBuf) \n", " inputDF = spark.createDataFrame(inputRDD, FeatureGroup.schema)\n", " #Write the data frame to Aerospike, the name field is used as the key\n", " inputDF.write \\\n", " .mode('overwrite') \\\n", " .format(\"aerospike\") \\\n", " .option(\"aerospike.writeset\", \"fg-metadata\")\\\n", " .option(\"aerospike.updateByKey\", \"name\") \\\n", " .save()\n", " return \n", "\n", " def load(name):\n", " fg = None\n", " schema = copy.deepcopy(FeatureGroup.schema)\n", " schema.add(\"__key\", StringType(), False)\n", " fgdf = spark.read \\\n", " .format(\"aerospike\") \\\n", " .option(\"aerospike.set\", \"fg-metadata\") \\\n", " .schema(schema) \\\n", " .load().where(\"__key = \\\"\" + name + \"\\\"\") \n", " if fgdf.count() > 0:\n", " fgtuple = fgdf.collect()[0]\n", " fg = FeatureGroup(*fgtuple[:-1]) \n", " return fg\n", " \n", " def query(predicate): #returns a dataframe\n", " fg_df = spark.read \\\n", " .format(\"aerospike\") \\\n", " .schema(FeatureGroup.schema) \\\n", " .option(\"aerospike.set\", \"fg-metadata\") \\\n", " .load().where(predicate)\n", " return fg_df" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "ok\r\n" ] } ], "source": [ "# Enable set index on fg-metadata\n", "!asinfo -v \"set-config:context=namespace;id=test;set=fg-metadata;enable-index=true\"" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ ": {'name': 'fg_name1', 'description': 'fg_desc1', 'source': 'fg_source1', 'attrs': {'fg_attr2': 'two', 'fg_attr1': '1'}, 'tags': ['fg_tag1', 'fg_tag2']} \n", "\n", "Feature Groups with name ending with '_name1' and having attribute 'fg_tag1'='1':\n", "+--------+-----------+----------+--------------------+------------------+\n", "| name|description| source| attrs| tags|\n", "+--------+-----------+----------+--------------------+------------------+\n", "|fg_name1| fg_desc1|fg_source1|{fg_attr1 -> 1, f...|[fg_tag1, fg_tag2]|\n", "+--------+-----------+----------+--------------------+------------------+\n", "\n" ] } ], "source": [ "# test feature group \n", "# test save and load\n", "# save\n", "fg1 = FeatureGroup(\"fg_name1\", \"fg_desc1\", \"fg_source1\", \n", " {\"fg_attr1\":\"1\", \"fg_attr2\":\"two\"}, [\"fg_tag1\", \"fg_tag2\"])\n", "fg1.save()\n", "# load\n", "fg2 = FeatureGroup.load(\"fg_name1\")\n", "print(fg2, '\\n')\n", "# test query\n", "print(\"Feature Groups with name ending with '_name1' and having attribute 'fg_tag1'='1':\")\n", "df = FeatureGroup.query(\"name like '%_name1' and attrs.fg_attr1 == '1'\")\n", "df.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Feature Implementation" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "class Feature:\n", " schema = StructType([StructField(\"fid\", StringType(), False),\n", " StructField(\"fgname\", StringType(), False),\n", " StructField(\"name\", StringType(), False),\n", " StructField(\"type\", StringType(), False),\n", " StructField(\"description\", StringType(), True),\n", " StructField(\"attrs\", MapType(StringType(), StringType()), True),\n", " StructField(\"tags\", ArrayType(StringType()), True)])\n", "\n", " def __init__(self, fgname, name, ftype, description, attrs, tags):\n", " self.fid = fgname + '_' + name\n", " self.fgname = fgname\n", " self.name = name\n", " self.ftype = ftype\n", " self.description = description\n", " self.attrs = attrs \n", " self.tags = tags \n", " return\n", " \n", " def __str__(self):\n", " return str(self.__class__) + \": \" + str(self.__dict__)\n", "\n", " def save(self):\n", " inputBuf = [(self.fid, self.fgname, self.name, self.ftype, self.description, self.attrs, self.tags)]\n", " inputRDD = spark.sparkContext.parallelize(inputBuf) \n", " inputDF = spark.createDataFrame(inputRDD, Feature.schema)\n", " #Write the data frame to Aerospike, the fid field is used as the key\n", " inputDF.write \\\n", " .mode('overwrite') \\\n", " .format(\"aerospike\") \\\n", " .option(\"aerospike.writeset\", \"feature-metadata\")\\\n", " .option(\"aerospike.updateByKey\", \"fid\") \\\n", " .save()\n", " return \n", "\n", " def load(fgname, name):\n", " f = None\n", " schema = copy.deepcopy(Feature.schema)\n", " schema.add(\"__key\", StringType(), False)\n", " f_df = spark.read \\\n", " .format(\"aerospike\") \\\n", " .schema(schema) \\\n", " .option(\"aerospike.set\", \"feature-metadata\") \\\n", " .load().where(\"__key = \\\"\" + fgname+'_'+name + \"\\\"\") \n", " if f_df.count() > 0:\n", " f_tuple = f_df.collect()[0]\n", " f = Feature(*f_tuple[1:-1]) \n", " return f\n", " \n", " def query(predicate): #returns a dataframe\n", " f_df = spark.read \\\n", " .format(\"aerospike\") \\\n", " .schema(Feature.schema) \\\n", " .option(\"aerospike.set\", \"feature-metadata\") \\\n", " .load().where(predicate)\n", " return f_df" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "ok\r\n" ] } ], "source": [ "# Enable set index on feature-metadata\n", "!asinfo -v \"set-config:context=namespace;id=test;set=feature-metadata;enable-index=true\"" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ ": {'fid': 'fg_name1_f_name1', 'fgname': 'fg_name1', 'name': 'f_name1', 'ftype': 'integer', 'description': 'f_desc1', 'attrs': {'f_attr2': 'two', 'f_attr1': '1'}, 'tags': ['f_tag1', 'f_tag2']} \n", "\n", "Features having name starting with 'f_name' and tagged with 'f_tag1':\n", "+----------------+--------+-------+-------+-----------+--------------------+----------------+\n", "| fid| fgname| name| type|description| attrs| tags|\n", "+----------------+--------+-------+-------+-----------+--------------------+----------------+\n", "|fg_name1_f_name2|fg_name1|f_name2| double| f_desc2|{f_attr1 -> 1.0, ...|[f_tag1, f_tag3]|\n", "|fg_name1_f_name1|fg_name1|f_name1|integer| f_desc1|{f_attr1 -> 1, f_...|[f_tag1, f_tag2]|\n", "+----------------+--------+-------+-------+-----------+--------------------+----------------+\n", "\n" ] } ], "source": [ "# test Feature\n", "# test save and load\n", "# save\n", "feature1 = Feature(\"fg_name1\", \"f_name1\", \"integer\", \"f_desc1\", \n", " {\"f_attr1\":\"1\", \"f_attr2\":\"two\"}, [\"f_tag1\", \"f_tag2\"])\n", "feature1.save()\n", "# load\n", "f1 = Feature.load(\"fg_name1\", \"f_name1\")\n", "print(f1, '\\n')\n", "# test query\n", "feature2 = Feature(\"fg_name1\", \"f_name2\", \"double\", \"f_desc2\", \n", " {\"f_attr1\":\"1.0\", \"f_attr3\":\"three\"}, [\"f_tag1\", \"f_tag3\"])\n", "feature2.save()\n", "print(\"Features having name starting with 'f_name' and tagged with 'f_tag1':\")\n", "f_df = Feature.query(\"name like 'f_name%' and array_contains(tags, 'f_tag1')\")\n", "f_df.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Entity Implementation" ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [], "source": [ "class Entity:\n", " \n", " def __init__(self, etype, record, id_col):\n", " # record is an array of triples (name, type, value)\n", " self.etype = etype\n", " self.record = record\n", " self.id_col = id_col\n", " return\n", " \n", " def __str__(self):\n", " return str(self.__class__) + \": \" + str(self.__dict__)\n", " \n", " def get_schema(record): \n", " schema = StructType()\n", " for f in record:\n", " schema.add(f[0], f[1], True)\n", " return schema\n", " \n", " def get_id_type(schema, id_col): \n", " return schema[id_col].dataType.typeName()\n", "\n", " def save(self, schema):\n", " fvalues = [f[2] for f in self.record]\n", " inputBuf = [tuple(fvalues)]\n", " inputRDD = spark.sparkContext.parallelize(inputBuf) \n", " inputDF = spark.createDataFrame(inputRDD, schema)\n", " #Write the data frame to Aerospike, the id_col field is used as the key\n", " inputDF.write \\\n", " .mode('overwrite') \\\n", " .format(\"aerospike\") \\\n", " .option(\"aerospike.writeset\", self.etype+'-features')\\\n", " .option(\"aerospike.updateByKey\", self.id_col) \\\n", " .save()\n", " return \n", "\n", " def load(etype, eid, schema, id_col):\n", " ent = None\n", " ent_df = spark.read \\\n", " .format(\"aerospike\") \\\n", " .schema(schema) \\\n", " .option(\"aerospike.set\", etype+'-features') \\\n", " .option(\"aerospike.keyType\", \"string\") \\\n", " .load().where(id_col + \" = \\\"\" + eid + \"\\\"\") \n", " if ent_df.count() > 0:\n", " ent_tuple = ent_df.collect()[0]\n", " record = [(schema[i].name, schema[i].dataType.typeName(), fv) for i, fv in enumerate(ent_tuple)]\n", " ent = Entity(etype, record, id_col) \n", " return ent\n", " \n", " def saveDF(df, etype, id_col): # save a dataframe\n", " # df: dataframe consisting of feature values\n", " # etyoe: entity type (such as user or sensor)\n", " # id_col: column name that holds the primary key\n", " #Write the data frame to Aerospike, the column in id_col is used as the key\n", " df.write \\\n", " .mode('overwrite') \\\n", " .format(\"aerospike\") \\\n", " .option(\"aerospike.writeset\", etype+'-features')\\\n", " .option(\"aerospike.updateByKey\", id_col) \\\n", " .save()\n", " return \n", " \n", " def query(etype, predicate, schema, id_col): #returns a dataframe\n", " ent_df = spark.read \\\n", " .format(\"aerospike\") \\\n", " .schema(schema) \\\n", " .option(\"aerospike.set\", etype+'-features') \\\n", " .option(\"aerospike.keyType\", Entity.get_id_type(schema, id_col)) \\\n", " .load().where(predicate)\n", " return ent_df\n", " \n", " def get_feature_vector(etype, eid, feature_list): # elements in feature_list are in \"fg_name|name\" form\n", " # deferred to Model Serving notebook \n", " pass" ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "ok\r\n" ] } ], "source": [ "# Enable set index on cctxn-features\n", "!asinfo -v \"set-config:context=namespace;id=test;set=cctxn-features;enable-index=true\"" ] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ ": {'etype': 'entity_type1', 'record': [('eid', 'string', 'eid1'), ('fg1:f_name1', 'integer', 1), ('fg1:f_name2', 'double', 2.0), ('fg1:f_name3', 'string', 'three')], 'id_col': 'eid'} \n", "\n", "Instances of entity type entity_type1 with eid ending in 1:\n", "+----+-----------+-----------+-----------+\n", "| eid|fg1:f_name1|fg1:f_name2|fg1:f_name3|\n", "+----+-----------+-----------+-----------+\n", "|eid1| 1| 2.0| three|\n", "+----+-----------+-----------+-----------+\n", "\n", "Instances of entity type entity_type2 with eid in ['eid2']:\n", "+----+-----------+-----------+-----------+\n", "| eid|fg1:f_name1|fg1:f_name2|fg1:f_name3|\n", "+----+-----------+-----------+-----------+\n", "|eid2| 10| 20.0| thirty|\n", "+----+-----------+-----------+-----------+\n", "\n" ] } ], "source": [ "# test Entity\n", "# test save and load\n", "# save\n", "features1 = [('fg1:f_name1', IntegerType(), 1), ('fg1:f_name2', DoubleType(), 2.0), ('fg1:f_name3', StringType(), 'three')]\n", "record1 = [('eid', StringType(), 'eid1')] + features1\n", "ent1 = Entity('entity_type1', record1, 'eid')\n", "schema = Entity.get_schema(record1)\n", "ent1.save(schema);\n", "# load\n", "e1 = Entity.load('entity_type1', 'eid1', schema, 'eid')\n", "print(e1, '\\n')\n", "# test query\n", "features2 = [('fg1:f_name1', IntegerType(), 10), ('fg1:f_name2', DoubleType(), 20.0), ('fg1:f_name3', StringType(), 'thirty')]\n", "record2 = [('eid', StringType(), 'eid2')] + features2\n", "ent2 = Entity('entity_type2', record2, 'eid')\n", "ent2.save(schema);\n", "# query 1\n", "print(\"Instances of entity type entity_type1 with eid ending in 1:\")\n", "instances = Entity.query('entity_type1', 'eid like \"%1\"', schema, 'eid')\n", "instances.show()\n", "# query 2\n", "print(\"Instances of entity type entity_type2 with eid in ['eid2']:\")\n", "instances = Entity.query('entity_type2', 'eid in (\"eid2\")', schema, 'eid')\n", "instances.show()" ] }, { "cell_type": "code", "execution_count": 17, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+-------------+-------------+\n", "| ID|fg2_feature_1|fg2_feature_2|\n", "+-----+-------------+-------------+\n", "|eid_1| 1| 12.4|\n", "|eid_2| 2| 30.1|\n", "|eid_3| 3| 100.01|\n", "+-----+-------------+-------------+\n", "\n" ] } ], "source": [ "# test save dataframe\n", "\n", "from pyspark.sql import SparkSession, Row\n", "\n", "data = [{\"ID\": 'eid_1', \"fg2_feature_1\": 1, \"fg2_feature_2\": 12.40},\n", " {\"ID\": 'eid_2', \"fg2_feature_1\": 2, \"fg2_feature_2\": 30.10},\n", " {\"ID\": 'eid_3', \"fg2_feature_1\": 3, \"fg2_feature_2\": 100.01}\n", " ]\n", "\n", "# create and save a dataframe\n", "df = spark.createDataFrame([Row(**i) for i in data])\n", "df.show()\n", "Entity.saveDF(df, \"etype_1\", \"ID\")" ] }, { "cell_type": "code", "execution_count": 18, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+-------------+-------------+\n", "| ID|fg2_feature_1|fg2_feature_2|\n", "+-----+-------------+-------------+\n", "|eid_1| 1| 12.4|\n", "|eid_3| 3| 100.01|\n", "+-----+-------------+-------------+\n", "\n" ] } ], "source": [ "# test query operation that returns a dataframe\n", "# need to define a schema first\n", "schema = StructType([\n", " StructField('ID', StringType(), False),\n", " StructField('fg2_feature_1', IntegerType(), True),\n", " StructField('fg2_feature_2', DoubleType(), True)])\n", "queryDF = Entity.query('etype_1', 'ID in (\"eid_1\", \"eid_3\")', schema, 'ID')\n", "queryDF.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Dataset Implementation" ] }, { "cell_type": "code", "execution_count": 19, "metadata": {}, "outputs": [], "source": [ "# Deferred to the second notebook in this series on Model Training." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Using Feature Store\n", "Let us now see how the feature store objects and operations can be leveraged through the various phases of the ML flow. In this notebook, the focus is on Feature Engineering. Future notebooks will look into Model Training and Model Serving scenarios." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Example: Credit Card Fraud Data\n", "The demo data is abridged from its original version from [here](https://www.kaggle.com/mlg-ulb/creditcardfraud). It represents real transactions by European cardholders in 2013. The original dataset has close to 300K transactions, whereas the abdridged version used here contains about a thousand records.\n", "\n", "We will illustrate use of the feature store for Feature Engineering through the following sequence:\n", "- Read the demo data from a csv file. \n", "- Perform feature engineering tasks.\n", "- Save the engineered features to the Feature Store." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Read Data into Dataframe\n", "Read and examine the columns and rows.\n", "\n", "The data contains transformed versions of PCA (\"V1\"-\"V28\") with 29 feature columns and 1 label (\"Class\") column. " ] }, { "cell_type": "code", "execution_count": 20, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "RangeIndex: 984 entries, 0 to 983\n", "Data columns (total 32 columns):\n", " # Column Non-Null Count Dtype \n", "--- ------ -------------- ----- \n", " 0 Unnamed: 0 984 non-null int64 \n", " 1 Time 984 non-null float64\n", " 2 V1 984 non-null float64\n", " 3 V2 984 non-null float64\n", " 4 V3 984 non-null float64\n", " 5 V4 984 non-null float64\n", " 6 V5 984 non-null float64\n", " 7 V6 984 non-null float64\n", " 8 V7 984 non-null float64\n", " 9 V8 984 non-null float64\n", " 10 V9 984 non-null float64\n", " 11 V10 984 non-null float64\n", " 12 V11 984 non-null float64\n", " 13 V12 984 non-null float64\n", " 14 V13 984 non-null float64\n", " 15 V14 984 non-null float64\n", " 16 V15 984 non-null float64\n", " 17 V16 984 non-null float64\n", " 18 V17 984 non-null float64\n", " 19 V18 984 non-null float64\n", " 20 V19 984 non-null float64\n", " 21 V20 984 non-null float64\n", " 22 V21 984 non-null float64\n", " 23 V22 984 non-null float64\n", " 24 V23 984 non-null float64\n", " 25 V24 984 non-null float64\n", " 26 V25 984 non-null float64\n", " 27 V26 984 non-null float64\n", " 28 V27 984 non-null float64\n", " 29 V28 984 non-null float64\n", " 30 Amount 984 non-null float64\n", " 31 Class 984 non-null int64 \n", "dtypes: float64(30), int64(2)\n", "memory usage: 246.1 KB\n" ] } ], "source": [ "import pandas as pd \n", "data=pd.read_csv(\"resources/creditcard_small.csv\")\n", "data.info()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The data contains all fraudulent transactions from the original dataset, and the same number of randomly selected non-fraudulent transactions." ] }, { "cell_type": "code", "execution_count": 21, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Total number of transactions: 984\n", "Number of normal transactions 492\n", "Number of fraudulent transactions: 492\n" ] } ], "source": [ "total = len(data)\n", "normal = len(data[data.Class == 0])\n", "fraudulent = len(data[data.Class == 1])\n", "fraud_percentage = round(fraudulent/total*100, 2)\n", "print('Total number of transactions: {}'.format(total))\n", "print('Number of normal transactions {}'.format(normal))\n", "print('Number of fraudulent transactions: {}'.format(fraudulent))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Here is how the data looks:" ] }, { "cell_type": "code", "execution_count": 22, "metadata": {}, "outputs": [ { "data": { "text/html": [ "
\n", "\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
Unnamed: 0TimeV1V2V3V4V5V6V7V8...V21V22V23V24V25V26V27V28AmountClass
012177.0-0.4271910.7457081.761811-0.1651300.058298-0.2134130.6473230.073464...-0.201681-0.4320700.0131640.161606-0.4013100.0474230.102549-0.1165719.120
1248296153875.0-0.6136963.698772-5.5349415.6204861.649263-2.335145-0.9071880.706362...0.319261-0.471379-0.075890-0.667909-0.6428480.0706000.4884100.2923450.001
2239160.01.1714390.4749740.0117611.2643030.116234-0.8659860.554393-0.276375...0.0700510.278843-0.0974910.4262780.744938-0.2747280.0084720.01549220.000
3239501150139.0-6.682832-2.714268-5.7745301.449792-0.661836-1.1486500.8496860.433427...0.2205261.1870130.3358210.2156830.8031100.044033-0.0549880.082337237.261
414333685285.0-6.7134073.921104-9.7466785.148263-5.151563-2.099389-5.9377673.578780...0.954272-0.4510860.127214-0.3394500.3940961.0752951.649906-0.394905252.921
\n", "

5 rows × 32 columns

\n", "
" ], "text/plain": [ " Unnamed: 0 Time V1 V2 V3 V4 V5 \\\n", "0 121 77.0 -0.427191 0.745708 1.761811 -0.165130 0.058298 \n", "1 248296 153875.0 -0.613696 3.698772 -5.534941 5.620486 1.649263 \n", "2 239 160.0 1.171439 0.474974 0.011761 1.264303 0.116234 \n", "3 239501 150139.0 -6.682832 -2.714268 -5.774530 1.449792 -0.661836 \n", "4 143336 85285.0 -6.713407 3.921104 -9.746678 5.148263 -5.151563 \n", "\n", " V6 V7 V8 ... V21 V22 V23 V24 \\\n", "0 -0.213413 0.647323 0.073464 ... -0.201681 -0.432070 0.013164 0.161606 \n", "1 -2.335145 -0.907188 0.706362 ... 0.319261 -0.471379 -0.075890 -0.667909 \n", "2 -0.865986 0.554393 -0.276375 ... 0.070051 0.278843 -0.097491 0.426278 \n", "3 -1.148650 0.849686 0.433427 ... 0.220526 1.187013 0.335821 0.215683 \n", "4 -2.099389 -5.937767 3.578780 ... 0.954272 -0.451086 0.127214 -0.339450 \n", "\n", " V25 V26 V27 V28 Amount Class \n", "0 -0.401310 0.047423 0.102549 -0.116571 9.12 0 \n", "1 -0.642848 0.070600 0.488410 0.292345 0.00 1 \n", "2 0.744938 -0.274728 0.008472 0.015492 20.00 0 \n", "3 0.803110 0.044033 -0.054988 0.082337 237.26 1 \n", "4 0.394096 1.075295 1.649906 -0.394905 252.92 1 \n", "\n", "[5 rows x 32 columns]" ] }, "execution_count": 22, "metadata": {}, "output_type": "execute_result" } ], "source": [ "data.head()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Perform Feature Engineering Tasks\n", "We will perform some simple data transformations:\n", "- order the rows \n", "- rename the columns to include the feature group prefix \"CC1\"\n", "- add a transaction id column \"TxnId\" and assign a unique generated value to it\n", "- select (drop the index column) and order columns" ] }, { "cell_type": "code", "execution_count": 23, "metadata": {}, "outputs": [ { "data": { "text/html": [ "
\n", "\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
OldIdxTimeV1V2V3V4V5V6V7V8...V21V22V23V24V25V26V27V28AmountClass
000.0-1.359807-0.0727812.5363471.378155-0.3383210.4623880.2395990.098698...-0.0183070.277838-0.1104740.0669280.128539-0.1891150.133558-0.021053149.620
110.01.1918570.2661510.1664800.4481540.060018-0.082361-0.0788030.085102...-0.225775-0.6386720.101288-0.3398460.1671700.125895-0.0089830.0147242.690
221.0-1.358354-1.3401631.7732090.379780-0.5031981.8004990.7914610.247676...0.2479980.7716790.909412-0.689281-0.327642-0.139097-0.055353-0.059752378.660
331.0-0.966272-0.1852261.792993-0.863291-0.0103091.2472030.2376090.377436...-0.1083000.005274-0.190321-1.1755750.647376-0.2219290.0627230.061458123.500
442.0-1.1582330.8777371.5487180.403034-0.4071930.0959210.592941-0.270533...-0.0094310.798278-0.1374580.141267-0.2060100.5022920.2194220.21515369.990
..................................................................
979279863169142.0-1.9278831.125653-4.5183311.749293-1.566487-2.010494-0.8828500.697211...0.778584-0.3191890.639419-0.2948850.5375030.7883950.2926800.147968390.001
980280143169347.01.3785591.289381-5.0042471.4118500.442581-1.326536-1.4131700.248525...0.3706120.028234-0.145640-0.0810490.5218750.7394670.3891520.1866370.761
981280149169351.0-0.6761431.126366-2.2137000.468308-1.120541-0.003346-2.2347391.210158...0.7518260.8341080.1909440.032070-0.7396950.4711110.3851070.19436177.891
982281144169966.0-3.1138320.585864-5.3997301.817092-0.840618-2.943548-2.2080021.058733...0.583276-0.269209-0.456108-0.183659-0.3281680.6061160.884876-0.253700245.001
983281674170348.01.9919760.158476-2.5834410.4086701.151147-0.0966950.223050-0.068384...-0.164350-0.295135-0.072173-0.4502610.313267-0.2896170.002988-0.01530942.531
\n", "

984 rows × 32 columns

\n", "
" ], "text/plain": [ " OldIdx Time V1 V2 V3 V4 V5 \\\n", "0 0 0.0 -1.359807 -0.072781 2.536347 1.378155 -0.338321 \n", "1 1 0.0 1.191857 0.266151 0.166480 0.448154 0.060018 \n", "2 2 1.0 -1.358354 -1.340163 1.773209 0.379780 -0.503198 \n", "3 3 1.0 -0.966272 -0.185226 1.792993 -0.863291 -0.010309 \n", "4 4 2.0 -1.158233 0.877737 1.548718 0.403034 -0.407193 \n", ".. ... ... ... ... ... ... ... \n", "979 279863 169142.0 -1.927883 1.125653 -4.518331 1.749293 -1.566487 \n", "980 280143 169347.0 1.378559 1.289381 -5.004247 1.411850 0.442581 \n", "981 280149 169351.0 -0.676143 1.126366 -2.213700 0.468308 -1.120541 \n", "982 281144 169966.0 -3.113832 0.585864 -5.399730 1.817092 -0.840618 \n", "983 281674 170348.0 1.991976 0.158476 -2.583441 0.408670 1.151147 \n", "\n", " V6 V7 V8 ... V21 V22 V23 \\\n", "0 0.462388 0.239599 0.098698 ... -0.018307 0.277838 -0.110474 \n", "1 -0.082361 -0.078803 0.085102 ... -0.225775 -0.638672 0.101288 \n", "2 1.800499 0.791461 0.247676 ... 0.247998 0.771679 0.909412 \n", "3 1.247203 0.237609 0.377436 ... -0.108300 0.005274 -0.190321 \n", "4 0.095921 0.592941 -0.270533 ... -0.009431 0.798278 -0.137458 \n", ".. ... ... ... ... ... ... ... \n", "979 -2.010494 -0.882850 0.697211 ... 0.778584 -0.319189 0.639419 \n", "980 -1.326536 -1.413170 0.248525 ... 0.370612 0.028234 -0.145640 \n", "981 -0.003346 -2.234739 1.210158 ... 0.751826 0.834108 0.190944 \n", "982 -2.943548 -2.208002 1.058733 ... 0.583276 -0.269209 -0.456108 \n", "983 -0.096695 0.223050 -0.068384 ... -0.164350 -0.295135 -0.072173 \n", "\n", " V24 V25 V26 V27 V28 Amount Class \n", "0 0.066928 0.128539 -0.189115 0.133558 -0.021053 149.62 0 \n", "1 -0.339846 0.167170 0.125895 -0.008983 0.014724 2.69 0 \n", "2 -0.689281 -0.327642 -0.139097 -0.055353 -0.059752 378.66 0 \n", "3 -1.175575 0.647376 -0.221929 0.062723 0.061458 123.50 0 \n", "4 0.141267 -0.206010 0.502292 0.219422 0.215153 69.99 0 \n", ".. ... ... ... ... ... ... ... \n", "979 -0.294885 0.537503 0.788395 0.292680 0.147968 390.00 1 \n", "980 -0.081049 0.521875 0.739467 0.389152 0.186637 0.76 1 \n", "981 0.032070 -0.739695 0.471111 0.385107 0.194361 77.89 1 \n", "982 -0.183659 -0.328168 0.606116 0.884876 -0.253700 245.00 1 \n", "983 -0.450261 0.313267 -0.289617 0.002988 -0.015309 42.53 1 \n", "\n", "[984 rows x 32 columns]" ] }, "execution_count": 23, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# rename the index column from the orignal dataset\n", "data = data.rename(columns={\"Unnamed: 0\": \"OldIdx\"})\n", "# order the rows by timestamp (original index order)\n", "data = data.sort_values(\"OldIdx\")\n", "data.reset_index(drop=True, inplace=True)\n", "data" ] }, { "cell_type": "code", "execution_count": 24, "metadata": {}, "outputs": [ { "data": { "text/html": [ "
\n", "\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
CC1_OldIdxCC1_TimeCC1_V1CC1_V2CC1_V3CC1_V4CC1_V5CC1_V6CC1_V7CC1_V8...CC1_V21CC1_V22CC1_V23CC1_V24CC1_V25CC1_V26CC1_V27CC1_V28CC1_AmountCC1_Class
000.0-1.359807-0.0727812.5363471.378155-0.3383210.4623880.2395990.098698...-0.0183070.277838-0.1104740.0669280.128539-0.1891150.133558-0.021053149.620
110.01.1918570.2661510.1664800.4481540.060018-0.082361-0.0788030.085102...-0.225775-0.6386720.101288-0.3398460.1671700.125895-0.0089830.0147242.690
221.0-1.358354-1.3401631.7732090.379780-0.5031981.8004990.7914610.247676...0.2479980.7716790.909412-0.689281-0.327642-0.139097-0.055353-0.059752378.660
331.0-0.966272-0.1852261.792993-0.863291-0.0103091.2472030.2376090.377436...-0.1083000.005274-0.190321-1.1755750.647376-0.2219290.0627230.061458123.500
442.0-1.1582330.8777371.5487180.403034-0.4071930.0959210.592941-0.270533...-0.0094310.798278-0.1374580.141267-0.2060100.5022920.2194220.21515369.990
..................................................................
979279863169142.0-1.9278831.125653-4.5183311.749293-1.566487-2.010494-0.8828500.697211...0.778584-0.3191890.639419-0.2948850.5375030.7883950.2926800.147968390.001
980280143169347.01.3785591.289381-5.0042471.4118500.442581-1.326536-1.4131700.248525...0.3706120.028234-0.145640-0.0810490.5218750.7394670.3891520.1866370.761
981280149169351.0-0.6761431.126366-2.2137000.468308-1.120541-0.003346-2.2347391.210158...0.7518260.8341080.1909440.032070-0.7396950.4711110.3851070.19436177.891
982281144169966.0-3.1138320.585864-5.3997301.817092-0.840618-2.943548-2.2080021.058733...0.583276-0.269209-0.456108-0.183659-0.3281680.6061160.884876-0.253700245.001
983281674170348.01.9919760.158476-2.5834410.4086701.151147-0.0966950.223050-0.068384...-0.164350-0.295135-0.072173-0.4502610.313267-0.2896170.002988-0.01530942.531
\n", "

984 rows × 32 columns

\n", "
" ], "text/plain": [ " CC1_OldIdx CC1_Time CC1_V1 CC1_V2 CC1_V3 CC1_V4 CC1_V5 \\\n", "0 0 0.0 -1.359807 -0.072781 2.536347 1.378155 -0.338321 \n", "1 1 0.0 1.191857 0.266151 0.166480 0.448154 0.060018 \n", "2 2 1.0 -1.358354 -1.340163 1.773209 0.379780 -0.503198 \n", "3 3 1.0 -0.966272 -0.185226 1.792993 -0.863291 -0.010309 \n", "4 4 2.0 -1.158233 0.877737 1.548718 0.403034 -0.407193 \n", ".. ... ... ... ... ... ... ... \n", "979 279863 169142.0 -1.927883 1.125653 -4.518331 1.749293 -1.566487 \n", "980 280143 169347.0 1.378559 1.289381 -5.004247 1.411850 0.442581 \n", "981 280149 169351.0 -0.676143 1.126366 -2.213700 0.468308 -1.120541 \n", "982 281144 169966.0 -3.113832 0.585864 -5.399730 1.817092 -0.840618 \n", "983 281674 170348.0 1.991976 0.158476 -2.583441 0.408670 1.151147 \n", "\n", " CC1_V6 CC1_V7 CC1_V8 ... CC1_V21 CC1_V22 CC1_V23 \\\n", "0 0.462388 0.239599 0.098698 ... -0.018307 0.277838 -0.110474 \n", "1 -0.082361 -0.078803 0.085102 ... -0.225775 -0.638672 0.101288 \n", "2 1.800499 0.791461 0.247676 ... 0.247998 0.771679 0.909412 \n", "3 1.247203 0.237609 0.377436 ... -0.108300 0.005274 -0.190321 \n", "4 0.095921 0.592941 -0.270533 ... -0.009431 0.798278 -0.137458 \n", ".. ... ... ... ... ... ... ... \n", "979 -2.010494 -0.882850 0.697211 ... 0.778584 -0.319189 0.639419 \n", "980 -1.326536 -1.413170 0.248525 ... 0.370612 0.028234 -0.145640 \n", "981 -0.003346 -2.234739 1.210158 ... 0.751826 0.834108 0.190944 \n", "982 -2.943548 -2.208002 1.058733 ... 0.583276 -0.269209 -0.456108 \n", "983 -0.096695 0.223050 -0.068384 ... -0.164350 -0.295135 -0.072173 \n", "\n", " CC1_V24 CC1_V25 CC1_V26 CC1_V27 CC1_V28 CC1_Amount CC1_Class \n", "0 0.066928 0.128539 -0.189115 0.133558 -0.021053 149.62 0 \n", "1 -0.339846 0.167170 0.125895 -0.008983 0.014724 2.69 0 \n", "2 -0.689281 -0.327642 -0.139097 -0.055353 -0.059752 378.66 0 \n", "3 -1.175575 0.647376 -0.221929 0.062723 0.061458 123.50 0 \n", "4 0.141267 -0.206010 0.502292 0.219422 0.215153 69.99 0 \n", ".. ... ... ... ... ... ... ... \n", "979 -0.294885 0.537503 0.788395 0.292680 0.147968 390.00 1 \n", "980 -0.081049 0.521875 0.739467 0.389152 0.186637 0.76 1 \n", "981 0.032070 -0.739695 0.471111 0.385107 0.194361 77.89 1 \n", "982 -0.183659 -0.328168 0.606116 0.884876 -0.253700 245.00 1 \n", "983 -0.450261 0.313267 -0.289617 0.002988 -0.015309 42.53 1 \n", "\n", "[984 rows x 32 columns]" ] }, "execution_count": 24, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# rename the columns to include the feature group prefix \"CC1\"\n", "curr_columns = data.columns\n", "data = data.rename(columns=dict(zip(curr_columns, [\"CC1_\"+c for c in curr_columns])))\n", "data" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Create a Spark dataframe to save data to Aerospike efficiently through the Aerospike Spark Connector." ] }, { "cell_type": "code", "execution_count": 25, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- CC1_OldIdx: long (nullable = true)\n", " |-- CC1_Time: double (nullable = true)\n", " |-- CC1_V1: double (nullable = true)\n", " |-- CC1_V2: double (nullable = true)\n", " |-- CC1_V3: double (nullable = true)\n", " |-- CC1_V4: double (nullable = true)\n", " |-- CC1_V5: double (nullable = true)\n", " |-- CC1_V6: double (nullable = true)\n", " |-- CC1_V7: double (nullable = true)\n", " |-- CC1_V8: double (nullable = true)\n", " |-- CC1_V9: double (nullable = true)\n", " |-- CC1_V10: double (nullable = true)\n", " |-- CC1_V11: double (nullable = true)\n", " |-- CC1_V12: double (nullable = true)\n", " |-- CC1_V13: double (nullable = true)\n", " |-- CC1_V14: double (nullable = true)\n", " |-- CC1_V15: double (nullable = true)\n", " |-- CC1_V16: double (nullable = true)\n", " |-- CC1_V17: double (nullable = true)\n", " |-- CC1_V18: double (nullable = true)\n", " |-- CC1_V19: double (nullable = true)\n", " |-- CC1_V20: double (nullable = true)\n", " |-- CC1_V21: double (nullable = true)\n", " |-- CC1_V22: double (nullable = true)\n", " |-- CC1_V23: double (nullable = true)\n", " |-- CC1_V24: double (nullable = true)\n", " |-- CC1_V25: double (nullable = true)\n", " |-- CC1_V26: double (nullable = true)\n", " |-- CC1_V27: double (nullable = true)\n", " |-- CC1_V28: double (nullable = true)\n", " |-- CC1_Amount: double (nullable = true)\n", " |-- CC1_Class: long (nullable = true)\n", "\n" ] }, { "data": { "text/html": [ "
\n", "\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
CC1_OldIdxCC1_TimeCC1_V1CC1_V2CC1_V3CC1_V4CC1_V5CC1_V6CC1_V7CC1_V8...CC1_V21CC1_V22CC1_V23CC1_V24CC1_V25CC1_V26CC1_V27CC1_V28CC1_AmountCC1_Class
000.0-1.359807-0.0727812.5363471.378155-0.3383210.4623880.2395990.098698...-0.0183070.277838-0.1104740.0669280.128539-0.1891150.133558-0.021053149.620
110.01.1918570.2661510.1664800.4481540.060018-0.082361-0.0788030.085102...-0.225775-0.6386720.101288-0.3398460.1671700.125895-0.0089830.0147242.690
221.0-1.358354-1.3401631.7732090.379780-0.5031981.8004990.7914610.247676...0.2479980.7716790.909412-0.689281-0.327642-0.139097-0.055353-0.059752378.660
331.0-0.966272-0.1852261.792993-0.863291-0.0103091.2472030.2376090.377436...-0.1083000.005274-0.190321-1.1755750.647376-0.2219290.0627230.061458123.500
442.0-1.1582330.8777371.5487180.403034-0.4071930.0959210.592941-0.270533...-0.0094310.798278-0.1374580.141267-0.2060100.5022920.2194220.21515369.990
..................................................................
979279863169142.0-1.9278831.125653-4.5183311.749293-1.566487-2.010494-0.8828500.697211...0.778584-0.3191890.639419-0.2948850.5375030.7883950.2926800.147968390.001
980280143169347.01.3785591.289381-5.0042471.4118500.442581-1.326536-1.4131700.248525...0.3706120.028234-0.145640-0.0810490.5218750.7394670.3891520.1866370.761
981280149169351.0-0.6761431.126366-2.2137000.468308-1.120541-0.003346-2.2347391.210158...0.7518260.8341080.1909440.032070-0.7396950.4711110.3851070.19436177.891
982281144169966.0-3.1138320.585864-5.3997301.817092-0.840618-2.943548-2.2080021.058733...0.583276-0.269209-0.456108-0.183659-0.3281680.6061160.884876-0.253700245.001
983281674170348.01.9919760.158476-2.5834410.4086701.151147-0.0966950.223050-0.068384...-0.164350-0.295135-0.072173-0.4502610.313267-0.2896170.002988-0.01530942.531
\n", "

984 rows × 32 columns

\n", "
" ], "text/plain": [ " CC1_OldIdx CC1_Time CC1_V1 CC1_V2 CC1_V3 CC1_V4 CC1_V5 \\\n", "0 0 0.0 -1.359807 -0.072781 2.536347 1.378155 -0.338321 \n", "1 1 0.0 1.191857 0.266151 0.166480 0.448154 0.060018 \n", "2 2 1.0 -1.358354 -1.340163 1.773209 0.379780 -0.503198 \n", "3 3 1.0 -0.966272 -0.185226 1.792993 -0.863291 -0.010309 \n", "4 4 2.0 -1.158233 0.877737 1.548718 0.403034 -0.407193 \n", ".. ... ... ... ... ... ... ... \n", "979 279863 169142.0 -1.927883 1.125653 -4.518331 1.749293 -1.566487 \n", "980 280143 169347.0 1.378559 1.289381 -5.004247 1.411850 0.442581 \n", "981 280149 169351.0 -0.676143 1.126366 -2.213700 0.468308 -1.120541 \n", "982 281144 169966.0 -3.113832 0.585864 -5.399730 1.817092 -0.840618 \n", "983 281674 170348.0 1.991976 0.158476 -2.583441 0.408670 1.151147 \n", "\n", " CC1_V6 CC1_V7 CC1_V8 ... CC1_V21 CC1_V22 CC1_V23 \\\n", "0 0.462388 0.239599 0.098698 ... -0.018307 0.277838 -0.110474 \n", "1 -0.082361 -0.078803 0.085102 ... -0.225775 -0.638672 0.101288 \n", "2 1.800499 0.791461 0.247676 ... 0.247998 0.771679 0.909412 \n", "3 1.247203 0.237609 0.377436 ... -0.108300 0.005274 -0.190321 \n", "4 0.095921 0.592941 -0.270533 ... -0.009431 0.798278 -0.137458 \n", ".. ... ... ... ... ... ... ... \n", "979 -2.010494 -0.882850 0.697211 ... 0.778584 -0.319189 0.639419 \n", "980 -1.326536 -1.413170 0.248525 ... 0.370612 0.028234 -0.145640 \n", "981 -0.003346 -2.234739 1.210158 ... 0.751826 0.834108 0.190944 \n", "982 -2.943548 -2.208002 1.058733 ... 0.583276 -0.269209 -0.456108 \n", "983 -0.096695 0.223050 -0.068384 ... -0.164350 -0.295135 -0.072173 \n", "\n", " CC1_V24 CC1_V25 CC1_V26 CC1_V27 CC1_V28 CC1_Amount CC1_Class \n", "0 0.066928 0.128539 -0.189115 0.133558 -0.021053 149.62 0 \n", "1 -0.339846 0.167170 0.125895 -0.008983 0.014724 2.69 0 \n", "2 -0.689281 -0.327642 -0.139097 -0.055353 -0.059752 378.66 0 \n", "3 -1.175575 0.647376 -0.221929 0.062723 0.061458 123.50 0 \n", "4 0.141267 -0.206010 0.502292 0.219422 0.215153 69.99 0 \n", ".. ... ... ... ... ... ... ... \n", "979 -0.294885 0.537503 0.788395 0.292680 0.147968 390.00 1 \n", "980 -0.081049 0.521875 0.739467 0.389152 0.186637 0.76 1 \n", "981 0.032070 -0.739695 0.471111 0.385107 0.194361 77.89 1 \n", "982 -0.183659 -0.328168 0.606116 0.884876 -0.253700 245.00 1 \n", "983 -0.450261 0.313267 -0.289617 0.002988 -0.015309 42.53 1 \n", "\n", "[984 rows x 32 columns]" ] }, "execution_count": 25, "metadata": {}, "output_type": "execute_result" } ], "source": [ "sparkDF = spark.createDataFrame(data) \n", "sparkDF.printSchema()\n", "sparkDF.toPandas()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Add an identifier column TxnId. Generate and assign a unique value to it." ] }, { "cell_type": "code", "execution_count": 26, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+---------+----------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+\n", "|TxnId|CC1_Class|CC1_Amount|CC1_V1|CC1_V2|CC1_V3|CC1_V4|CC1_V5|CC1_V6|CC1_V7|CC1_V8|CC1_V9|CC1_V10|CC1_V11|CC1_V12|CC1_V13|CC1_V14|CC1_V15|CC1_V16|CC1_V17|CC1_V18|CC1_V19|CC1_V20|CC1_V21|CC1_V22|CC1_V23|CC1_V24|CC1_V25|CC1_V26|CC1_V27|CC1_V28|\n", "+-----+---------+----------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+\n", "| 1| 0| 149| -1.| -0.| 2.5| 1.3| -0.| 0.4| 0.2| 0.0| 0.3| 0.0| -0.| -0.| -0.| -0.| 1.4| -0.| 0.2| 0.0| 0.4| 0.2| -0.| 0.2| -0.| 0.0| 0.1| -0.| 0.1| -0.|\n", "| 2| 0| 2.6| 1.1| 0.2| 0.1| 0.4| 0.0| -0.| -0.| 0.0| -0.| -0.| 1.6| 1.0| 0.4| -0.| 0.6| 0.4| -0.| -0.| -0.| -0.| -0.| -0.| 0.1| -0.| 0.1| 0.1| -0.| 0.0|\n", "| 3| 0| 378| -1.| -1.| 1.7| 0.3| -0.| 1.8| 0.7| 0.2| -1.| 0.2| 0.6| 0.0| 0.7| -0.| 2.3| -2.| 1.1| -0.| -2.| 0.5| 0.2| 0.7| 0.9| -0.| -0.| -0.| -0.| -0.|\n", "+-----+---------+----------+------+------+------+------+------+------+------+------+------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+\n", "only showing top 3 rows\n", "\n" ] } ], "source": [ "# add a new column TxnId for transaction id\n", "# define a udf to update the new TxnId column \n", "\n", "import pyspark.sql.functions as F\n", "\n", "curr_txn_id = 1\n", "def get_txn_id():\n", " global curr_txn_id\n", " txn_id = str(curr_txn_id)\n", " curr_txn_id += 1\n", " return txn_id\n", "txn_id_udf = F.UserDefinedFunction(get_txn_id, StringType())\n", "sparkDF = sparkDF.withColumn('TxnId', txn_id_udf())\n", "\n", "# select the needed columns \n", "sparkDF = sparkDF.select(['TxnId','CC1_Class','CC1_Amount']+['CC1_V'+str(i) for i in range(1,29)])\n", "sparkDF.show(3, truncate=3)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Save Features to Feature Store\n", "We are ready to save the features in the Spark dataframe to Aerospike. \n", "\n", "Of the following three steps, the first two steps are needed only the first time when the feature group and features are created. Subsequent updates will use only the third step.\n", "1. Create a feature group.\n", "2. Create feature metadata.\n", "3. Save feature values in entity records." ] }, { "cell_type": "code", "execution_count": 27, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Features stored to Feature Store.\n" ] } ], "source": [ "# 1. Create a feature group.\n", "FG_NAME = 'CC1'\n", "FG_DESCRIPTION = 'Credit card transaction data'\n", "FG_SOURCE = 'European cardholder dataset from Kaggle'\n", "fg = FeatureGroup(FG_NAME, FG_DESCRIPTION, FG_SOURCE,\n", " attrs={'entity':'cctxn', 'class':'fraud'}, tags=['kaggle', 'demo'])\n", "fg.save()\n", "\n", "# 2. Create feature metadata\n", "FEATURE_AMOUNT = 'Amount'\n", "f = Feature(FG_NAME, FEATURE_AMOUNT, 'double', \"Transaction amount\", \n", " attrs={'entity':'cctxn'}, tags=['usd'])\n", "f.save()\n", "FEATURE_CLASS = 'Class'\n", "f = Feature(FG_NAME, FEATURE_CLASS, 'integer', \"Label indicating fraud or not\", \n", " attrs={'entity':'cctxn'}, tags=['label'])\n", "f.save()\n", "FEATURE_PCA_XFORM = \"V\"\n", "for i in range(1,29):\n", " f = Feature(FG_NAME, FEATURE_PCA_XFORM+str(i), 'double', \"Transformed version of PCA\", \n", " attrs={'entity':'cctxn'}, tags=['pca'])\n", " f.save()\n", "\n", "# 3. Save feature values in entity records\n", "ENTITY_TYPE = 'cctxn'\n", "ID_COLUMN = 'TxnId'\n", "Entity.saveDF(sparkDF, ENTITY_TYPE, ID_COLUMN)\n", "print('Features stored to Feature Store.')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Query Features\n", "Let's issue a query against the saved features to get fraudulent transactions with amount greater than $2000. There appears to be only one such transaction in the demo data." ] }, { "cell_type": "code", "execution_count": 28, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-----+---------+----------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+\n", "|TxnId|CC1_Class|CC1_Amount| CC1_V1| CC1_V2| CC1_V3| CC1_V4| CC1_V5| CC1_V6| CC1_V7| CC1_V8| CC1_V9|CC1_V10|CC1_V11|CC1_V12|CC1_V13|CC1_V14|CC1_V15|CC1_V16|CC1_V17|CC1_V18|CC1_V19|CC1_V20|CC1_V21|CC1_V22|CC1_V23|CC1_V24|CC1_V25|CC1_V26|CC1_V27|CC1_V28|\n", "+-----+---------+----------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+\n", "| 362| 1| 2125.87|-2.0...|-7.1...|-4.0...|1.30...|-2.0...|-0.0...|2.88...|-0.7...|1.46...|-1.5...|-1.3...|-0.2...|-1.5...|1.07...|0.38...|-0.6...|0.09...|0.33...|0.05...|3.97...|1.24...|-1.0...|-1.8...|0.65...|-0.4...|-0.8...|-0.3...|0.31...|\n", "+-----+---------+----------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+-------+\n", "\n" ] } ], "source": [ "schema = StructType([StructField(ID_COLUMN, StringType(), False),\n", " StructField(FG_NAME+'_'+FEATURE_CLASS, IntegerType(), False),\n", " StructField(FG_NAME+'_'+FEATURE_AMOUNT, DoubleType(), False)])\n", "for i in range(1,29):\n", " schema.add(FG_NAME+'_'+FEATURE_PCA_XFORM+str(i), DoubleType(), True)\n", "queryDF = Entity.query(ENTITY_TYPE, 'CC1_Class == 1 and CC1_Amount > 2000', schema, ID_COLUMN)\n", "queryDF.show(5, truncate=7)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Examine the data through aql with the following command:" ] }, { "cell_type": "code", "execution_count": 37, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "select * from test.cctxn-features where PK='362'\r\n", "+-------+------------+-----------+-------------------+-------------------+-------------------+--------------------+-------------------+-----------------+-------------------+--------------------+--------------------+-------------------+--------------------+-------------------+------------------+------------------+-------------------+-------------------+-------------------+--------------------+--------------------+-------------------+-------------------+-------------------+------------------+-------------------+---------------------+------------------+--------------------+-----------------+-------+\r\n", "| PK | CC1_Amount | CC1_Class | CC1_V1 | CC1_V10 | CC1_V11 | CC1_V12 | CC1_V13 | CC1_V14 | CC1_V15 | CC1_V16 | CC1_V17 | CC1_V18 | CC1_V19 | CC1_V2 | CC1_V20 | CC1_V21 | CC1_V22 | CC1_V23 | CC1_V24 | CC1_V25 | CC1_V26 | CC1_V27 | CC1_V28 | CC1_V3 | CC1_V4 | CC1_V5 | CC1_V6 | CC1_V7 | CC1_V8 | CC1_V9 | TxnId |\r\n", "+-------+------------+-----------+-------------------+-------------------+-------------------+--------------------+-------------------+-----------------+-------------------+--------------------+--------------------+-------------------+--------------------+-------------------+------------------+------------------+-------------------+-------------------+-------------------+--------------------+--------------------+-------------------+-------------------+-------------------+------------------+-------------------+---------------------+------------------+--------------------+-----------------+-------+\r\n", "| \"362\" | 2125.87 | 1 | -2.00345953080582 | -1.53160798206082 | -1.39432826167269 | -0.220718797789479 | -1.53099043146804 | 1.0752476539262 | 0.388383209307268 | -0.660655352312646 | 0.0933209955444861 | 0.335742221574637 | 0.0575510393537501 | -7.15904171709445 | 3.97321702726744 | 1.24428677489095 | -1.01523228673153 | -1.80098486605048 | 0.657585626965743 | -0.435617246752788 | -0.894508922176968 | -0.39755738695085 | 0.314261714087509 | -4.05097631587393 | 1.30957974749918 | -2.05810158798669 | -0.0986209270722274 | 2.88008272715204 | -0.727484046608914 | 1.4603805509699 | \"362\" |\r\n", "+-------+------------+-----------+-------------------+-------------------+-------------------+--------------------+-------------------+-----------------+-------------------+--------------------+--------------------+-------------------+--------------------+-------------------+------------------+------------------+-------------------+-------------------+-------------------+--------------------+--------------------+-------------------+-------------------+-------------------+------------------+-------------------+---------------------+------------------+--------------------+-----------------+-------+\r\n", "1 row in set (0.000 secs)\r\n", "\r\n", "OK\r\n", "\r\n", "\r\n" ] } ], "source": [ "!aql -c \"select * from test.cctxn-features where PK='362'\"" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Example: Salary Data\n", "Let's look at another example of feature store use for data engineering, which is taken from another Aerospike Spark notebook.\n", "\n", "We will illustrate use of the feature store for Feature Engineering through the following sequence:\n", "- Generate demo data.\n", "- Perform feature engineering tasks.\n", "- Save the engineered features to Feature Store." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Generate Demo Data" ] }, { "cell_type": "code", "execution_count": 30, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Data created\n" ] } ], "source": [ "# We create age vs salary data, using three different Gaussian distributions\n", "import numpy as np\n", "import matplotlib.pyplot as plt\n", "import pandas as pd\n", "import math\n", "\n", "# Make sure we get the same results every time this workbook is run\n", "# Otherwise we are occasionally exposed to results not working out as expected\n", "np.random.seed(12345)\n", "\n", "# Create covariance matrix from std devs + correlation\n", "def covariance_matrix(std_dev_1,std_dev_2,correlation):\n", " return [[std_dev_1 ** 2, correlation * std_dev_1 * std_dev_2], \n", " [correlation * std_dev_1 * std_dev_2, std_dev_2 ** 2]]\n", "\n", "# Return a bivariate sample given means/std dev/correlation\n", "def age_salary_sample(distribution_params,sample_size):\n", " mean = [distribution_params[\"age_mean\"], distribution_params[\"salary_mean\"]]\n", " cov = covariance_matrix(distribution_params[\"age_std_dev\"],distribution_params[\"salary_std_dev\"],\n", " distribution_params[\"age_salary_correlation\"])\n", " return np.random.multivariate_normal(mean, cov, sample_size).T\n", "\n", "# Define the characteristics of our age/salary distribution\n", "age_salary_distribution_1 = {\"age_mean\":25,\"salary_mean\":50000,\n", " \"age_std_dev\":1,\"salary_std_dev\":5000,\"age_salary_correlation\":0.3}\n", "\n", "age_salary_distribution_2 = {\"age_mean\":45,\"salary_mean\":80000,\n", " \"age_std_dev\":4,\"salary_std_dev\":8000,\"age_salary_correlation\":0.7}\n", "\n", "age_salary_distribution_3 = {\"age_mean\":35,\"salary_mean\":70000,\n", " \"age_std_dev\":2,\"salary_std_dev\":9000,\"age_salary_correlation\":0.1}\n", "\n", "distribution_data = [age_salary_distribution_1,age_salary_distribution_2,age_salary_distribution_3]\n", "\n", "# Sample age/salary data for each distributions\n", "sample_size_1 = 100;\n", "sample_size_2 = 120;\n", "sample_size_3 = 80;\n", "sample_sizes = [sample_size_1,sample_size_2,sample_size_3]\n", "group_1_ages,group_1_salaries = age_salary_sample(age_salary_distribution_1,sample_size=sample_size_1)\n", "group_2_ages,group_2_salaries = age_salary_sample(age_salary_distribution_2,sample_size=sample_size_2)\n", "group_3_ages,group_3_salaries = age_salary_sample(age_salary_distribution_3,sample_size=sample_size_3)\n", "\n", "ages=np.concatenate([group_1_ages,group_2_ages,group_3_ages])\n", "salaries=np.concatenate([group_1_salaries,group_2_salaries,group_3_salaries])\n", "\n", "print(\"Data created\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Perform Feature Engineering Tasks" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Display simulated age/salary data." ] }, { "cell_type": "code", "execution_count": 31, "metadata": {}, "outputs": [ { "data": { "image/png": "iVBORw0KGgoAAAANSUhEUgAAAZcAAAEGCAYAAACpXNjrAAAAOXRFWHRTb2Z0d2FyZQBNYXRwbG90bGliIHZlcnNpb24zLjUuMSwgaHR0cHM6Ly9tYXRwbG90bGliLm9yZy/YYfK9AAAACXBIWXMAAAsTAAALEwEAmpwYAAA4M0lEQVR4nO2de5QdVZ3vP7/uThObECFNJsMkpDsO4SFMBNJiEAedRF6BK6jcUaaBXMdrBgIOc+cuNA7Li4MTx9eMhotEgzwCaUVEGbjeCMagzOAFtBMxvE0gDzoDIekAASIk6fzuH7UrXX26qk7VOXVO1en+fdY665za9dj7HMj+9v69tqgqhmEYhpElTXkPwDAMwxh5mLgYhmEYmWPiYhiGYWSOiYthGIaROSYuhmEYRua05D2AonDooYdqZ2dn3sMwDMNoKFavXr1dVSeWtpu4ODo7O+nt7c17GIZhGA2FiGwKazezmGEYhpE5Ji6GYRhG5pi4GIZhGJljPpcY9uzZQ19fH2+++WbeQyk8Y8eOZcqUKYwZMybvoRiGUQBMXGLo6+vjoIMOorOzExHJeziFRVXp7++nr6+PadOm5T0cwzAKgJnFYnjzzTdpb283YSmDiNDe3m4rPMNoIHp6oLMTmpq8956ebJ9vK5cymLAkw34nw2gcenpg/nzYtcs73rTJOwbo7s6mD1u5GIZhjDKuumpQWHx27fLas8LEpeBs3bqVv/qrv+Id73gHM2fO5OSTT+auu+6q6xh++MMfcuyxx9LU1GSJpoYxAti8OV17JZi4FBhV5bzzzuPUU0/lueeeY/Xq1dx+++309fUNu3bv3r01G8dxxx3Hj3/8Y0499dSa9WEYRv2YOjVdeyWYuGRJxh6y+++/n9bWVi655JL9bR0dHXz6058G4JZbbuFDH/oQs2fPZs6cOezYsYPzzjuPGTNmMGvWLNauXQvAF77wBb7+9a/vf8Zxxx3Hxo0b2bhxI0cffTTd3d0cc8wxnH/++ewqXSsDxxxzDEcddVRV38UwjOKwaBG0tQ1ta2vz2rPCxCUrfA/Zpk2gOughq0JgnnjiCU488cTYa9asWcOdd97JAw88wNVXX80JJ5zA2rVr+dKXvsTFF19cto9nnnmGBQsW8NRTTzF+/Hiuv/76isdrGEZj0N0NS5dCRweIeO9Ll2bnzAcTl+yog4fssssu413vehfvfve797eddtppTJgwAYAHH3yQiy66CIDZs2fT39/Pzp07Y595+OGHc8oppwBw4YUX8uCDD2Y2XsMwikt3N2zcCPv2ee9ZCguYuGRHDTxkxx57LGvWrNl//K1vfYtVq1axbdu2/W0HHnhg2ee0tLSwb9++/cfBfJTSEGILKTYMIwtMXLKiBh6y2bNn8+abb7JkyZL9bWE+EZ8///M/p8eZ4X75y19y6KGHMn78eDo7O/eL1Jo1a9iwYcP+ezZv3sxDDz0EwPe+9z3e9773VTxewzAMHxOXrKiBh0xE+Ld/+zceeOABpk2bxkknncS8efP4yle+Enr9F77wBVavXs2MGTNYuHAhy5YtA+CjH/0oO3bs4Nhjj+W6667jyCOP3H/PUUcdxbe+9S2OOeYYXn75ZS699NJhz73rrruYMmUKDz30EGeffTZnnHFGxd/JMIzRgahq3mMoBF1dXVqaw/HUU09xzDHHJH9IT4/nY9m82VuxLFqUvSEzQzZu3Mg555zD448/nsnzUv9ehmE0PCKyWlW7Stut/EuWdHcXWkwMwzDqhZnFRjGdnZ2ZrVoMwzCCmLgYhmEYmWPiYhiGYWROzcRFRG4SkZdE5PFA2wQRWSki69z7Ia5dRORaEVkvImtF5MTAPfPc9etEZF6gfaaIPObuuVZcgkZUH4ZhGEb9qOXK5RbgzJK2hcAqVZ0OrHLHAGcB091rPrAEPKEArgbeA5wEXB0QiyXApwL3nVmmD8MwDKNO1ExcVPXfgR0lzecCy9znZcB5gfZb1eNh4GAROQw4A1ipqjtU9WVgJXCmOzdeVR9WL5b61pJnhfXRkBSh5P6VV17J0UcfzYwZM/jwhz/MK6+8Utf+DcNoPOrtc5mkqi+4zy8Ck9znycDzgev6XFtce19Ie1wfwxCR+SLSKyK9wZIqRaEoJfdPO+00Hn/8cdauXcuRRx7JP//zP9esL8MwRga5OfTdiqOmGZzl+lDVparapapdEydOrLq/rPekLkrJ/dNPP52WFi8latasWaHiZhiGEaTe4rLVmbRw7y+59i3A4YHrpri2uPYpIe1xfdSUGlTcL2TJ/Ztuuomzzjor1fcwDGP0UW9xuQfwI77mAXcH2i92UWOzgFedaes+4HQROcQ58k8H7nPndorILBcldnHJs8L6qCn12JM675L7ixYtoqWlhW6rQmAYRhlqGYr8feAh4CgR6RORTwJfBk4TkXXAB90xwArgOWA9cAOwAEBVdwBfBH7jXte4Ntw133X3PAv81LVH9VFTarEndZFK7t9yyy385Cc/oaenx8ryG0aArM3hI4VaRotdoKqHqeoYVZ2iqjeqar+qzlHV6ar6QV8oXJTYZar6p6r6Z6raG3jOTap6hHvdHGjvVdXj3D2XO/8KUX3UmlrsSV2Ukvv33nsvX/3qV7nnnntoK638bBijmFqYw0cKlqGfEbXYk7ooJfcvv/xyXnvtNU477TSOP/74IQEGhjGaqYc5vByFXTmpqr1UmTlzppby5JNPDmuLY/ly1Y4OVRHvffnyVLfXnQ0bNuixxx6b2fPS/l6GUXTK/ZsWUfXWLENfIvUbX1vb0L7b2uo79wC9GjKn2solQ2q9J7VhGPUjicmrFubwNBRh5RSFicsoxkruG0Y0SSbuWpjD01CLQKKsMHExDMMIIcnE3d0NS5dCRweIeO9Ll9bPapH3yikOExfDMIwQkk7ceZrD8145xWHiYhiGEUKRJ26fNCunekeVmbgYhlFXChs6W0LeJq+kJFk55ZGPY+JScIpQcv/zn/88M2bM4Pjjj+f000/nP//zP+vavzFyaLSkw5ESAZpHVJmJS4HRgpTcv/LKK1m7di2PPvoo55xzDtdcc03N+jJGNlGT3Lx5xRWYkUAeUWUmLlmytR8eXgsP9HrvW/urelxRSu6PHz9+/+c33njDaosZFRM1mQ0MFHsF0+jkEVVm4pIVW/vh95vgrd3e8Vu7veMqBKZIJfevuuoqDj/8cHp6emzlYlRM3GRWlOS/kUgewQkmLlmxYYtnmA2yb5/XnhF5ltxftGgRzz//PN3d3Vx33XVZfB1jFBI2yQUpQvJfHI0SjFBKHsEJJi5Z4a9YkrYnoEgl9326u7v50Y9+VLZPwwjDn+Sam8PPFyH5L4pGC0Yopd7BCSYuWXFAa7r2BBSl5P66dev2f7777rs5+uijK/5OhtHdDcuWFT+HpJQi1/EqIiYuWTFtsrdWDtLU5LVXSFFK7i9cuJDjjjuOGTNm8LOf/YzFixdX/J0MAxonhyRIket4FRHxKiYbXV1d2tvbO6Ttqaee4phjjkn+kK39no/lrd3eimXaZJjUnvFIs2Pjxo2cc845mRWvTP17GUYD0dnpmcJK6ejwzEyjFRFZrapdpe22csmSSe0wawa8v8t7L7CwGEbeNIpz3B9nmLAAzJ1b1+E0DCYuoxgruV8hGeczjUYaxTkeHGcUK1ake14jCGoWmLiUwcyGyRg1v1MN8pnyohYTXdJnNopzPGycpST1uTSKoGZFLuIiIleIyOMi8oSI/J1rmyAiK0VknXs/xLWLiFwrIutFZK2InBh4zjx3/ToRmRdonykij7l7rpUKU8rHjh1Lf3//6Jk4K0RV6e/vZ+zYsXkPpfbUIZ+pHtRiokvzzCyd47VcDSQdT5K+G0VQs6LuDn0ROQ64HTgJ2A3cC1wCzAd2qOqXRWQhcIiqflZE5gKfBuYC7wEWq+p7RGQC0At0AQqsBmaq6ssi8mvgb4FHgBXAtar607hxhTn09+zZQ19f35C8ECOcsWPHMmXKFMaMGZP3UGrLA73R594/zKdZWGrhnE7zzKz69wWtdNJub4fFi6uPPovztYTR1hYd9dbU5IluKSLD/15pJKIc+i05jOUY4BFV3QUgIg8AHwHOBT7grlkG/BL4rGu/VT0VfFhEDhaRw9y1K1V1h3vOSuBMEfklMF5VH3bttwLnAbHiEsaYMWOYNm1aRV/SGKEc0BqeGFtFPlMe1CKsNs0zFy0aLgqV5LlEma36+73nQ3UCEzbOOPyVSFifU6eGC1WRE0erIQ+z2OPAn4tIu4i04a1IDgcmqeoL7poXgUnu82Tg+cD9fa4trr0vpN0wqqcG+Ux5UItChmmeGZfnksbMFSeGWZicSseZhKgxNcLmY1lSd3FR1aeArwA/wzOJPQoMlFyjeKaumiIi80WkV0R6gyVVDCOSSe1wZMfgSuWAVu+4wcLOazHRpX1mWDmStL6gcmK4aVNyP0yUqAXH2dFR/jlRY2rExNGqUNVcX8CXgAXAM8Bhru0w4Bn3+TvABYHrn3HnLwC+E2j/jms7DHg60D7kuqjXzJkz1TBGE8uXq3Z0qIp478uXZ/dMUG1u9t7TPNu/t/TV0RHdX1tb+D3BV1tb/BjCnuPfE/yd2ttVW1sr72ckAvRq2Nwe1ljrF/BH7n0q8DRwMPA1YKFrXwh81X0+G89fIsAs4NeufQKwATjEvTYAE9y5X7trxd07t9yYTFwMo3qWL/cm4EonXZHwSVskfZ9JBUo1WtTa24eLzpgxXrsvNv7nrAS60SiauPwH8CTwO2COa2sHVgHrgJ8HhEKAbwHPAo8BXYHn/DWw3r0+EWjvwvPtPAtch4uKi3uZuBhGdZRbRcRN7j5pVy6l/UfdX06gokStmu8yWogSF6st5ggLRTaMYTRY/bh6Ui5stzTktqfHc7hv3uz5KXzfTFgUWRrfRLkw57B+r7oqXchxo4cPZ4nVFjOMahlB2flZUq72lk/Q0R3luIfqnd5xgQVR/c6dG35Pe8TfDSM1fDhL8shzMYzGJC47f5SuXqKSGEspjRqLy1avdiMr/97S1Ul3tyeCYf2uWOGJWNKV1EgNH84SM4s5zCxmlGWEZOdnSZIVS1i2fF7Z6pX0G2ZGG7HhwxVgZjHDqJYa7Dba6MQlMXZ0wPLlsH378Mm4FkmcSaik33pvDzxSMHExjKSMkOz8LImalH3nedREnFe2+mjLks8TExejsann3iojJDs/S8Ima5HymfF5ZauPuiz5HDFxMRqXPKK3Ruluo3GlUfzJGrwJ2/dpBEu3FGmTrFIz169+BS0t3thbWmDBgvzGNpIwh77DHPoNyMNroysUz5pR//GMUMIiwsJyT6Kc++3t8Ic/DL2/tdUToT174p9ZaxYsgCVLhrdfeilcf339xtHIRDn0TVwcJi4NiEVv1YWke69ERWKlobkZli2rn8C0tMDAwPD25mbYu7c+Y2h0LFrMGHlY9FZdSLpPSxaRXgMD9d36N0xY4tqN5Ji4GI1LLaO3SgMFfr+xfoEDBSNp+G5UJFZUlnsUu3bBFVd4KybfDyJSG19Nc3O6diM5Ji5G41Kr6K2wQIEXtlcWOFDPaLYqiXK6Jw3fjYrEWrx4+P3l6O8fNMX5q4hye7tUgl9yJmm7kRzzuTjM52LsJypQoJRygQO+SAVTv5uaChm+XM5pX0mWevCeCRPgrbfg9derH2upr6daFizwvufAgLdimT/fnPlpMId+GUxcjP3EBQqUEhc40EDRbEmd9kkJE6tgmHI1WEXiYmEOfcNIStKAgHLXRa1+kqyK6kxSp30YQXPaoYd6rwsvHF4gMomwJNmn3ioSNwYmLoZRSligQClJAgcaKJqt0lpfpSXs+/u9V6UkEaC5cyt/vk+RkjpHKiYuRnEoivM7LFDgsEPTBw40UC2ySmtuhZXOjyPJyqQcK1ZUd3/Uni4mMNliPheH+VxypoGc36looJ0r0zjt/WvT7N7Y1gbz5sEdd8SvbsIy+oNU63PJ2r802jGfi1Fs4jbiKgKVrqrqXIusGnNP0tLywb/8k+KHJV9/vVeCf/ny8PyXtjYvdHnp0uhck2p9LtX4l0YK9TALmrgYxaDIzu8G2d64XuaeNKawtjZPSErFqrt7UGSiKhQffHD489KUxw+bRPPaS6Yo1Ov/ExMXoxgU2fld9FWVI27r4HKk+Us26V/448aVL0QZtlryJ79S01l7e7rCllGT6Ny5o3tPl2r+P0lDLuIiIv9DRJ4QkcdF5PsiMlZEponIIyKyXkR+ICKt7toD3PF6d74z8JzPufZnROSMQPuZrm29iCzM4SsaaSmy87vIq6oAlZp70v4lm/Qv/DfeGHx+GhNM1Mpo3Lh0BS2jJtEVK0b3ni71MgvW3aEvIpOBB4F3quofROQOYAUwF/ixqt4uIt8GfqeqS0RkATBDVS8RkY8DH1bVj4nIO4HvAycBfwL8HDjSdfN74DSgD/gNcIGqPhk3LnPoF4CiOr/rnQxZ4e8QV/J+3LhoR31aB3dYgmQUYc75MWNg/HjYsSN8PJXscx9GVs8ZaWQd0FA0h34L8DYRaQHagBeA2cCd7vwy4Dz3+Vx3jDs/R0TEtd+uqm+p6gZgPZ7QnASsV9XnVHU3cLu71ig6Rd2Iq56rqir8O2HhxGPGwGuvxa9K0v4lW1pDLK7IY3//cBHas8drjxpP1MpowoTofsIY7b6VKOq11XPdxUVVtwBfBzbjicqrwGrgFVX1d1DoA/x/uZOB5929e9317cH2knui2ochIvNFpFdEerdt21b9lzOKRVZ5M/Xc3rgK/05Y4cjx42F3yaKr1L6eZhL2TVwXXeQd33abt/9KNfkru3Z5Gf3+DpCLFnmbiZWyc2c6p3PUFsxZJGE2MvXa6rnu4iIih+CtJKbhmbMOBM6s9zgAVHWpqnapatfEiRPzGIJRK7KO8KrXqqpK/47vIL/tNu84Kp9k06ZBH8gRR4Rfs337UD9JlG8G4JJLhgtM2nL7S5Z4AtPdDQcdNPz8nj3pnM7d3V5eTXBcqp4YJhGpkZzFnzTsvBryMIt9ENigqttUdQ/wY+AU4GBnJgOYAvh/qm0BDgdw598O9AfbS+6JajdGEw0S4TWMDKLmkuah+AJx//3h5994Y6iIXHFFdJTR9dd7glZtuf2lS733HTvCz6d1Oq9YMdzvkiQyyrL4qycPcdkMzBKRNuc7mQM8CfwCON9dMw+4232+xx3jzt+vXhTCPcDHXTTZNGA68Gs8B/50F33WCnzcXWuMJpKsAIpSbiZIBv6dtCVZksT07NoVvQryJ/ywv4ZLTTDt7d7mX1H4e7ek9ZdErTIqjYyqV7juSCYPn8sjeI75NcBjbgxLgc8Cfy8i6/F8Kje6W24E2l373wML3XOeAO7AE6Z7gctUdcD5ZS4H7gOeAu5w1xqjiXIrgDolRqY2rWTg36l3pnk5B3lQdBYvjvfP+MEBUX6RsPa4VUalTn3L4s8AVbWXKjNnzlRjBPHidtV/X636y98Mvv59tdeuqvrQ74ae818P/S6zISxfrtrWpupNed6rrc1rryUdHUP79F8dHdHnkrza28t/n+XLvT5EvPfS71qu/0svLf8d0nzfSv8bpOl/tAP0asicahn6xsik3AqgDomReZlW4kJNw84lIVjzKyzKqKdncB+X0hXEBz/omcJE4v1Al14Kp5wSnYcB4SuHuFVGpZFR9QrXHdGEKc5ofNnKZZTx4JrwlcuDazLrQiT8r1+RzLqI5NJLVZubvf6amwdXBKrDVxft7fGrCZGh95cStjpI+4paZVS7cqmGcqswwwNbuRijgqRO+igndgLndlLySuLr6fHCbX3n+MDA0PDbUsd7uYgu1eF7qAR9SfPmpQsgKMVfEZQLRIhaOdRqlVGPcN2RjImLUSyqieBK46T3Z96k7RWQl2klyhx3xRXhwQVB01EUQdNTqQO90p+s1EwV5yzv6PBE7Kqr4sc/GmuFFRXbLMxhtcUKQLUbhqWpARZ1rX99sJ5XoNbX63tb+YcbJnPdHe1lN9SCwU21Tpnez1cu2cLk9t3I2NrWTYuqqVVKW9vwSThJ3ak4n0hSmpth796hbXF9L1o0vJ5Z2PiN+lNVbTERiakeZBgZUW3iYxonfVg+SfB6f8VTshoa17Kbb/7NBq799MZEiXXd3bDxkX56/tcmphy62wvDrfF+MEnNbmHBBUlWW1mE4/qZ/Un7tryTxiOpWWydiHzNVSI2jNpQbQRXmuz20miyUvbtg/Wb4ekNwwSvqQkWnLedC+b0J5vg6lwtIE1EWKlQJDExRYlXc/NgouSYMdF9zpnjZfSXEte35Z00HknF5V14Zey/KyIPu4KP42s4LmM0Um3pk7TZ7X69sCj2RjsTmprgS5/yxKHsBBcnmhlWBggWlXzb2wbresUlLYYJRTlHdtQKY9ky757t2+Hmmwd9OH5iZEeHt/Pkz38ePZ7S2mgXXuiFMUeZ+UZ7heMiE1OIYRBVfQ24AbhBRN4PfA/4hojcCXxRVdfXcIzGaGHa5HCfS9LSJ74PI24vlLC9Ug5orSi/ZeofefeUneDinu+byILjr4DSPVb6+wcLR0aVbak0uMAXm6uuit4jxi/9Ugml3yUqYMDyTopNYp+LiHxIRO4Cvgn8C/AO4P/gbfRlGNWTRWn7uOrFUdFkYyvbSnnzS63JJrg4/w5kYiKL8klECQsMTYBMW/23dHUD2VUQTlIbzSLCik+ilQuwDq+w5NdU9f8F2u8UkVOzH5YxapnUXl0UVdwujlG+j1dfT93NG2828a93TU42wZWuqMKosjJAWt9DR8fQ/er9yTxYRj/tXvXVPCNIue8iUtmOiUZ9KbtycZFit6jqJ0uEBQBV/duajMww0lIuzyWr0i4HtHLg8R1ce3t78snTX1FlUFI/jCjTXHt7fPRX2iissFVO1pFc5cyM5mdpDMqKi6oOAOfUYSyGUR3lorLSTuBNTYP3NDdDS7KI/FgzU8qgA79ml4j3OvTQcJNTXITYvHnR0V9porCiqg9H5bxUmgsT913Mz9I4JDWL/UpErgN+ALzhN6rqmpqMyjAg3sQVRjmT07TJXmhxUvbt81Yb/opoYN/g8yKc8GVNREmCDgLP+uu/HrpNcX8/fOITgecx9PMVVwz1s/T3e1FcUea7qVPDRSC4OvBXJ2HXxflGmivMjuvuhl/9Cr797aFRYiKeUJqfpTFIlKEvIr8IaVZVnZ39kPLBMvQLRrls/TDhifJpNDfD+07wPv/qt7EhxsN4f1fZzH9/8t282RtiWHRTMMM9iuBzpk6F11+PdshHPS9Jhn1pn3GZ72Hn01BpAZC038PIj6gMfSv/4jBxKRhxE3pUyPKkCfDC9vDnNTfDdPfneOm9Ufii9ED0/xc9fV2JJl+R+C7TTuJRz4sq/RLXf6moBcOKk5R6aW6uXFCjqOR7GPlQVfkX94CzReQzIvK//Fe2QzSMAHEmrijfyo6d0X6RgYFBU1ZcZn4QX4xinPBJtxQu54ROuzVx2orLcf2XJi5edNGgr6hc5FZbG3zgA8MTNdP6Rkr9VBMmhF9nzvzGIWmey7eBjwGfBgT4r0BMDVXDqJK4qKo44YkzefnO/WAuTBy+H2Ta5OGzpwhMm8zmzXDBnH423L6Wgft72XD7Wi6YM9SWlWSiTRNKPGZM9PMqrcQc5ayPmuRhsFLxQw9V5xsJ63vnTmgt+V/AnPmNRdKVy3tV9WLgZVX9R+Bk4MjaDcsY0SQpqx8XVRUnPOVWJG/tHtpflNe5tL3URuOOL//Lfm64chOdf7zb+6v7j3dzw5Wb6D6tP1X596R/kbe3e6VVop5Xafn5qHBiCBer5cu91c6KFcPvC9v/JW3fe/bAQQdZGf1GJqlD/xFVfY+IPAx8BOgHnlDVI2o9wHphPpc6kaasflS0WNwzoHxEWLC/KAd/SzOc4oIAYvw/r7/hVUou5fW9rYybE1O3rIRyPpf2dq9mV62I83Hcdlu0TyYL30i9/StxPiYjPdX6XH4iIgcDXwPWABuB71c4kKNE5NHAa6eI/J2ITBCRlSKyzr0f4q4XEblWRNaLyFoROTHwrHnu+nUiMi/QPlNEHnP3XCsSV7rPqCtpKgRHlXKJKxOTJLs/2F+UGS3YHmOGCxMWCBecOPwVRxRxZVyyIGrl1NTk+WDAE5nSQpZZ7LZZzx07o8x/1ZSrMcJJJC6q+kVVfUVVf4TnazlaVT9fSYeq+oyqHq+qxwMzgV3AXcBCYJWqTgdWuWOAs4Dp7jUfWAIgIhOAq4H3ACcBV/uC5K75VOC+MysZq1EDsip/EldDLImz3u8vScZ8JWa4FAmbwWrGeRGVuDgwED8JZ7HbZj137LR9YepHrLiIyEdKX8DZwBz3uVrmAM+q6ibgXGCZa18GnOc+nwvcqh4PAweLyGHAGcBKVd2hqi8DK4Ez3bnxqvqweja/WwPPMvKmRuVPhlCuUGSwvyQZ83HXpC3zX0LpX9JRtCdYkFVDd7fnhI9b44dNwllsMVzJMyoptgm2L0w9KZeh/19izinw4yr7/ziD5rVJqvqC+/wiMMl9ngw8H7inz7XFtfeFtA9DRObjrYaYajGO9aHasvpJKFcoMthfkoz5JNekqSQQIEkI8pgxsHhxoscB6X0KcRn4pYRNwtWU16/kGdUUykxSkcDIhlhxUdVP1KpjEWkFPgR8LqRfFZGaZ3eq6lJgKXgO/Vr3Z5Cq/EnV/fjPLFdGJomvJu6aKio5x/3FLJLe4Zx24k2bvFmESTjOtFXud1q0KLwigYU4Z0/S2mKIyNnAscBYv01Vr6mi77OANaq61R1vFZHDVPUFZ9p6ybVvAQ4P3DfFtW0BPlDS/kvXPiXkeqMoVFtWv+j9pSDqL+lKs9vTTrxpkjeLMglXY9pKstGZkQ15JlFewNCIs3sAP+JrHnB3oP1iFzU2C3jVmc/uA04XkUOcI/904D53bqeIzHJRYhcHnmUYhSJrZ3baiTfJ3ilQrDyTaqPLym3jbGRD0pXLe1V1hoisVdV/FJF/AX5aaaciciBwGvA3geYvA3eIyCeBTcBfuvYVwFxgPV5k2ScAVHWHiHwR+I277hpV3eE+LwBuAd7mxlnxWI0CkqRactqKymWu9/0Sp0zv59q/3cyEgwa8ibelGY6YWvHKKOu/pNP6FKKuB09QivhXvZm2GoOk4vIH975LRP4E2AEcVmmnqvoG0F7S1o8XPVZ6rQKXRTznJuCmkPZe4LhKx2cUmNIEytLy91v7Yd3moZUU/Wtefc2rP1YuKbPkmb5f4tyT+7nxMxuG7oq8dwCe2TjYfwLCHO5ZVfpNO/FGXV+UVUoYZtpqDNImUX4VWA1soMIkSsOoirgkzP37roQkRu7b51VMDtulMuqZT2+AB3p5/9i1nHtyP1/61JahwuKjOiQJNC5MttZJfGnDerMIJc4DM20Vn9jyLyLybuB5VX3RHV8MXAg8DXwhYIZqeKz8S4MQU/4+tqhllff4/0xiaz28v6vs/ii2T4kx0qi0/Mt3gN3uAafi+UW+A7yKC+E1akClGWKjgUqqJcfhm8jK4G8zXG5c5TLALYnPGC2UE5fmwOrkY8BSVf2RK/0yYopWFgorfhRPJdWS4/B9L+Uy+uNw5fehvHjUs46WYeRJWXEREd/pPwe4P3AucY6MkQIrfhRPXNHKuCx/kWhRKn1mGVQDpVpamuGozv3O/HLiUc86WoaRJ+UE4vvAAyKyHS9i7D8AROQIPNOYkTVmNylPVFLkpPbocvuqngis3zxY8VjEiyALhiAfPc1ri9ouGZCxrV6xzBDKRWtZpJMxWihX/mWRiKzCCzv+mQ56/5vwEiqNrBntxY/S5qeUEuV78Vcl+wIBLAMDQ0Xkrd1eWHFcBckyddCSiEcWtbgMo+iUNW25SsSlbb+vzXCMUZ0hVi6HJQlxhTHDQo5LiROWhGJn4mEY5jcpHqPZbhKXwxI1oYetdI7sCF/9lNuhshy+QD29wdsGWfBMbLUqvmkYDYyJSxEZrX/6JtlILCgmLc3Dd4z8/SZPXMJ8IpWGK/sExSmsAgCYwBiGo4r4S8PImHIbiflmM18gwrYojtoyGWDC+OrHGEVcv4YxCrGVi1Ecym0klsRnAoPiE1zlNDcnu1fEiyqrxIRWzarIMEYYJi5GcSi3kVjSyfuA1uHBAWH1xsKIc+gn6dcwDMDExSgacRt7JfGZpIkMi2LDFm+lk1SQgv0ahgGYz8VoJMLKtIh4QgBDs/WrMVG9tTtamPy+mpu9gILSfg3DAGzlYjQS5cxmQaqNDAszj7U0wyknVP5Mn1psdmYYBcPExWgs4sxmQcKCA0p5+zh4bdfwAIKoe/YOwMNrq5vokySKZpFMahg5Y2axImIl96vHL0YZx6uvDxUS37wV55gPbjJWCXGJommuMYyCY+JSNKzkfnak+Ss/WCG5XAn+aib6JImiSa4xjIJjZrGiEVdyfyRn7dfKx5DU9xIsM1Pq2wkj7plx36VcYc2k1xhGwbGVS9EYjSX3SzPvqzU9BUmxEdi+N3cPLhAntXslZMpVDSil3HeJ2+wsbswW6mw0GLmIi4gcLCJ3isjTIvKUiJwsIhNEZKWIrHPvh7hrRUSuFZH1IrJWRE4MPGeeu36diMwLtM8UkcfcPdeKxG5QWyyy2qqwkfw2WfkYtvZ7DvcHer33rf2Dvhc/hDiGzVtbh1sgo0rGRLWX+y5xm535JLnGMApOXiuXxcC9qno08C7gKWAhsEpVpwOr3DHAWcB095oPLAEQkQnA1cB7gJOAq31Bctd8KnDfmXX4TtlQyVaFpUKyYEFj+W2y8DHErRgmtcP7TvA2AotYcbzxZhP/cMPk4Zt+7tgZ3l9Ue5nv0tMDne9pp+m9M+ic10XPszOiNz6bNQPe3+W9m7AYDUbdxUVE3g6cCtwIoKq7VfUV4FxgmbtsGXCe+3wucKt6PAwcLCKHAWcAK1V1h6q+DKwEznTnxqvqw25zs1sDzyo+3d2wdCl0dHgJgh0d3nGUvyUsAODb326srZLTmp7CSLL6CU7Yhx3K3r3eT7Z3L9y8YgLfX+VN4EMskGmFL+a71DpWo5EWq8bIJ4+VyzRgG3CziPxWRL4rIgcCk1T1BXfNi8Ak93ky8Hzg/j7XFtfeF9I+DBGZLyK9ItK7bdu2Kr9WhnR3w8aN3uS4cWO8Iz8sACCqPlYlfpswU1PWZOFjSCMCW/th6w5aWjz9bmmBT8zdwQVzvO82xAKZUPj8ib3785PZ9Vb4d4mL1agWCzI0ikYe4tICnAgsUdUTgDcYNIEB4FYcVVQQTIaqLlXVLlXtmjhxYq27qw1pBCOt36aWjvYgWfgY0qx+QlY5B47dx5c+tWW4BTKB8AUn9u/9vJ3//tUONm9t9TQ+8F1qGatRS+EyjErIQ1z6gD5VfcQd34knNludSQv3/pI7vwU4PHD/FNcW1z4lpL2xSGrjSCoYlWyVXM9kvmp9DGlWPxGrnKmTdg+3QCYQvtKJ/fur2un42Aym/beh3yWrWI0wRmOQoVFs6i4uqvoi8LyIHOWa5gBPAvcAfsTXPOBu9/ke4GIXNTYLeNWZz+4DTheRQ5wj/3TgPndup4jMclFiFwee1RiksXEsWgStIX+dNzdDe3syv00UjZTMl2b1E7HKaRrbGv4TlRG+pBN7JbEaSamlcBlGJeQVLfZpoEdE1gLHA18CvgycJiLrgA+6Y4AVwHPAeuAGYAGAqu4Avgj8xr2ucW24a77r7nkW+Gntv1KGpLFxdHfDQQcNbx8YgHHjkvltosjC0V5Pkq5+onJf9g5UZPJLOrGnjdVIQy2FyzAqQbSazZFGEF1dXdrb25v3MDyamsKd8iLhRRXTXp+U0gKKfl8jIediaz+s3zx8q+QKvp+/0Az+PdDWlp1wpBnHVVd5K6apUz1hGclFHYxiICKrVbWrtN0y9ItIGhtHT090Bnq1NpGRnMw3qT08sbICn1ItVyRpx5E0yNAwao3VFisCpX9yHnGE9zm4Gmlrg7lzPee+f93cubBsWfiOiVnZRJKWuC86YfW+MvQpdXfbZG4YQUxc8qbUprJpk/cqpbPTE5Lgdd/+dnROy9veVpPhNiRR+6NEbWVcVJ+SYTQQJi55E+a8D+PJJ4e3xfnL+vs90QL7kzoqpLqlefjmYFYg0jAywXwueVPLRATLovOIMnPtHRi5PiXDyBkTlzwIJkgmLAdfMZs2WbGpuJBqKxBpGDXBxKVe+IIiAhddNJggGWbzj6J054CkOwmM9mJTtj+KYdQdE5d6EMy4h3BfSbn9RkRg9uyh8a5pc5RGq5lsJIdUG0ZBsSRKR02TKDs7wyPAqqHdTYz9KTPKq02szJNabYWcd1+G0cBEJVFatFg9qIXTPk5U4oSnUYtNRYUTQ/aTfj37MowRipnF6sGECfXtr78fdu4cXtCykYtN1bNCcz37MowRiolLrenpgddeq3+/e/Z4BS3zrkmSFfWs0NxI1aANo6CYWazWXHUV7A6ZlEqT92rBjh2wfXtt+6gXB7SGT+61yKavZ1+GMUIxcak1Uf6Wffu8FUVcQEW1AtSo/pUwpk0Or9CcdTjx1v7hlZLT9GWBAIYBmFms9kRN8B0dcMkl0fdVKyyN7F8Jox7hxL4jvzT3qKU5WV/12hbaMBoAE5daM3dudPv118OBB4af37dvMOorCePGDSZVNjfDvHmN61+JotbZ9GGOfPB+zyR9WSCAYezHxKXWrFgR3d7TA2+8EX9/2BbGpVx6qTeJ+Sa2gQGvgvJozMavhmod+RYIYBj7MXGpNXEbrJfLlu/vh717y/exdGn0tsjBOmajub5YEqrd1rnRtoU2jBpi4lJr4naVTJK1n8TvElWfzK8n5tcxG831xZJQbQ0yq2FmGPsxcak1ixZ5zvUgItmWg4mqS9bcHL2iMYZTbdCA1TAzjP3kIi4islFEHhORR0Wk17VNEJGVIrLOvR/i2kVErhWR9SKyVkRODDxnnrt+nYjMC7TPdM9f7+5NWD64CqLMT8EN1r3BpS84WY6WkIjytrboFU0t95BpdKoNGrAS/oYB5Lty+QtVPT5Q8GwhsEpVpwOr3DHAWcB095oPLAFPjICrgfcAJwFX+4LkrvlU4L4za/pNglWPw8xP3d2wcWNllYyT8NZbQ49FvGgxX9BKGUn5L4ZhFJIimcXOBZa5z8uA8wLtt6rHw8DBInIYcAawUlV3qOrLwErgTHduvKo+rF7J51sDz6oNYVsVh5mf6rViUIU77oDXXx9+bqTlvxiGUUjyEhcFfiYiq0XEbfTOJFV9wX1+EZjkPk8Gng/c2+fa4tr7QtqHISLzRaRXRHq3bdtW+beJiwgLUs8VQ3//8KrI7e2NXV/MMIyGIS9xeZ+qnohn8rpMRE4NnnQrjppvNKOqS1W1S1W7Jk6cWPmD4iLCgqRdMYwdW9l4ohg3zoTFMIy6kIu4qOoW9/4ScBeez2SrM2nh3l9yl28BDg/cPsW1xbVPCWmvHWERYWPGeGapoIO/uztd1v2bb2Y6THPkG4ZRL+ouLiJyoIgc5H8GTgceB+4B/IivecDd7vM9wMUuamwW8Kozn90HnC4ihzhH/unAfe7cThGZ5aLELg48qzYEI8JEPAER8cxSpQ7+xYuHC1FbW3QZmLQ0N0cLmDnyDcOoE3msXCYBD4rI74BfA/9XVe8FvgycJiLrgA+6Y4AVwHPAeuAGYAGAqu4Avgj8xr2ucW24a77r7nkW+GnNv5UfEbZvn2d+Ki2z7zv4S4Woo8OL7CpXBiYJbW1e2ZcoATNHvmEYdUK0FqGxDUhXV5f29vZm87CmpvCQ49L963t6PMHJKqFy+fJBn4r/7M2bvRXLokXmbzEMI3NEZHUgpWQ/RQpFHjnEOfj9ZEsRuOii7ISludl7XtC/46+kNm40YTEMo66YuNSCuJIvQUHJctU4MGD1wwzDKAwmLrUgruRLVoLi+2zC6opZ/TDDMHLGxKVW1Lrky9y5MGGC1Q8zDKOQhFQ8NDKlVpP8kiXx5y3s2DCMHLGVS63JY5IXid5e2TAMow6YuNQKPyps06bBve3jaGryQomjKhmnQdW2OTYMI1dMXGpBsAQ/eJN9OYHZt8+7Z+5caM1gW1xz6huGkSMmLrUgrAR/Eqf+rl1eqfysAgDMqW8YRk6YuGRJ0BRWKf39sGdPNuMxp75hGDlh4pIVpaawMNrbvWrJWeGHOS9fbrXEDMMoFCYuWRFmCgvS1uYVlLz55mQO/lJK7wlGhIUVw7RNwQzDyBErXOmounBlVLFK8Cb7YOHIuGvDEIHZs+H++4fe19ZmImIYRq5Y4cpaE+Xf6OgYXjgyrS9EFdavHy5IFhFmGEZBMXHJirBilb7fw3f0NzXBoYfC9u3pnt3RER35ZRFhhmEUEBOXrIjye8Cgo1/ViwZLszFYa6snUHFl/A3DMAqG1RbLku7u4f6Pzs54R38c7e1eEID/zPnzhz7LIsIMwygoJi61plKzlchQ85kvMLa7pGEYDYCZxWpNpWarsPvidpcM+nX83SgNwzBywsSl1oQ5+oO0tg5PrExr7gomcNpulIZhFIDcxEVEmkXktyLyE3c8TUQeEZH1IvIDEWl17Qe44/XufGfgGZ9z7c+IyBmB9jNd23oRWVj3Lxek1NHf3u69fKf/TTd5iZXVJECGJXBamLJhGDmS58rlCuCpwPFXgG+o6hHAy8AnXfsngZdd+zfcdYjIO4GPA8cCZwLXO8FqBr4FnAW8E7jAXZsfQXPW9u2ek37qVM934gtAlLkrCRambBhGwchFXERkCnA28F13LMBs4E53yTLgPPf5XHeMOz/HXX8ucLuqvqWqG4D1wEnutV5Vn1PV3cDt7tpiUAsTloUpG4ZRMPJauXwT+Aywzx23A6+o6l533AdMdp8nA88DuPOvuuv3t5fcE9U+DBGZLyK9ItK7bdu2Kr9SQmphwopL4DQMw8iBuouLiJwDvKSqq+vddymqulRVu1S1a+LEidk9OC5yqxYmLCtcaRhGwcgjz+UU4EMiMhcYC4wHFgMHi0iLW51MAba467cAhwN9ItICvB3oD7T7BO+Jaq89vtnLX534Zi/wJvupU8PL8ldrwgpL4DQMw8iJuq9cVPVzqjpFVTvxHPL3q2o38AvgfHfZPOBu9/ked4w7f796pZzvAT7uosmmAdOBXwO/Aaa76LNW18c9NfkyYSuUcmavRYuGhx6PGWMmLMMwRhRFytD/LHC7iPwT8FvgRtd+I3CbiKwHduCJBar6hIjcATwJ7AUuU9UBABG5HLgPaAZuUtUnMh9t1AolqtRL0OwVtjeLYRjGCML2c3Gk3s8lajvj5mYYGBje7pfej7rPP28YhtFA2H4uWRPlgB8YiI/cspwUwzBGASYulRK3OVhc5JblpBiGMQowcamUuNySuAKTlpNiGMYowMSlUirNLbGcFMMwRgHm0HekdugbhmEY5tA3DMMw6oeJi2EYhpE5Ji6GYRhG5pi4GIZhGJlj4mIYhmFkjkWLOURkGxBSlyVXDgW25z2IlDTamBttvGBjrgeNNl7Ib8wdqjpszxITlwIjIr1hIX5FptHG3GjjBRtzPWi08ULxxmxmMcMwDCNzTFwMwzCMzDFxKTZL8x5ABTTamBttvGBjrgeNNl4o2JjN52IYhmFkjq1cDMMwjMwxcTEMwzAyx8SlIIjI4SLyCxF5UkSeEJErXPsXRGSLiDzqXnPzHiuAiIwVkV+LyO/ceP/RtU8TkUdEZL2I/EBEWvMeq0/MmG8RkQ2B3/j4nIc6BBFpFpHfishP3HFhf2OfkDEX/TfeKCKPubH1urYJIrJSRNa590PyHqdPxHgLNVeYuBSHvcD/VNV3ArOAy0Tkne7cN1T1ePdakd8Qh/AWMFtV3wUcD5wpIrOAr+CN9wjgZeCT+Q1xGFFjBrgy8Bs/mtcAI7gCeCpwXOTf2Kd0zFDs3xjgL9zY/FyRhcAqVZ0OrHLHRaJ0vFCgucLEpSCo6guqusZ9fg3vH+bkfEcVjXq87g7HuJcCs4E7Xfsy4Lz6jy6cmDEXFhGZApwNfNcdCwX+jWH4mBuYc/F+Xyjg71x0TFwKiIh0AicAj7imy0VkrYjcVLClebOIPAq8BKwEngVeUdW97pI+CiaQpWNWVf83XuR+42+IyAH5jXAY3wQ+A+xzx+0U/Ddm+Jh9ivobg/dHxs9EZLWIzHdtk1T1Bff5RWBSPkMLJWy8UKC5wsSlYIjIOOBHwN+p6k5gCfCneGacF4B/yW90Q1HVAVU9HpgCnAQcne+IylM6ZhE5Dvgc3tjfDUwAPpvfCAcRkXOAl1R1dd5jSUrMmAv5Gwd4n6qeCJyFZ5I+NXhSvZyNIq1yw8ZbqLnCxKVAiMgYPGHpUdUfA6jqVjch7gNuwJvEC4WqvgL8AjgZOFhEWtypKcCWvMYVR2DMZzqTpKrqW8DNFOc3PgX4kIhsBG7HM4ctpti/8bAxi8jyAv/GAKjqFvf+EnAX3vi2ishhAO79pfxGOJSw8RZtrjBxKQjOln4j8JSq/mug/bDAZR8GHq/32MIQkYkicrD7/DbgNDw/0S+A891l84C7cxlgCBFjfjowgQieXb0Qv7Gqfk5Vp6hqJ/Bx4H5V7abAv3HEmC8s6m8MICIHishB/mfgdLzx3YP3+0KBfueo8RZtrmgpf4lRJ04BLgIecz4BgH8ALnBhmwpsBP4mj8GFcBiwTESa8f5IuUNVfyIiTwK3i8g/Ab/FE8yiEDXm+0VkIiDAo8AlOY4xCZ+luL9xFD0F/o0nAXd5ukcL8D1VvVdEfgPcISKfxNuO4y9zHGOQqPHeVqS5wsq/GIZhGJljZjHDMAwjc0xcDMMwjMwxcTEMwzAyx8TFMAzDyBwTF8MwDCNzTFwMowCIyHkioiJS+CoHhpEEExfDKAYXAA+6d8NoeExcDCNnXD259+GVzv+4a2sSketF5Gm3l8gKETnfnZspIg+4ooX3lWRmG0YhMHExjPw5F7hXVX8P9IvITOAjQCfwTrzKDSfD/vpz/xs4X1VnAjcBi/IYtGHEYeVfDCN/LsArSAlesccL8P5t/tAVIXxRRH7hzh8FHAesdOU/mvEq4BpGoTBxMYwcEZEJeNWO/0xEFE8sFK/SbegtwBOqenKdhmgYFWFmMcPIl/OB21S1Q1U7VfVwYAOwA/io871MAj7grn8GmCgi+81kInJsHgM3jDhMXAwjXy5g+CrlR8Af4+0y+SSwHFgDvKqqu/EE6Ssi8ju8CsPvrdtoDSMhVhXZMAqKiIxT1ddFpB34NXCKqr6Y97gMIwnmczGM4vITt7lZK/BFExajkbCVi2EYhpE55nMxDMMwMsfExTAMw8gcExfDMAwjc0xcDMMwjMwxcTEMwzAy5/8DrtZ0jXKlhxYAAAAASUVORK5CYII=\n", "text/plain": [ "
" ] }, "metadata": { "needs_background": "light" }, "output_type": "display_data" } ], "source": [ "# Plot the sample data\n", "group_1_colour, group_2_colour, group_3_colour ='red','blue', 'pink'\n", "plt.xlabel('Age',fontsize=10)\n", "plt.ylabel(\"Salary\",fontsize=10) \n", "\n", "plt.scatter(group_1_ages,group_1_salaries,c=group_1_colour,label=\"Group 1\")\n", "plt.scatter(group_2_ages,group_2_salaries,c=group_2_colour,label=\"Group 2\")\n", "plt.scatter(group_3_ages,group_3_salaries,c=group_3_colour,label=\"Group 3\")\n", "\n", "plt.legend(loc='upper left')\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 32, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+---+---------------+------------------+----------+\n", "| id| SAL_name| SAL_age|SAL_salary|\n", "+---+---------------+------------------+----------+\n", "| 1|Individual: 001| 25.39547052370498| 48976|\n", "| 2|Individual: 002|24.314035458986748| 47402|\n", "| 3|Individual: 003|26.918958635987888| 59828|\n", "| 4|Individual: 004| 25.29664106310324| 50464|\n", "| 5|Individual: 005|26.419729731447458| 53845|\n", "+---+---------------+------------------+----------+\n", "only showing top 5 rows\n", "\n" ] } ], "source": [ "# Turn the above records into a Data Frame\n", "# First of all, create an array of arrays\n", "inputBuf = []\n", "\n", "for i in range(0, len(ages)) :\n", " id = i + 1 # Avoid counting from zero\n", " name = \"Individual: {:03d}\".format(id)\n", " # Note we need to make sure values are typed correctly\n", " # salary will have type numpy.float64 - if it is not cast as below, an error will be thrown\n", " age = float(ages[i])\n", " salary = int(salaries[i])\n", " inputBuf.append((id, name,age,salary))\n", "\n", "# Convert to an RDD \n", "inputRDD = spark.sparkContext.parallelize(inputBuf)\n", " \n", "# Convert to a data frame using a schema\n", "# Note the feature group SAL is prefixed to each feature column\n", "schema = StructType([\n", " StructField(\"id\", IntegerType(), True),\n", " StructField(\"SAL_name\", StringType(), True),\n", " StructField(\"SAL_age\", DoubleType(), True),\n", " StructField(\"SAL_salary\",IntegerType(), True)\n", "])\n", "\n", "inputDF=spark.createDataFrame(inputRDD,schema)\n", "inputDF.show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Save Features to Feature Store\n", "1. Create a feature group.\n", "2. Create feature metadata.\n", "3. Save feature values in entity records." ] }, { "cell_type": "code", "execution_count": 33, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Features stored to Feature Store.\n" ] } ], "source": [ "# 1. Create a feature group.\n", "FG_NAME = 'SAL'\n", "FG_DESCRIPTION = 'Age salary data'\n", "FG_SOURCE = 'Generated demo data'\n", "fg = FeatureGroup(FG_NAME, FG_DESCRIPTION, FG_SOURCE, \n", " attrs={'access':'all'}, tags=['test'])\n", "fg.save()\n", "\n", "# 2. Create features metadata.\n", "FEATURE_NAME = 'name'\n", "f = Feature(FG_NAME, FEATURE_NAME, \"string\", \"Name of the person\", \n", " attrs={'unique':'no'}, tags=['test'])\n", "f.save()\n", "\n", "FEATURE_AGE = 'age'\n", "f = Feature(FG_NAME, FEATURE_AGE, \"double\", \"Age of the person\", \n", " attrs={'range':'0-100'}, tags=['test'])\n", "f.save()\n", "\n", "FEATURE_SALARY = 'salary'\n", "f = Feature(FG_NAME, FEATURE_SALARY, 'integer', \"Salary of the person\", \n", " attrs={'range':'20-999K'}, tags=['test'])\n", "f.save()\n", "\n", "# 3. Save feature values in entity records.\n", "ENTITY_TYPE = 'user'\n", "ID_COLUMN = 'id'\n", "Entity.saveDF(inputDF, ENTITY_TYPE, ID_COLUMN)\n", "print('Features stored to Feature Store.')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Query Features \n", "Issue a query against the saved features to get a dataframe with records in age group 25 to 30. There are 50 such users in the dataset." ] }, { "cell_type": "code", "execution_count": 34, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Number of users of age between 25 and 30: 50\n", "+---+---------------+------------------+----------+\n", "| id| SAL_name| SAL_age|SAL_salary|\n", "+---+---------------+------------------+----------+\n", "| 85|Individual: 085|26.288287033377657| 59603|\n", "| 1|Individual: 001| 25.39547052370498| 48976|\n", "| 14|Individual: 014| 25.59043077849547| 51513|\n", "| 79|Individual: 079|25.887490702675926| 48162|\n", "| 4|Individual: 004| 25.29664106310324| 50464|\n", "+---+---------------+------------------+----------+\n", "only showing top 5 rows\n", "\n" ] } ], "source": [ "queryDF = Entity.query(ENTITY_TYPE, '`SAL_age` between 25 and 30', schema, ID_COLUMN)\n", "print(\"Number of users of age between 25 and 30: {}\".format(queryDF.count()))\n", "queryDF.show(5)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Examine the data through aql with the following command:" ] }, { "cell_type": "code", "execution_count": 35, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "select * from test.user-features where PK=85\r\n", "+----+----+-------------------+-------------------+------------+\r\n", "| PK | id | SAL_age | SAL_name | SAL_salary |\r\n", "+----+----+-------------------+-------------------+------------+\r\n", "| 85 | 85 | 26.28828703337766 | \"Individual: 085\" | 59603 |\r\n", "+----+----+-------------------+-------------------+------------+\r\n", "1 row in set (0.001 secs)\r\n", "\r\n", "OK\r\n", "\r\n", "\r\n" ] } ], "source": [ "!aql -c \"select * from test.user-features where PK=85\"" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Takeaways and Conclusion\n", "In this notebook, we explored how Aerospike can be used as a Feature Store for ML applications. Specifically, we showed how features engineered using the Spark platform can be efficiently stored in Aerospike feature store via the Aerospike Spark Connector. We implemented a simple example feature store interface that leverages the Aerospike Spark connector capabilities for this purpose. We used the API to save and query features created in two data engineering examples.\n", "\n", "This is a first in the series of notebooks on how Aerospike can be used as a feature store. Subseequent notebooks in this series will explore use of Aerospike Feature Store for Model Training and Model Serving." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Cleaning Up\n", "Close the spark session, and remove the tutorial data by executing the cell below." ] }, { "cell_type": "code", "execution_count": 36, "metadata": {}, "outputs": [], "source": [ "try:\n", " spark.stop()\n", "except: \n", " ; # ignore\n", "# To remove all data in the namespace test, uncomment the following line and run: \n", "#!aql -c \"truncate test\"" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Further Exploration and Resources\n", "Here are some links for further exploration.\n", "\n", "## Resources\n", "- Related notebooks\n", " - [Model Training with Aerospike Feature Store (Part 2)](feature-store-model-training.ipynb)\n", " - [Model Serving with Aerospike Feature Store (Part 3)](feature-store-model-serving.ipynb)\n", " - [Aerospike Connect for Spark Tutorial for Python](AerospikeSparkPython.ipynb)\n", " - [Pushdown Expressions for Spark Connector](resources/pushdown-expressions.ipynb)\n", "- Related blog posts\n", " - [Let AI/ML workloads take off with Aerospike and Spark 3.0](https://medium.com/aerospike-developer-blog/let-ai-ml-workloads-take-off-with-aerospike-and-spark-3-0-82de2d834b99)\n", " - [Using Aerospike Connect For Spark](https://medium.com/aerospike-developer-blog/aerospike-is-a-highly-scalable-key-value-database-offering-best-in-class-performance-5922450aaa78)\n", "- Aerospike Developer Hub\n", " - [Developer Hub](https://developer.aerospike.com/)\n", "- Github repos\n", " - [Spark Aerospike Example](https://github.com/aerospike-examples/spark-aerospike-example)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Exploring Other Notebooks\n", "\n", "Visit [Aerospike notebooks repo](https://github.com/aerospike-examples/interactive-notebooks) to run additional Aerospike notebooks. To run a different notebook, download the notebook from the repo to your local machine, and then click on File->Open in the notebook menu, and select Upload." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "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.8.6" }, "toc": { "base_numbering": 1, "nav_menu": {}, "number_sections": true, "sideBar": false, "skip_h1_title": false, "title_cell": "Table of Contents", "title_sidebar": "Contents", "toc_cell": true, "toc_position": {}, "toc_section_display": false, "toc_window_display": false } }, "nbformat": 4, "nbformat_minor": 2 }