Tutorial Lengkap Vertex AI Feature Store: Manajemen Fitur Terpusat

Vertex AI Feature Store adalah repositori terpusat untuk mengorganisir, menyimpan, dan menyajikan fitur ML. Memungkinkan penggunaan ulang fitur, mengurangi training-serving skew, dan menyediakan akses fitur yang konsisten antar tim.

Mengapa Feature Store?

Manfaat Utama:

• Fitur terpusat: Satu sumber kebenaran
• Penggunaan ulang fitur: Berbagi fitur antar model
• Serving latensi rendah: Pengambilan fitur online cepat
• Konsistensi: Fitur sama untuk training dan serving
• Time travel: Pencarian fitur point-in-time

Prerequisites

pip install google-cloud-aiplatform

gcloud auth login
gcloud config set project your-project-id

Setup

1. Inisialisasi Vertex AI

from google.cloud import aiplatform

aiplatform.init(project="your-project-id", location="us-central1")

2. Buat Feature Store

# Buat feature store
featurestore = aiplatform.Featurestore.create(

    featurestoreid="myfeaturestore",
    onlinestorefixednodecount=1
)

print(f"Feature store dibuat: {featurestore.resourcename}")

Entity Types

1. Buat Entity Type

# Buat customer entity type
customerentity = featurestore.createentitytype(
    entitytypeid="customer",
    description="Customer entity untuk prediksi churn"
)

Buat product entity type
productentity = featurestore.createentitytype(
    entitytypeid="product",
    description="Product entity untuk rekomendasi"
)

2. List Entity Types

entitytypes = featurestore.listentitytypes()
for et in entitytypes:

    print(f"{et.entitytypeid}: {et.description}")

Features

1. Buat Features

# Buat features untuk customer entity
customerentity.createfeature(

    featureid="age",
    valuetype="INT64",

    description="Usia customer"
)

customerentity.createfeature(

    featureid="tenuremonths",

    valuetype="INT64",
    description="Bulan sebagai customer"
)

customerentity.createfeature(
    featureid="monthlycharges",
    valuetype="DOUBLE",

    description="Tagihan bulanan"
)

customerentity.createfeature(

    featureid="totalcharges",

    valuetype="DOUBLE",
    description="Total tagihan sampai saat ini"
)

customerentity.createfeature(
    featureid="contracttype",
    valuetype="STRING",

    description="Tipe kontrak"
)

2. Batch Create Features

# Buat multiple features sekaligus
featuresconfig = {
    "age": {"valuetype": "INT64", "description": "Usia customer"},

    "tenuremonths": {"valuetype": "INT64", "description": "Tenure dalam bulan"},

    "monthlycharges": {"valuetype": "DOUBLE", "description": "Tagihan bulanan"},

    "totalcharges": {"valuetype": "DOUBLE", "description": "Total tagihan"},

    "contracttype": {"valuetype": "STRING", "description": "Tipe kontrak"}

}

customerentity.batchcreatefeatures(featuresconfig)

Ingesting Features

1. Ingest dari BigQuery

# Ingest features dari BigQuery
customerentity.ingestfrombq(
    featureids=["age", "tenuremonths", "monthlycharges", "totalcharges"],
    featuretime="updatetime",
    bqsourceuri="bq://project.dataset.customerfeatures",

    entityidfield="customerid"
)

2. Ingest dari DataFrame

import pandas as pd
from datetime import datetime

Buat feature dataframe
df = pd.DataFrame({
    "customerid": ["C001", "C002", "C003"],

    "age": [25, 35, 45],
    "tenuremonths": [12, 24, 36],
    "monthlycharges": [50.0, 75.0, 100.0],

    "totalcharges": [600.0, 1800.0, 3600.0],
    "featuretimestamp": [datetime.now()]  3

})

Ingest features
customerentity.ingestfromdf(
    featureids=["age", "tenuremonths", "monthlycharges", "totalcharges"],
    featuretime="featuretimestamp",
    dfsource=df,

    entityidfield="customerid"
)

3. Ingest dari GCS

# Ingest dari GCS Avro files
customerentity.ingestfromgcs(

    featureids=["age", "tenuremonths", "monthlycharges"],
    featuretime="timestamp",

    gcssourceuris=["gs://bucket/features/.avro"],
    gcssourcetype="avro",

    entityidfield="customerid"
)

Membaca Features

1. Online Serving

# Baca features untuk single entity
features = customerentity.read(

    entityids=["C001"],
    featureids=["age", "tenuremonths", "monthlycharges"]

)

print(features)

2. Batch Serving

# Baca features untuk multiple entities
featuresdf = customerentity.batchservetodf(
    bqdestinationoutputuri="bq://project.dataset.featuresoutput",
    servingfeatureids={
        "customer": ["age", "tenuremonths", "monthlycharges", "totalcharges"]

    },
    readinstancesuri="bq://project.dataset.entityids"
)

3. Point-in-Time Lookup

# Baca features pada timestamps spesifik
import pandas as pd

instancesdf = pd.DataFrame({

    "customerid": ["C001", "C002", "C003"],
    "timestamp": [
        "2024-01-01 00:00:00",
        "2024-01-15 00:00:00",
        "2024-02-01 00:00:00"
    ]
})

Batch serve dengan point-in-time
featurestore.batchservetodf(
    servingfeatureids={
        "customer": ["age", "tenuremonths", "monthlycharges"]
    },
    readinstancesdf=instancesdf,

    passthroughfields=["customerid", "timestamp"]
)

Training dengan Feature Store

1. Buat Training Dataset

from google.cloud import aiplatform

Definisikan training query
trainingdata = featurestore.batchservetobq(

    bqdestinationoutputuri="bq://project.dataset.trainingdata",

    servingfeatureids={

        "customer": ["age", "tenuremonths", "monthlycharges", "totalcharges"]
    },
    readinstancesuri="bq://project.dataset.traininginstances"

)

