<jupyter_start><jupyter_text>Feature StoreWhen the data grows fast, performing ETL pipelines for multiple machine learning projects becomes expensive since repetitive operations. A feature store is a solution to this problem. It's possible to reuse the features in different projects and don't need to repeat similar processes in other projects.This tutorial will cover how to create a feature store for Starbucks transactions. We will build ETLs pipelines using Butterfree library to upload data to a Feature Store so that data can be provided for machine learning algorithms, even for training or for prediction.  Example:Simulating the following scenario:- We have a streaming JSON data source with events of starbucks orders being captured in real time.- We have a csv data set with more information about drinks.Objective: We want to parse the JSON from the streaming source, performing aggregations operations, and store all rows in a cheap structure(like s3) and get more recent transactions on a low latency database like Cassandra.We desire to have an output with the schema:- **id_employer**: int- **name_employer**: string- **name_client**: string- **payment**: string- **timestamp**: timestamp- **product_name**: timestamp- **product_size**: string- **product_price**: int- **percent_carbo**: float- **final_price**: floatThe following code blocks will show how to generate this feature set using Butterfree library using the above architecture:- Apache Kafka as data sources (Streaming input data);- A hive metastore to store metadata (like their schema and location) in a relational database.(For this tutorial we will use Postgresql)- Apache Cassandra to store more recent data.- Amazon S3 to store historical features or table views for debug mode.Historical Feature Store: all features calculated over time;Online Feature Store: hot/latest(last record by key) data stored at a low latency data storage(Cassandra).  ObservationsIn this tutorial, the historical data will be stored locally. However, you can easily add an s3 bucket.            Check the documentation here https://butterfree.readthedocs.io/en/latest/configuration.html?highlight=s3historical-feature-store-spark-metastore-and-s3  We will do a batch process, but you can switch to online processing with minor modifications         Check the documentation here https://butterfree.readthedocs.io/en/latest/stream.html   Download packages<jupyter_code>import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--master local[6] --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0,com.datastax.spark:spark-cassandra-connector_2.12:3.0.0-beta pyspark-shell'<jupyter_output><empty_output><jupyter_text>Spark InstanceConnecting to hive metastore<jupyter_code># setup spark
from pyspark import SparkContext, SparkConf
from pyspark.sql import session, SparkSession
from pyspark.sql import HiveContext
# butterfree spark client
from butterfree.clients import SparkClient


spark = (
    SparkSession
    .builder
    .appName("Feature Store")
    .config("hive.metastore.uris", "thrift://hive-metastore:9083")
    .enableHiveSupport()
    .getOrCreate())

sc=spark.sparkContext

spark_client = SparkClient()
hive_context = HiveContext(sc)<jupyter_output><empty_output><jupyter_text>ExtractFirst, we need to define our data schemas.<jupyter_code>from pyspark.sql.types import StringType, IntegerType, StructType, StructField, DoubleType

schema_kafka = StructType([StructField('name_employer', StringType(), True),
                          StructField('id_employer', IntegerType(), True),
                          StructField('name_client', StringType(), True),
                          StructField('transaction_id', IntegerType(), True),
                          StructField('payment', StringType(), True),
                          StructField('timestamp', StringType(), True),
                          StructField('product_name', StringType(), True),
                          StructField('product_size', StringType(), True),
                          StructField('product_price', DoubleType(), True),
                          StructField('percent_discount', IntegerType(), True)])


schema_file = StructType([StructField('name', StringType(), True),
                         StructField('calories', IntegerType(), True),
                         StructField('fat(g)', IntegerType(), True),
                         StructField('carb(g)', IntegerType(), True),
                         StructField('fiber(g)', IntegerType(), True),
                         StructField('protein', IntegerType(), True),
                         StructField('sodium', IntegerType(), True)])<jupyter_output><empty_output><jupyter_text>Connecting with cassandra database<jupyter_code>from butterfree.extract import Source
from butterfree.extract.readers import FileReader
from butterfree.extract.readers import KafkaReader

kafka_reader = KafkaReader(
    id="events",
    topic="queueing.transactions",
    value_schema=schema_kafka,
    connection_string="broker:9092",
    stream=False
)

readers = [
    kafka_reader,
    FileReader(id="nutrients", path="starbucks-menu-nutrition-drinks.csv", format="csv", schema=schema_file)
]

query = """
select
    *
from
    events
    join nutrients
        on events.product_name = nutrients.name
"""

