Neredeyse 2 yıl sonra yazdığım SageMaker 101 yazısının devamı niteliğinde olan bu yazıda AWS SageMaker Feature Store kullanarak makine öğrenmesi projelerinde veri yönetimini nasıl daha etkili hale getirebileceğimizi göstereceğim. Kaggle’da bulunan örnek veri seti ile feature group oluşturup veriyi ekledikten sonra veri sorgulama işlemlerini gerçekleştireceğiz. İlk olarak Feature Store nedir, ne için ve nasıl kullanılır ona bakalım.

AWS SageMaker Feature Store Nedir?

AWS SageMaker Feature Store (FS), makine öğrenmesi (ML) modellerinde kullanılan değişkenlerin merkezi ve yönetilebilir bir şekilde saklanması için geliştirilmiş bir çözümdür. ML projelerinde veriler genellikle model eğitiminden önce çeşitli işlemlerden geçirilir. Bu süreç çoğu zaman tekrarlayan, zaman alıcı ve hata yapmaya açık bir yapıya sahiptir. FS bu süreci standartlaştırarak veri tutarlılığı sağlar ve hem eğitim hem de tahmin aşamalarında aynı verilerin güvenilir bir şekilde kullanılmasına olanak tanır.

Görsel Kaynağı

Temel İşlevleri ve Sağladığı Kolaylıklar

FS oluşturulan değişkenlerin tek bir noktada toplanmasını ve tekrar kullanılabilir hale gelmesini sağlar. Değişkenler hem eğitim hem de tahmin süreçlerinde tek kaynaktan beslendiği için tutarlılık korunur ve veri uyumsuzlukları en aza indirilir.

Sistem, online ve offline olmak üzere iki farklı depolama katmanı sunar. Online Store düşük gecikme süreleriyle gerçek zamanlı tahmin gerektiren senaryolarda değişkenlerin anlık erişimini mümkün kılar. Offline Store ise tüm geçmiş veriyi Athena ya da Glue üzerinden sorgulanabilir bir şekilde S3 üzerinde versiyonlu olarak saklar. Bu yapı model eğitimi ve geçmişe dönük analizlere olanak sağlar.

Bir diğer önemli özellik ise zaman damgası ve versiyonlama desteğidir. FS point-in-time correctness mantığıyla çalışır ve geçmişe dönük tutarlı veri kümelerinin oluşturulmasına imkan tanır. Böylece eğitim sırasında gelecekteki verilerin yanlışlıkla modele beslenmesi (data leakage) riski ortadan kaldırılır.

Bunların yanı sıra FS SageMaker Training, Pipelines ve Model Registry gibi MLOps bileşenleriyle entegre çalışır. Amazon Athena, AWS Glue, S3 ve Kinesis gibi veri servisleriyle de uyumlu şekilde çalışarak veri altyapısının genel esnekliğini artırır. Yönetilen bir servis (managed service) olduğu için de kullanıcılar ölçeklenme, güvenlik veya yüksek erişilebilirlik gibi konularla uğraşmak zorunda kalmaz.

Kullanım Avantajları

FS’nin en büyük avantajlarından biri aynı değişkenlerin farklı projelerde tekrar oluşturmaya gerek kalmadan kullanılabilmesidir. Bu sayede hem zaman hem de maliyet açısından önemli bir tasarruf sağlanır. Özellikle gerçek zamanlı tahmin gerektiren senaryolarda online store’un sunduğu milisaniyelik erişim imkanı büyük bir fark yaratır.

Eğitim ve tahmin aşamalarında aynı veri kaynağının kullanılması veri tutarlılığını sağlarken MLOps süreçleriyle entegrasyonu sayesinde otomatize ML akışlarının daha sorunsuz olmasına olanak tanır. Ayrıca zaman yolculuğu (time travel) desteğiyle geçmişe dönük model eğitimi ve versiyonlama kolaylıkla gerçekleştirilebilir.

Her Zaman Kullanılmalı?

Her ne kadar FS birçok avantaj sunsa da her projede kullanımı gerekli değildir. Küçük ve tek seferlik projelerde, değişkenlerin tekrar kullanılmayacağı senaryolarda ya da yalnızca offline batch tahminlerin yapıldığı durumlarda ek bir katman kurmak gereksiz karmaşıklık ve maliyet getirebilir. Mevcut veri ambarlarının (DWH) güçlü ve ML için optimize edilmiş olduğu ortamlarda da FS yerine mevcut altyapıyı kullanmak daha uygun olacaktır.

Klasik Veritabanlarından Farkı

Klasik veritabanları uygulama verilerini saklamak, raporlamak ve CRUD işlemleri yapmak için tasarlanmışken FS tamamen makine öğrenmesi süreçlerine odaklanır. Veritabanlarında aynı özelliği farklı ekipler farklı sorgularla üretmek zorunda kalır ve bu da tutarsız sonuçlara yol açabilir. FS ise bir kez tanımlanan özelliği tüm süreçlerde tutarlı şekilde kullanılabilir hale getirir.

Görsel Kaynağı

