Complete Feast Tutorial: Feature Store for Machine Learning

Feast (Feature Store) is an open-source feature store that helps ML teams manage, discover, and serve features for machine learning models. It bridges the gap between data engineering and machine learning by providing a consistent way to define, store, and retrieve features.

Why Feast?

Feast Advantages:

Feature consistency: Same features in training and serving
Feature sharing: Reuse features across teams
Point-in-time correctness: Prevent data leakage
Low latency serving: Online feature retrieval
Feature discovery: Centralized feature catalog

Use Cases:

ML model training and serving
Real-time recommendation systems
Fraud detection
Personalization engines
Feature sharing across teams

Installation

# Basic installation
pip install feast

With specific providers
pip install feast[redis]      # Redis online store
pip install feast[gcp]        # Google Cloud
pip install feast[aws]        # AWS
pip install feast[snowflake]  # Snowflake

Verify installation
feast version

Quick Start

1. Initialize Project

# Create new Feast project feast init myfeaturerepo cd myfeaturerepo Project structure: myfeaturerepo/ ├── featurerepo/ │ ├── init.py │ ├── examplerepo.py │ └── featurestore.yaml └── data/ └── driverstats.parquet

2. Feature Store Configuration

# featurestore.yaml project: myproject provider: local registry: data/registry.db onlinestore: type: sqlite path: data/onlinestore.db offlinestore: type: file entitykeyserializationversion: 2

3. Define Features

# featurerepo/features.py
from datetime import timedelta
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from feast.types import Float32, Int64, String

Define data source
driverstatssource = FileSource(

    name="driverstatssource",

    path="data/driverstats.parquet",
    timestampfield="eventtimestamp",
    createdtimestampcolumn="created",
)

Define entity
driver = Entity(
    name="driverid",

    valuetype=ValueType.INT64,
    description="Driver identifier",
)

Define feature view
driverstatsfv = FeatureView(
    name="driverstats",

    entities=[driver],
    ttl=timedelta(days=1),
    schema=[
        Feature(name="convrate", dtype=Float32),
        Feature(name="accrate", dtype=Float32),

        Feature(name="avgdailytrips", dtype=Int64),

    ],
    source=driverstatssource,

    online=True,
    tags={"team": "driverperformance"},
)

4. Apply and Materialize

# Apply feature definitions feast apply Materialize features to online store feast materialize-incremental $(date +%Y-%m-%dT%H:%M:%S) Or materialize specific date range feast materialize 2024-01-01T00:00:00 2024-01-31T00:00:00

5. Retrieve Features

from feast import FeatureStore
import pandas as pd

Initialize feature store
store = FeatureStore(repopath=".")


Get historical features (for training)
entitydf = pd.DataFrame({
    "driverid": [1001, 1002, 1003],

    "eventtimestamp": pd.todatetime([

        "2024-01-15 10:00:00",
        "2024-01-15 10:00:00",
        "2024-01-15 10:00:00",
    ])
})

trainingdf = store.gethistoricalfeatures(
    entitydf=entitydf,
    features=[
        "driverstats:convrate",
        "driverstats:accrate",
        "driverstats:avgdailytrips",

    ],
).todf()

print(trainingdf)


Get online features (for serving)
onlinefeatures = store.getonlinefeatures(
    features=[
        "driverstats:convrate",
        "driverstats:accrate",
    ],
    entityrows=[

        {"driverid": 1001},
        {"driverid": 1002},

    ],
).todict()

print(onlinefeatures)

Feature Definitions

1. Entities

from feast import Entity, ValueType

Simple entity
user = Entity(
    name="userid",
    valuetype=ValueType.INT64,

    description="User identifier",
    joinkeys=["userid"],  # Optional, defaults to name

)

Composite entity
order = Entity(
    name="order",
    valuetype=ValueType.STRING,
    description="Order identifier",
    joinkeys=["orderid"],
)

Multiple join keys
transaction = Entity(
    name="transaction",
    valuetype=ValueType.STRING,

    description="Transaction with user and merchant",
    joinkeys=["userid", "merchantid"],
)