source = Source(readers=readers, query=query)
source_df = source.construct(spark_client)
# showing that it is a Spark's streaming df
source_df.isStreaming
# schema
source_df.printSchema()<jupyter_output>root
 |-- name_employer: string (nullable = true)
 |-- id_employer: integer (nullable = true)
 |-- name_client: string (nullable = true)
 |-- transaction_id: integer (nullable = true)
 |-- payment: string (nullable = true)
 |-- timestamp: string (nullable = true)
 |-- product_name: string (nullable = true)
 |-- product_size: string (nullable = true)
 |-- product_price: double (nullable = true)
 |-- percent_discount: integer (nullable = true)
 |-- kafka_metadata: struct (nullable = false)
 |    |-- key: string (nullable = true)
 |    |-- topic: string (nullable = true)
 |    |-- value: string (nullable = true)
 |    |-- partition: integer (nullable = true)
 |    |-- offset: long (nullable = true)
 |    |-- timestamp: timestamp (nullable = true)
 |    |-- timestampType: integer (nullable = true)
 |-- name: string (nullable = true)
 |-- calories: integer (nullable = true)
 |-- fat(g): integer (nullable = true)
 |-- carb(g): integer (nullable = true)
 |-- fiber(g): integer (nullable = true[...]<jupyter_text>Transform- At the transform part, a set of `Feature` objects is declared.- An Instance of `FeatureSet` is used to hold the features.- A `FeatureSet` can only be created when it is possible to define a unique tuple formed by key columns and a time reference. This is an **architectural requirement** for the data. So least one `KeyFeature` and one `TimestampFeature` is needed.- Every `Feature` needs a unique name, a description, and a data-type definition.  Transform<jupyter_code>from pyspark.sql import functions as F

from butterfree.transform import FeatureSet
from butterfree.transform.features import Feature, KeyFeature, TimestampFeature
from butterfree.transform.transformations import SQLExpressionTransform, SparkFunctionTransform, CustomTransform
from butterfree.transform.transformations.h3_transform import H3HashTransform
from butterfree.constants import DataType
from butterfree.transform.utils import Function


def divide(df, parent_feature, column1, column2):
    name = parent_feature.get_output_columns()[0]
    df = df.withColumn(name, F.col(column1) / F.col(column2))
    return df


keys = [
    KeyFeature(
        name="id_employer",
        description="Unique identificator code for employer.",
        from_column="id_employer",
        dtype=DataType.INTEGER,
    )
]

# from_ms = True because the data originally is not in a Timestamp format.
ts_feature = TimestampFeature(from_column="timestamp")

features = [
    Feature(
        name="name_employer",
        description="name_employer",
        dtype=DataType.STRING,
    ),
    Feature(
        name="name_client",
        description="name_client",
        dtype=DataType.STRING,
    ),
    Feature(
        name="product_name",
        description="product_name.",
        dtype=DataType.STRING,
    ),
    Feature(
        name="product_price",
        description="product_price.",
        dtype=DataType.FLOAT,
    ),
    Feature(
        name="payment",
        description="payment.",
        dtype=DataType.STRING,
    ),
    Feature(
        name="calories",
        description="calories",
        dtype=DataType.INTEGER,
    ),
    # custom transformation
    Feature(
           name="percent_carbo",
           description="percent_carbo",
           transformation=CustomTransform(transformer=divide, column1="carb(g)", column2="calories"), 
           dtype=DataType.FLOAT,
    ),
    # SQL transformation
    Feature(
           name="final_price",
           description="percent_carbo",
           transformation=SQLExpressionTransform("product_price * ((100 - percent_discount)/100)"), 
           dtype=DataType.FLOAT,
    ),
]

# events will be sotred in our metasotore as a table. You can acess starbucks_order_events 
feature_set = FeatureSet(
    name="starbucks_order_events",
    entity="events",  # entity: to which "business context" this feature set belongs
    description="Features describring events about starbucks store.",
    keys=keys,
    timestamp=ts_feature,
    features=features,
)
feature_set_df = feature_set.construct(source_df, spark_client)
# schema
feature_set_df.printSchema()<jupyter_output>root
 |-- id_employer: integer (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- name_employer: string (nullable = true)
 |-- name_client: string (nullable = true)
 |-- product_name: string (nullable = true)
 |-- product_price: float (nullable = true)
 |-- payment: string (nullable = true)
 |-- calories: integer (nullable = true)
 |-- percent_carbo: double (nullable = true)
 |-- final_price: double (nullable = true)<jupyter_text>Load- Using debug mode to create a temporary view with the historical data<jupyter_code>from butterfree.load.writers import (
    HistoricalFeatureStoreWriter,
    OnlineFeatureStoreWriter,
)
from butterfree.load import Sink

from butterfree.configs.db import CassandraConfig
from butterfree.load.writers import OnlineFeatureStoreWriter

db_config  = CassandraConfig(
    username="cassandra", 
    password="<PASSWORD>",
    host="feature_store_cassandra",
    keyspace="feature_store",
    stream_checkpoint_path="./"
)

writers = [HistoricalFeatureStoreWriter(debug_mode=True),OnlineFeatureStoreWriter(db_config=db_config)]
sink = Sink(writers=writers)<jupyter_output><empty_output><jupyter_text>Cassandra tables- Lets create a keyspace and a table to store the online features<jupyter_code>from cassandra.cluster import Cluster

keyspace = "feature_store"
table_name = "starbucks_order_events"

cassandra_mapping = {
        "TimestampType": "timestamp",
        "BinaryType": "boolean",
        "BooleanType": "boolean",
        "DateType": "timestamp",
        "DecimalType": "decimal",
        "DoubleType": "double",
        "FloatType": "float",
        "IntegerType": "int",
        "LongType": "bigint",
        "StringType": "text",
        "ArrayType(LongType,true)": "frozen<list<bigint>>",
        "ArrayType(StringType,true)": "frozen<list<text>>",
        "ArrayType(FloatType,true)": "frozen<list<float>>",
    }

cluster = Cluster(['feature_store_cassandra'])
session = cluster.connect()

session.execute("CREATE KEYSPACE IF NOT EXISTS "+ keyspace +" WITH REPLICATION = { 'class' : 'NetworkTopologyStrategy', 'datacenter1' : 1 };")

sql = ", ".join([feature.name +str(" ") + cassandra_mapping[str(feature.dataType)] for feature in feature_set_df.schema]).replace("id_employer int", "id_employer int PRIMARY KEY")
sql = "CREATE TABLE IF NOT EXISTS {}.{} (" + sql + ");"
sql = sql.format(keyspace, table_name)
session.execute(sql)
cluster.shutdown()<jupyter_output><empty_output><jupyter_text>Final Pipeline<jupyter_code>from butterfree.pipelines import FeatureSetPipeline

pipeline = FeatureSetPipeline(source=source, feature_set=feature_set, sink=sink)
# asinc run when creating an in memory streaming view for sink 
pipeline.run()<jupyter_output><empty_output><jupyter_text>Showing the results  Online features for cassandra<jupyter_code>print(">>> Online Feature Store  table:")

from cassandra.cluster import Cluster
cluster = Cluster(['feature_store_cassandra'])
session = cluster.connect()
df = session.execute("SELECT * FROM feature_store.starbucks_order_events")
cluster.shutdown()
# Create data frame
df = spark.createDataFrame(df)
df.toPandas()<jupyter_output>>>> Online Feature Store  table:<jupyter_text>Acessging metastore  Historical features<jupyter_code>print(">>> Historical Feature Store:")
spark.table("historical_feature_store__starbucks_order_events").toPandas()<jupyter_output>>>> Historical Feature Store:<jupyter_text>MetaStoreIf you not specify a s3 bucket, then historical data will be store into metastore<jupyter_code>hive_context.sql("select * from historical_feature_store__starbucks_order_events;").show()<jupyter_output>+-----------+-------------------+-------------+-----------------+--------------------+-------------+-------+--------+-------------------+------------------+----+-----+---+
|id_employer|          timestamp|name_employer|      name_client|        product_name|product_price|payment|calories|      percent_carbo|       final_price|year|month|day|
+-----------+-------------------+-------------+-----------------+--------------------+-------------+-------+--------+-------------------+------------------+----+-----+---+
|          1|2023-05-18 01:04:05|         Alex|     <NAME>|<NAME>|         3.65|   cash|     260|0.15384615384615385|2.9200000762939453|2023|    5| 18|
|          1|2021-05-14 17:04:05|         Alex| <NAME>|Cinnamon Dolce Latte|         4.25|   cash|     260|0.15384615384615385|3.4000000000000004|2021|    5| 14|
|          1|2021-07-24 03:04:05|         Alex|    <NAME>|White Chocolate M...|         4.45| credit|     360|0.14722222222222223|3.1[...]
{'fraction_non_alphanumeric': 0.1300561797752809, 'fraction_numerical': 0.015028089887640449, 'mean_word_length': 3.6267056530214425, 'pattern_counts': {'":': 13, '<': 51, '<?xml version=': 0, '>': 63, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '199600', 'license': 'MIT', 'n_tokens_mistral': 4130, 'n_tokens_neox': 3953, 'n_words': 1085, 'path': 'spark/notebook/Starbucks_ETL.ipynb', 'repo_name': 'rpatil524/Feature-Store', 'size': '14276'}