Tutorial Lengkap Feast: Feature Store untuk Machine Learning

Feast (Feature Store) adalah feature store open-source yang membantu tim ML mengelola, menemukan, dan menyajikan features untuk model machine learning. Feast menjembatani gap antara data engineering dan machine learning dengan menyediakan cara konsisten untuk mendefinisikan, menyimpan, dan mengambil features.

Mengapa Feast?

Keunggulan Feast:

Feature consistency: Features sama di training dan serving
Feature sharing: Reuse features antar tim
Point-in-time correctness: Cegah data leakage
Low latency serving: Online feature retrieval
Feature discovery: Centralized feature catalog

Use Cases:

ML model training dan serving
Real-time recommendation systems
Fraud detection
Personalization engines
Feature sharing antar tim

Instalasi

# Basic installation
pip install feast

Dengan 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

# Buat Feast project baru 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 dan Materialize

# Apply feature definitions feast apply Materialize features ke online store feast materialize-incremental $(date +%Y-%m-%dT%H:%M:%S) Atau materialize date range tertentu 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 (untuk 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 (untuk 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, default ke 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 dengan user dan 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 dengan description
productfeatures = FeatureView(
    name="productfeatures",

    description="Product-level features untuk 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 untuk recommendation recommendationservice = FeatureService( name="recommendationfeatures", features=[ userstats, productfeatures, userpurchasefrequency, ], )

Konfigurasi Offline Store

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

Konfigurasi Online Store

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

Integrasi Training Pipeline

1. Generate Training Dataset

from feast import FeatureStore import pandas as pd from datetime import datetimestore = FeatureStore(repopath=".") Define entity dataframe dengan 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 dengan point-in-time join trainingdf = store.gethistoricalfeatures( entitydf=entitydf, features=[ "driverstats:convrate", "driverstats:accrate", "driverstats:avgdailytrips", ], ).todf() Split features dan 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 untuk Training

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

3. Simpan Training Dataset

# Simpan dataset untuk 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()

Integrasi Serving Pipeline

1. Online Feature Retrieval

from feast import FeatureStorestore = FeatureStore(repopath=".") Get features untuk inference features = store.getonlinefeatures( features=[ "driverstats:convrate", "driverstats:accrate", "driverstats:avgdailytrips", ], entityrows=[{"driverid": 1001}], ).todict() Gunakan dengan 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. Integrasi FastAPI

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, # Untuk historical retrieval ) Feature view dengan 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. Integrasi Kafka

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 ke Feast store.push( pushsourcename="driverstatspush", df=pd.DataFrame([{ "driverid": event["driverid"], "currenttrips": event["trips"], "eventtimestamp": datetime.now(), }]), to=PushMode.ONLINE, )

Integrasi CI/CD

1. Feature Validation

# tests/testfeatures.py import pytest from feast import FeatureStore @pytest.fixture def store(): return FeatureStore(repopath="featurerepo") def testfeatureviewsexist(store): """Test semua expected feature views ada""" views = store.listfeatureviews() viewnames = [v.name for v in views] assert "driverstats" in viewnames assert "userfeatures" in viewnames def testfeaturesretrievable(store): """Test features bisa di-retrieve""" 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), # sertakan time window Feature(name="ispremiumuser", dtype=Bool), # prefix untuk booleans ], )

2. Feature Documentation

userfeatures = FeatureView( name="userfeatures", description=""" User-level features aggregated dari transaction history. Updated daily jam 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

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

Kesimpulan

Feast adalah essential untuk production ML systems dengan:

Feature consistency: Features sama di training dan serving

Point-in-time correctness: Cegah data leakage

Feature reuse: Share antar tim dan model

Low latency serving: Online feature retrieval

Centralized catalog*: Feature discovery

Key takeaways:

Define features sekali, gunakan dimana-mana
Gunakan feature services untuk versioning
Materialize secara regular untuk online serving
Integrate dengan ML pipeline Anda
Dokumentasi dan tag features untuk discoverability