2. Data Sources

from feast import FileSource, BigQuerySource, RedshiftSource, SnowflakeSource
from feast.infra.offlinestores.contrib.postgresofflinestore.postgressource import (
    PostgreSQLSource
)

File source (Parquet)
filesource = FileSource(

    name="userfeaturessource",

    path="s3://bucket/userfeatures.parquet",
    timestampfield="eventtimestamp",
    createdtimestampcolumn="created",
)

BigQuery source
bqsource = BigQuerySource(

    name="userfeaturesbq",

    table="project.dataset.userfeatures",
    timestampfield="eventtimestamp",
)

Redshift source
redshiftsource = RedshiftSource(

    name="userfeaturesredshift",

    query="SELECT  FROM userfeatures",
    timestampfield="eventtimestamp",
    createdtimestampcolumn="created",
)

PostgreSQL source
postgressource = PostgreSQLSource(

    name="userfeaturespg",

    query="SELECT  FROM userfeatures",
    timestampfield="eventtimestamp",
)

3. Feature Views

from datetime import timedelta
from feast import FeatureView, Feature
from feast.types import Float32, Int64, String, UnixTimestamp

Basic feature view
userstats = FeatureView(

    name="userstats",
    entities=[user],
    ttl=timedelta(days=7),
    schema=[
        Feature(name="totalpurchases", dtype=Int64),

        Feature(name="avgpurchaseamount", dtype=Float32),

        Feature(name="lastpurchasedate", dtype=UnixTimestamp),

    ],
    source=filesource,
    online=True,
    tags={"team": "useranalytics"},

)

Feature view with description
productfeatures = FeatureView(
    name="productfeatures",

    description="Product-level features for recommendations",
    entities=[product],
    ttl=timedelta(days=1),
    schema=[
        Feature(name="category", dtype=String, description="Product category"),
        Feature(name="price", dtype=Float32, description="Current price"),
        Feature(name="popularityscore", dtype=Float32, description="Popularity metric"),
    ],
    source=productsource,

    online=True,
)

4. On-Demand Feature Views

from feast import ondemandfeatureview, Field
from feast.types import Float64
import pandas as pd

On-demand transformation
@ondemandfeatureview(

    sources=[userstats],
    schema=[
        Field(name="purchasefrequency", dtype=Float64),

    ],
)
def userpurchasefrequency(inputs: pd.DataFrame) -> pd.DataFrame:

    df = pd.DataFrame()
    df["purchasefrequency"] = inputs["totalpurchases"] / 30  # per day

    return df

Multiple sources
@ondemandfeatureview(
    sources=[userstats, productfeatures],
    schema=[
        Field(name="userproductscore", dtype=Float64),
    ],
)
def userproductaffinity(inputs: pd.DataFrame) -> pd.DataFrame:
    df = pd.DataFrame()
    df["userproductscore"] = (
        inputs["avgpurchaseamount"]  inputs["popularityscore"]

    )
    return df

5. Feature Services