Bunun yanı sıra veritabanları genellikle ya batch ya da transactional işlemlere odaklanır. İkisini birden düşük gecikmeyle ve versiyon kontrolüyle yönetmek zordur. FS hem online hem de offline depolamayı entegre bir şekilde sunarak bu ihtiyacı karşılar. Zaman yolculuğu desteğiyle geçmişe dönük tutarlı veri kümeleri oluşturulabilir ve veri kaçağı riski en aza indirilir. Ayrıca feature lineage, otomatik şema yönetimi ve model eğitim/tahmin tutarlılığı gibi MLOps’a özgü yetenekler de FS’nin sunduğu avantajlar arasındadır.

Görsel Kaynağı

Performans açısından da önemli farklar bulunur. Veritabanlarında büyük sorgular genellikle saniyeler veya dakikalar sürebilirken FS online store ile milisaniyeler içinde tek bir kayda ulaşılabilir. Offline store ise büyük veri sorgularını batch olarak yönetir.

Uygulama

Örnek senaryo için Kaggle’da bulunan Blinkit Marketing and Customer Feedback Dashboard veri setini kullanıyoruz. Uygulamada SageMaker notebook kullanarak nasıl yeni bir feature group oluşturulacağı ve oluşturulan feature group’a nasıl veri ekleneceğini göreceğiz. Ayrıca oluşturulan feature group içerisinden kayıtların hem online hem de offline için nasıl sorgulanacağı ve içerisindeki verilere nasıl erişileceği de gösterilmiştir.

İlk olarak SageMaker AI altında notebooks ile bir notebook instance oluşturuyoruz. ml.t3.medium seçmeniz yeterli olur ¹. Notebook instance’ı oluşturduktan sonra aşağıdaki kodları çalıştırarak gerekli kütüphaneleri import ediyoruz.

import time, uuid, json, boto3, pandas as pd, numpy as np
from datetime import datetime, timezone
from time import strftime, gmtime

import sagemaker
from sagemaker.session import Session
from sagemaker.feature_store.feature_group import FeatureGroup
from sagemaker.feature_store.feature_definition import (
    FeatureDefinition,
    FeatureTypeEnum)

Gerekli kütüphaneleri çağırdıktan sonra aşağıdaki kod bloğunu çalıştırarak hem gerekli bağlantıları hem de değişkenleri tanımlıyoruz.

project = "blinkit"
database_name = "sagemaker_featurestore"
sagemaker_session = sagemaker.Session()
region = sagemaker_session.boto_region_name
smfs_rt = boto3.client("sagemaker-featurestore-runtime", region_name=region)
sm = boto3.client("sagemaker", region_name=region)
bucket = sagemaker_session.default_bucket()
offline_s3_uri = f"s3://{bucket}/{project}/feature-store"
role = sagemaker.get_execution_role()

Veriyi SageMaker ortamı (domain) ayağa kaldırırken oluşturulan S3 bucket’ına yükledikten sonra S3’teki veri aşağıdaki şekilde pandas ile okunabilir.

s3_path_o = f"s3://{bucket}/data/blinkit_orders.csv"
s3_path_c = f"s3://{bucket}/data/blinkit_customers.csv"
orders = pd.read_csv(s3_path_o)
customers = pd.read_csv(s3_path_c)

Verileri okuduktan sonra ilk olarak FeatureGroup oluşturuyoruz. Feature group oluşturulurken tekillik sağlanabilmesi için zaman damgası ekliyoruz.

customers_feature_group_name = "customers-feature-group-" + strftime("%d-%H-%M-%S", gmtime())
orders_feature_group_name = "orders-feature-group-" + strftime("%d-%H-%M-%S", gmtime())

customers_feature_group = FeatureGroup(
    name=customers_feature_group_name, sagemaker_session=sagemaker_session)

orders_feature_group = FeatureGroup(
    name=orders_feature_group_name, sagemaker_session=sagemaker_session)

FS için zorunlu olan EventTime değişkeni oluşturuyoruz ve dataframe’e yeni bir değişken olarak ekliyoruz. Bununla birlikte customer_id değişkenini de record_identifier_feature_name olarak tanımlıyoruz².

current_time_sec = int(round(time.time()))
record_identifier_feature_name = "customer_id"
customers["EventTime"] = pd.Series([current_time_sec] * len(customers), dtype="float64")
orders["EventTime"] = pd.Series([current_time_sec] * len(orders), dtype="float64")

Her bir FeatureGroup için veri şemasını (dtypes) otomatik oluşturmak için load_feature_definitions fonksiyonunu çalıştırıyoruz. Dilerseniz bunu kendiniz de FeatureDefinition kullanarak ekleyebilirsiniz.

customers_feature_group.load_feature_definitions(data_frame=customers)
orders_feature_group.load_feature_definitions(data_frame=orders)