Buat Vertex AI dataset
dataset = aiplatform.TabularDataset.create(
    displayname="churn-training-dataset",
    bqsource="bq://project.dataset.trainingdata"
)

2. Integrasi Training Pipeline

from kfp import dsl
from kfp.dsl import component

@component(packagestoinstall=["google-cloud-aiplatform"])
def fetchfeatures(

    project: str,
    location: str,
    featurestoreid: str,

    entitytypeid: str,

    featureids: list,
    instancesuri: str,

    outputuri: str
):
    from google.cloud import aiplatform

    aiplatform.init(project=project, location=location)

    fs = aiplatform.Featurestore(featurestoreid)
    entitytype = fs.getentitytype(entitytypeid)


    entitytype.batchservetobq(

        bqdestinationoutputuri=outputuri,

        servingfeatureids=featureids,
        readinstancesuri=instancesuri

    )

@dsl.pipeline
def trainingpipeline():
    # Fetch features
    fetchtask = fetchfeatures(
        project="your-project",
        location="us-central1",
        featurestoreid="myfeaturestore",
        entitytypeid="customer",
        featureids=["age", "tenuremonths", "monthlycharges"],

        instancesuri="bq://project.dataset.instances",
        outputuri="bq://project.dataset.trainingfeatures"
    )

    # Train model
    traintask = trainmodel(datauri=fetchtask.outputs["outputuri"])

Online Serving

1. Konfigurasi Online Store

# Update feature store untuk online serving
featurestore = aiplatform.Featurestore(
    featurestorename="myfeaturestore"

)

Scale online serving nodes
featurestore.update(onlinestorefixednodecount=3)

2. Serve Features Online

# Baca features online
onlinefeatures = customerentity.read(
    entityids=["C001", "C002"],

    featureids=["age", "tenuremonths", "monthlycharges"]
)

for entityid, features in onlinefeatures.items():
    print(f"Entity {entityid}:")

    for featureid, value in features.items():
        print(f"  {featureid}: {value}")

3. Streaming Ingestion

# Stream features untuk update real-time
import time

def streamfeatures(entityid, features):

    customerentity.writefeaturevalues(
        entityid=entityid,
        featurevalues=features

    )

Contoh: Update features secara real-time
streamfeatures("C001", {
    "monthlycharges": 55.0,

    "totalcharges": 655.0
})

Feature Monitoring

1. Aktifkan Monitoring

# Buat monitoring job
monitoringjob = featurestore.createmonitoringjob(
    displayname="feature-monitoring",

    entitytypeids=["customer"],

    featureids=["age", "tenuremonths", "monthlycharges"],
    objectiveconfig={

        "trainingdataset": "bq://project.dataset.trainingdata",

        "targetfield": "churn"
    },
    schedule="0 0   *"  # Harian
)

2. Lihat Hasil Monitoring

# Dapatkan monitoring stats
stats = customerentity.getfeaturemonitoringstats(
    featureid="monthlycharges",
    starttime="2024-01-01",

    endtime="2024-02-01"
)

for stat in stats:
    print(f"Tanggal: {stat.date}")
    print(f"Mean: {stat.mean}")
    print(f"Stddev: {stat.stddev}")

Feature Groups (V2)

1. Buat Feature Group

from google.cloud.aiplatformv1beta1 import (
    FeatureRegistryServiceClient,
    FeatureGroup,
    BigQuerySource
)

client = FeatureRegistryServiceClient()

featuregroup = FeatureGroup(
    bigquery=BigQuerySource(

        uri="bq://project.dataset.customerfeatures",
        entityidcolumns=["customerid"]

    ),
    description="Customer features dari BigQuery"
)

operation = client.createfeaturegroup(

    parent=f"projects/{project}/locations/{location}",
    featuregroupid="customerfeatures",
    featuregroup=featuregroup
)

result = operation.result()
print(f"Feature group dibuat: {result.name}")

2. Buat Feature View

from google.cloud.aiplatformv1beta1 import FeatureView

featureview = FeatureView(
    bigquerysource=BigQuerySource(
        uri="bq://project.dataset.customerfeatures",

        entityidcolumns=["customerid"]
    ),
    featureregistrysource={
        "featuregroups": {

            "customerfeatures": {
                "featureids": ["age", "tenuremonths", "monthlycharges"]

            }
        }
    }
)

Buat feature view untuk online serving
client.createfeatureview(

    parent=f"projects/{project}/locations/{location}/featureOnlineStores/{storeid}",
    featureviewid="customerview",

    featureview=featureview

)

Best Practices

1. Penamaan Feature

# Gunakan konvensi penamaan yang konsisten
features = {
    "customerageyears": {"valuetype": "INT64"},
    "customertenuremonths": {"valuetype": "INT64"},

    "customermonthlychargesusd": {"valuetype": "DOUBLE"},

    "customercontracttypecat": {"valuetype": "STRING"}

}

2. Resource Cleanup

# Hapus features
customerentity.deletefeatures(["oldfeature"])

Hapus entity type
customerentity.delete()


Hapus feature store
featurestore.delete(force=True)

Kesimpulan

Vertex AI Feature Store menyediakan:

Sentralisasi: Repositori fitur tunggal

Konsistensi: Paritas training-serving

Reusability: Berbagi fitur antar model

Latensi rendah: Serving online cepat

Time travel: Pencarian point-in-time

Key takeaways:

• Organisasi fitur berdasarkan entity types
• Gunakan batch ingestion untuk efisiensi
• Aktifkan online serving untuk real-time
• Monitor distribusi fitur
• Gunakan konvensi penamaan yang konsisten