from feast import FeatureService Define feature servicedriverfeatureservice = FeatureService( name="driverfeaturesv1", features=[ driverstatsfv, driverratingsfv[["ratingavg"]], # Select specific features ], tags={"release": "v1"}, ) Feature service for recommendation recommendationservice = FeatureService( name="recommendationfeatures", features=[ userstats, productfeatures, userpurchasefrequency, ], )

Offline Store Configurations

1. File-based (Local/S3)

# featurestore.yaml offlinestore: type: file

2. BigQuery

# featurestore.yaml project: myfeastproject provider: gcp offlinestore: type: bigquery project: my-gcp-project location: US

3. Snowflake

# featurestore.yaml offlinestore: type: snowflake.offline account: myaccount user: feastuser password: ${SNOWFLAKEPASSWORD} role: feastrole warehouse: feastwarehouse database: FEASTDB

4. Redshift

# featurestore.yaml offlinestore: type: redshift region: us-west-2 clusterid: my-cluster database: feast user: feastuser s3staginglocation: s3://bucket/staging/ iamrole: arn:aws:iam::123456789:role/feast-role

Online Store Configurations

1. SQLite (Local Development)

# featurestore.yaml onlinestore: type: sqlite path: data/onlinestore.db

2. Redis

# featurestore.yaml onlinestore: type: redis connectionstring: redis://localhost:6379

# Redis Clusteronlinestore: type: redis redistype: rediscluster connectionstring: redis://node1:6379,node2:6379,node3:6379

3. DynamoDB

# featurestore.yaml onlinestore: type: dynamodb region: us-west-2

4. Datastore (GCP)

# featurestore.yaml onlinestore: type: datastore project: my-gcp-project

Training Pipeline Integration

1. Generate Training Dataset

from feast import FeatureStore import pandas as pd from datetime import datetimestore = FeatureStore(repopath=".") Define entity dataframe with labels entitydf = pd.DataFrame({ "driverid": [1001, 1002, 1003, 1004, 1005], "eventtimestamp": [ datetime(2024, 1, 15, 10, 0), datetime(2024, 1, 15, 11, 0), datetime(2024, 1, 15, 12, 0), datetime(2024, 1, 15, 13, 0), datetime(2024, 1, 15, 14, 0), ], "label": [1, 0, 1, 1, 0], # Training labels }) Get training data with point-in-time join trainingdf = store.gethistoricalfeatures( entitydf=entitydf, features=[ "driverstats:convrate", "driverstats:accrate", "driverstats:avgdailytrips", ], ).todf() Split features and labels X = trainingdf.drop(columns=["driverid", "eventtimestamp", "label"]) y = trainingdf["label"] Train model from sklearn.ensemble import RandomForestClassifier model = RandomForestClassifier() model.fit(X, y)

2. Feature Service for Training

# Use feature service for consistent features trainingdf = store.gethistoricalfeatures( entitydf=entitydf, features=store.getfeatureservice("driverfeaturesv1"), ).todf()

3. Save Training Dataset

# Save dataset for reproducibility store.createsaveddataset( from=store.gethistoricalfeatures( entitydf=entitydf, features=["driverstats:convrate", "driverstats:accrate"], ), name="drivertrainingdatasetv1", storage=SavedDatasetFileStorage("data/trainingdataset.parquet"), ) Retrieve saved dataset saveddataset = store.getsaveddataset("drivertrainingdatasetv1") df = saveddataset.todf()

Serving Pipeline Integration

1. Online Feature Retrieval

from feast import FeatureStorestore = FeatureStore(repopath=".") Get features for inference features = store.getonlinefeatures( features=[ "driverstats:convrate", "driverstats:accrate", "driverstats:avgdailytrips", ], entityrows=[{"driverid": 1001}], ).todict() Use with model featurevector = [ features["convrate"][0], features["accrate"][0], features["avgdailytrips"][0], ] prediction = model.predict([featurevector])

2. Feature Server (REST API)

# Start feature server feast serve -h 0.0.0.0 -p 6566

# Client request import requests response = requests.post( "http://localhost:6566/get-online-features", json={ "features": [ "driverstats:convrate", "driverstats:accrate", ], "entities": { "driverid": [1001, 1002] } } ) features = response.json()

3. FastAPI Integration

from fastapi import FastAPI from feast import FeatureStore import joblib app = FastAPI()store = FeatureStore(repopath="featurerepo") model = joblib.load("model.joblib") @app.post("/predict") async def predict(driverid: int): # Get features features = store.getonlinefeatures( features=[ "driverstats:convrate", "driverstats:accrate", "driverstats:avgdailytrips", ], entityrows=[{"driverid": driverid}], ).todict() # Create feature vector featurevector = [[ features["convrate"][0], features["accrate"][0], features["avgdailytrips"][0], ]] # Predict prediction = model.predict(featurevector) return {"driverid": driverid, "prediction": int(prediction[0])}

Streaming Features

1. Push Source

from feast import PushSource, FeatureViewfrom feast.dataformat import JsonFormat Define push source pushsource = PushSource( name="driverstatspush", batchsource=driverstatssource, # For historical retrieval ) Feature view with push source driverstatsrealtime = FeatureView( name="driverstatsrealtime", entities=[driver], ttl=timedelta(hours=1), schema=[ Feature(name="currenttrips", dtype=Int64), Feature(name="currentrating", dtype=Float32), ], source=pushsource, online=True, )

2. Push Features

from feast import FeatureStore import pandas as pdstore = FeatureStore(repopath=".") Push real-time features store.push( pushsourcename="driverstatspush", df=pd.DataFrame({ "driverid": [1001], "currenttrips": [5], "currentrating": [4.8], "eventtimestamp": [datetime.now()], }), to=PushMode.ONLINEANDOFFLINE, )

3. Kafka Integration

from kafka import KafkaConsumer from feast import FeatureStore import json store = FeatureStore(repopath=".") consumer = KafkaConsumer( "driverevents", bootstrapservers=["localhost:9092"], valuedeserializer=lambda m: json.loads(m.decode("utf-8")), ) for message in consumer: event = message.value # Push to Feast store.push( pushsourcename="driverstatspush", df=pd.DataFrame([{ "driverid": event["driverid"], "currenttrips": event["trips"], "eventtimestamp": datetime.now(), }]), to=PushMode.ONLINE, )

CI/CD Integration

1. Feature Validation

# tests/testfeatures.py import pytest from feast import FeatureStore @pytest.fixture def store(): return FeatureStore(repopath="featurerepo") def testfeatureviewsexist(store): """Test all expected feature views exist""" views = store.listfeatureviews() viewnames = [v.name for v in views] assert "driverstats" in viewnames assert "userfeatures" in viewnames def testfeaturesretrievable(store): """Test features can be retrieved""" features = store.getonlinefeatures( features=["driverstats:convrate"], entityrows=[{"driverid": 1001}], ).todict() assert "convrate" in features

2. GitHub Actions

# .github/workflows/feast.yml name: Feast CI/CD on: push: paths:'featurerepo/' jobs: validate: runs-on: ubuntu-latest steps: uses: actions/checkout@v3 name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.10' name: Install dependencies run: | pip install feast pytest name: Validate feature definitions run: | cd featurerepo feast plan name: Run tests run: pytest tests/ deploy: needs: validate runs-on: ubuntu-latest if: github.ref == 'refs/heads/main' steps: uses: actions/checkout@v3 name: Apply features run: | cd featurerepo feast apply name: Materialize features run: | cd featurerepo feast materialize-incremental $(date +%Y-%m-%dT%H:%M:%S)

Best Practices

1. Feature Naming

# Good naming conventionsuserprofilefv = FeatureView( name="userprofile", # snakecase schema=[ Feature(name="totalorders", dtype=Int64), # descriptive Feature(name="avgordervalue30d", dtype=Float32), # include time window Feature(name="ispremiumuser", dtype=Bool), # prefix for booleans ], )

2. Feature Documentation

userfeatures = FeatureView( name="userfeatures", description=""" User-level features aggregated from transaction history. Updated daily at 2 AM UTC. """, entities=[user], schema=[ Feature( name="lifetimevalue", dtype=Float32, description="Total spend by user over all time", ), ], tags={ "owner": "ml-team@company.com", "version": "2.0", "source": "transactionsdb", }, )

3. Feature Versioning

# Use feature services for versioningrecommendationv1 = FeatureService( name="recommendationv1", features=[userfeaturesv1, productfeatures], ) recommendationv2 = FeatureService( name="recommendationv2", features=[userfeaturesv2, productfeatures, newfeatures], )

Conclusion

Feast is essential for production ML systems with:

Feature consistency: Same features in training and serving

Point-in-time correctness: Prevent data leakage

Feature reuse: Share across teams and models

Low latency serving: Online feature retrieval

Centralized catalog*: Feature discovery

Key takeaways:

Define features once, use everywhere
Use feature services for versioning
Materialize regularly for online serving
Integrate with your ML pipeline
Document and tag features for discoverability