# customers için çıktı örneği
[FeatureDefinition(feature_name='customer_id', feature_type=<FeatureTypeEnum.INTEGRAL: 'Integral'>, collection_type=None),
 FeatureDefinition(feature_name='customer_name', feature_type=<FeatureTypeEnum.STRING: 'String'>, collection_type=None)
 ...
 FeatureDefinition(feature_name='EventTime', feature_type=<FeatureTypeEnum.FRACTIONAL: 'Fractional'>, collection_type=None)]

Veriye ait tanımları da oluşturduktan sonra FeatureGroup’ları create komutuyla oluşturuyoruz.

customers_feature_group.create(
    s3_uri=offline_s3_uri,
    record_identifier_name=record_identifier_feature_name,
    event_time_feature_name="EventTime",
    role_arn=role,
    enable_online_store=True,
)
orders_feature_group.create(
    s3_uri=offline_s3_uri,
    record_identifier_name=record_identifier_feature_name,
    event_time_feature_name="EventTime",
    role_arn=role,
    enable_online_store=True,
)

Bu aşamada isterseniz arayüzden isterseniz de feature_group.describe().get("FeatureGroupStatus") fonksiyonu ile oluşturulma durumunu kontrol edebilirsiniz.

Sorunsuz bir şekilde feature group’ların oluşturulmasından sonra sıra veriyi Feature Store’a yüklemeye geliyor. ingest fonksiyonu ile veriyi kolayca yükleyebiliriz.

# replace içerisindeki karakterlerden dolayı hata aldığı için düzenlendi
# customers["address"] = customers["address"]
#     .str.replace("/", " ")
#     .str.replace("\n", " ")
#     .str.replace(",", " ")

customers_feature_group.ingest(data_frame=customers, max_workers=5, wait=True)
orders_feature_group.ingest(data_frame=orders, max_workers=5, wait=True)

Yükleme başarıyla tamamlandıktan sonra online store’dan veriyi sorgulayabiliriz. Rastgele bir müşteri kaydını sorgulamak için aşağıdaki kodu kullanabiliriz.

sample_id = int(customers.sample(1, random_state=2025)["customer_id"].iloc[0])
resp = smfs_rt.get_record(
    FeatureGroupName=customers_feature_group_name,
    RecordIdentifierValueAsString=str(sample_id),
    FeatureNames=[
        "customer_name",
        "email",
        "phone",
        "address",
        "area",
        "pincode",
        "registration_date",
        "customer_segment",
        "total_orders",
        "avg_order_value",
    ],
)
record = {f["FeatureName"]: f["ValueAsString"] for f in resp.get("Record", [])}
record

# {'customer_name': 'Harita Chatterjee',
#  'email': 'gaurika82@example.com',
#  'phone': '911099088799',
#  'address': 'H.No. 249  Malhotra Ghaziabad-893659',
#  'area': 'Raipur',
#  'pincode': '674102',
#  'registration_date': '2023-09-27',
#  'customer_segment': 'Regular',
#  'total_orders': '14',
#  'avg_order_value': '1956.45'}

Bu noktada sorgulama sonucunda elde ettiğimiz veriyi kullanarak çeşitli analizler yapabiliriz. Sorunsuz bir şekilde Feature Store oluşturup, veri yükleyip yüklediğimiz veriyi de kullanabilir durumdayız. Bununla birlikte Feature Store oluşturduğumuzda veriler S3’te depolanır ve gerektiğinde bu verilere erişim sağlanabilir. Bu verilere erişimi Glue üzerinden Athena ile gerçekleştirebiliriz.

Athena ile dilerseniz arayüzden dilerseniz de aşağıdaki kod ile offline veriyi sorgulayabilirsiniz.

athena_query = customers_feature_group.athena_query()
table_name = athena_query.table_name
print("Athena table name:", table_name, "| DB:", database_name)
# Athena table name: customers_feature_group_25_16_58_54_1756141135 | DB: sagemaker_featurestore

query_str = f"""
SELECT 
  {record_identifier_feature_name},
  customer_name, email, phone, address, area,
  from_iso8601_timestamp(registration_date) AS event_time
FROM "{database_name}"."{table_name}"
"""

output_s3 = f"s3://{bucket}/{project}/athena-results/"
athena_query.run(query_string=query_str, output_location=output_s3)
athena_query.wait()

# athena sorgu sonucu
train_df = athena_query.as_dataframe()
train_df.head(2)

Glue ve Athena’ya ait ekran görüntüleri aşağıda görülebilir.

Yazıda bahsetmedim ama oluşturduğunuz SageMaker ortamına ait rol için gerekli izinleri & policy’leri (S3, Glue, Athena vb.) tanımlamanız gerekiyor. Yine yazıda herhangi bir modelleme sürecine dair detaylar verilmedi. FS’nin ücretlendirilmesi ise Online Store istekleri, S3 saklama ve Athena sorguları için ayrı maliyetler şeklinde olacaktır. Yine de en doğru ve güncel bilgiye AWS dökümantasyonundan ulaşabilirsiniz.

Uygulama kodlarının olduğu notebook’a buradan ulaşabilirsiniz.

Kaynaklar

• Feature Store concepts
• Amazon SageMaker Feature Store
• github.com/aws/amazon-sagemaker-examples