Özellik Görünümleri

Important

Bu özellik Genel Önizleme aşamasındadır. Çalışma alanı yöneticileri Bu özelliğe erişimi Önizlemeler sayfasından denetleyebilir. Bkz. Azure Databricks önizlemelerini yönetme.

Özellik Görünümleri, veri kaynaklarından özellikleri tanımlamanızı ve hesaplamanızı sağlar. Özellikler çeşitli kaynaklar (Delta tablosu, Kafka Stream ve istek zamanı verileri) ve hesaplamalar (zaman pencereli toplamalar, basit sütun seçimleri ve daha fazlası) kullanılarak tanımlanabilir. Bu kılavuz aşağıdaki iş akışlarını kapsar:

• Özellik geliştirme iş akışı
  • Model eğitiminde ve iş akışlarını sunmada kullanılabilecek Unity Kataloğu özellik nesnelerini tanımlamak için kullanın create_feature .
  • Alternatif olarak, Feature nesnelerini yerel olarak oluşturun ve ardından onları Unity Kataloğu'na kaydetmek için register_feature kullanın. Yerel olarak yapılan özellikler create_training_set ile kayıt öncesinde birlikte kullanılabilir.
• Model eğitim iş akışı
  • Makine öğrenmesi için zamana bağlı toplanmış özellikleri hesaplamak için create_training_set kullanın. Özellik Görünümleri ile eğitim hakkında ayrıntılı belgeler için bkz. Özellik Görünümleri ile modelleri eğitma.
• Özellik maddileştirme ve sunma iş akışı
  • Bir özelliği create_feature ile tanımladıktan veya get_feature kullanarak aldıktan sonra, özelliği veya özellik kümesini verimli bir şekilde yeniden kullanmak üzere, çevrimdışı bir mağazada veya çevrimiçi sunum için çevrimiçi bir mağazada uygulatmak için materialize_features kullanabilirsiniz.
  • Malzemeleşmiş görünümle create_training_set kullanarak çevrimdışı toplu eğitim veri kümesini hazırlayın.

API ayrıntıları için bkz. Özellik Görünümleri API başvurusu.

Requirements

• Sunucusuz işlem veya Databricks Runtime 17.0 ML veya üzerini çalıştıran klasik bir işlem kümesi.

• Özel Python paketini yüklemeniz gerekir. Not defterini her çalıştırdığınızda aşağıdaki kod satırlarını çalıştırın:

  %pip install databricks-feature-engineering>=0.16.0
  dbutils.library.restartPython()
  

Hızlı başlangıç örneği

Çalıştırılabilir hızlı başlangıç not defteri için bkz. Örnek not defteri.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    CronSchedule, DeltaTableSource, Feature, AggregationFunction,
    Sum, Avg, ColumnSelection, TableTrigger,
    TumblingWindow, SlidingWindow,
    OfflineStoreConfig, OnlineStoreConfig,
)
from datetime import timedelta

CATALOG_NAME = "main"
SCHEMA_NAME = "feature_store"
TABLE_NAME = "transactions"

# 1. Create data source
source = DeltaTableSource(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name=TABLE_NAME,
)

# 2. Define features locally (no catalog/schema needed yet)
avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), TumblingWindow(window_duration=timedelta(days=30))),
    name="avg_transaction_30d",
)

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), SlidingWindow(window_duration=timedelta(days=7), slide_duration=timedelta(days=1))),
    # name auto-generated: "amount_sum_sliding_7d_1d"
)

fe = FeatureEngineeringClient()

# 3. Explore features with compute_features
feature_df = fe.compute_features(features=[avg_feature, sum_feature])
feature_df.display()

# 4. Create training set using local features
# `labeled_df` should have columns "user_id", "transaction_time", and "target".
training_set = fe.create_training_set(
    df=labeled_df,
    features=[avg_feature, sum_feature],
    label="target",
)
training_set.load_df().display()

# 5. Register features in Unity Catalog
avg_feature = fe.register_feature(
    feature=avg_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)
sum_feature = fe.register_feature(
    feature=sum_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)

# 6. Or use create_feature for a one-step define-and-register workflow
latest_amount = fe.create_feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    name="latest_amount",
)

# 7. Train model
with mlflow.start_run():
    training_df = training_set.load_df()

    # training code

    fe.log_model(
        model=model,
        artifact_path="recommendation_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name=f"{CATALOG_NAME}.{SCHEMA_NAME}.recommendation_model",
    )

# 8. (Optional) Materialize features for serving
# Features must be registered in UC before calling materialize_features
online_config = OnlineStoreConfig(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name_prefix="customer_features_serving",
    online_store_name="customer_features_store",
)

# Aggregation features use CronSchedule and support both offline and online configs
fe.materialize_features(
    features=[avg_feature, sum_feature],
    offline_config=OfflineStoreConfig(
        catalog_name=CATALOG_NAME,
        schema_name=SCHEMA_NAME,
        table_name_prefix="customer_features",
    ),
    online_config=online_config,
    trigger=CronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # Hourly
        timezone_id="UTC",
    ),
)

# ColumnSelection features use TableTrigger and only support online config
fe.materialize_features(
    features=[latest_amount],
    online_config=online_config,
    trigger=TableTrigger(),
)

Örnek not defteri

Özellik Görünümleri hızlı başlangıç not defteri

Dizüstü bilgisayar al

Akış özellikleri

Delta tablolarındaki toplu iş özelliklerine ek olarak, gerçek zamanlı kullanım örnekleri için akış kaynaklarından özellikler tanımlayabilirsiniz. Akış özellikleri, toplu iş özellikleriyle aynı Özellik sınıfını (aynı Feature oluşturucular, aynı toplama işlevleri, aynı eğitim ve sunum iş akışları) kullanır. Bu nedenle toplu işlemden gerçek zamanlıya yükseltmek için en az kod değişikliği gerekir. Akış özellikleri materialize edildiklerinde, doğrudan model sunum uç noktalarınıza uçtan uca bir saniyenin altında güncellik (200 ms p99 gecikmesi) sağlar.

Akış özelliklerini kullanmak için önce bir Stream oluşturun, ardından buna bir StreamSource kullanarak başvurun. Akış kaynakları, Kafka’yı girdi olarak destekler ve eğitim için verilerin geçmişe dönük bir kopyasını veri alımı (Delta) tablosunda otomatik olarak tutar.

Akış özelliği tanımlama

StreamSource, bir Stream'e üç parçalı adıyla (catalog.schema.stream_name) başvurur. Akış, Unity Catalog'da güvenli bir nesne değildir; ancak Unity Catalog şeması kapsamındadır ve erişim, akışın alma tablosu tarafından yönetilir. Kafka iletisinin hangi bölümünün okunacağını belirtmek için, varlık, zaman serisi ve işlev tanımlarındaki sütun başvurularının önüne value. veya key. getirilmelidir. İç içe alanlar nokta gösterimi kullanılarak desteklenir (örneğin, value.user.address.city).

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    StreamSource,
    Feature,
    AggregationFunction,
    Sum,
    RollingWindow,
)
from datetime import timedelta

client = FeatureEngineeringClient()

stream_source = StreamSource(
    full_name="my_catalog.my_schema.my_stream",
)

feature = Feature(
    name="user_purchase_sum",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(
        operator=Sum(input="value.amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
)

StreamSource'ta filtreleme koşulları

Akıştan satırları toplulaştırmadan önce filtrelemek için, tıpkı filter_condition üzerinde olduğu gibi DeltaTableSource kullanın.

stream_source = StreamSource(
    full_name="my_catalog.my_schema.my_stream",
    filter_condition="value.event_type = 'purchase'",
)

Akışlardan sütun seçimi

ColumnSelection özellikler akış kaynaklarıyla çalışır. Seçilen sütun, belirli bir noktadaki doğruluğu dikkate alırken her varlık için Stream'den en son değeri temsil eder.

from databricks.feature_engineering.entities import ColumnSelection

passenger_count = Feature(
    name="passenger_count",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=ColumnSelection(column="value.passenger_count"),
)

İç içe yerleştirilmiş alanlara erişme

nokta gösterimi kullanarak iç içe JSON alanlarına erişebilirsiniz (örneğin, value.nested_field.amount). Sunum sırasında istek yükü ile yanıt, yaprak düğüm adlarını kullanır (örneğin, amount yerine value.amount). Sunum uç noktası değerleri yönlendirmek için yaprak adları kullandığından, yaprak düğüm adları model veya Özellik Belirtimi içindeki tüm varlık, zaman aralıkları ve özellik çıktı sütunları arasında benzersiz olmalıdır.

Akış özellikleri için zaman pencereleri

Akış özellikleri yalnızca RollingWindow toplulaştırmalarını destekler. Kayan pencereler, en güncel veriler üzerinde sürekli olarak yeniden hesaplama yapar; bu da akış kaynaklarının gerçek zamanlı doğasıyla uyumludur. TumblingWindow ve SlidingWindow sabit geçmiş aralıklar üzerinden toplu işlem için tasarlanmıştır.

Akış özellikleri örnek not defteri

Akış Özellik Görünümleri Hızlı Başlangıç Not Defteri

Dizüstü bilgisayar al

Model eğitimi ve çıkarımı

Modelleri eğitmek ve log_model(), score_batch() ve create_training_set() dâhil olmak üzere Özellik Görünümleriyle toplu çıkarım gerçekleştirmek için Özellik Görünümleriyle modelleri eğitme bölümüne bakın.

Özellik somutlaştırma

Özellikleri tanımladıktan sonra, eğitim ve hizmet iş akışlarında verimli bir şekilde yeniden kullanmak için bunları çevrimdışı veya çevrimiçi mağazalarda gerçekleştirebilirsiniz. Özellikleri gerçekleştirdikten sonra, CPU modeli sunma özelliğini kullanarak modelleri servis edebilirsiniz. Ayrıntılar için bkz. Özellik Görünümlerini Somutlaştırma.

En iyi yöntemler

Özellik adlandırma

• İş açısından kritik özellikler için açıklayıcı adlar kullanın.
• Ekipler arasında tutarlı adlandırma kurallarını izleyin.
• Özellik geliştirmeye başlarken otomatik olarak oluşturulan adları kullanın.

Zaman aralıkları

• Pencere sınırlarını iş döngüleriyle (günlük, haftalık) hizalayın.
• Kısa pencereler son eğilimleri yakalar ancak gürültülü olabilir. Daha uzun pencereler daha kararlı özellik dağıtımları üretir, ancak son davranış değişikliklerini kaçırabilir. Kullanım örneğiniz için temel alınan sinyalin ne kadar hızlı değiştiğine göre seçin. Örneğin, 7 günlük bir pencere günlük dalgalanmaları düzeltir ve tutarlı model girişleri üretirken, 1 saatlik bir zaman aralığı davranış değişikliklerine hızlı tepki verir, ancak model performansını düşüren varyansa neden olabilir. Dağıtım kaydığında modelinizin doğruluğu azalırsa girişleri dengelemek için daha uzun bir pencere kullanın.
• Atlayan ve kayan pencereler, sıralı (sürekli) pencerelerden daha ölçeklenebilir. Çoğu kullanım örneği için kayan pencerelerle başlayın.

Performance

• Veri taramalarını en aza indirmek için aynı veri kaynağındaki özellikleri tek materialize_features bir çağrıda gerçekleştirin.
• Gerçekleştirme sırasında daha iyi gruplandırma sağlamak için aynı veri kaynağındaki özellikler için aynı ayrıntı düzeyini (örneğin, 1 saatlik veya 1 günlük slayt sürelerinin tümü) kullanın.

Varlık sütunları ve filtre koşulları karşılaştırması

Aynı kaynak tablodaki özelliklerle çalışırken bu karar kılavuzunu kullanın:

Farklı toplama düzeylerine ihtiyacınız olduğunda (üzerindeentity) kullanın create_feature :

• Müşteri düzeyinde özellikler (müşteri başına bir satır): entity=["customer_id"]
• Müşteri-satıcı özellikleri (müşteri başına birden çok satır): entity=["customer_id", "merchant_id"]
• Farklı toplama düzeyleri aynı DeltaTableSourcedeğeri paylaşabilir: Her özellik tanımında farklı entity değerler belirtin

Satırları aynı toplama düzeyinde filtrelemeniz gerektiğinde,filter_condition (üzerinde DeltaTableSource) kullanın:

• Yalnızca yüksek değerli işlemler: filter_condition="amount > 100" (yine de müşteri başına toplanır)
• Yalnızca tamamlanan siparişler: filter_condition="status = 'completed'" (yine de müşteri başına toplanır)

Temel kural: Değişikliğiniz varlık değeri başına farklı sayıda satırla sonuçlanırsa özellik tanımlarınızda farklı entity değerler kullanın. Yalnızca aynı toplamaya katkıda bulunan satırları filtreliyorsanız kaynakta kullanın filter_condition .

Ortak desenler

Müşteri analizi

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow

fe = FeatureEngineeringClient()
features = [
    # Recency: Number of transactions in the last day
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=1)))),

    # Frequency: transaction count over the last 90 days
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=90)))),

    # Monetary: total spend in the last month
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30)))),
]

Eğilim çözümlemesi

# Compare recent vs. historical behavior
fe = FeatureEngineeringClient()
recent_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

historical_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7), delay=timedelta(days=7))),
)

Mevsimsel desenler

# Same day of week, 4 weeks ago
fe = FeatureEngineeringClient()
weekly_pattern = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=1), delay=timedelta(weeks=4))),
)

Sınırlamalar

• Api'de create_training_set kullanıldığında varlık ve zaman aralığı sütunlarının adları eğitim (etiketli) veri kümesi ile özellik tanımları arasında eşleşmelidir.
• Eğitim veri kümesinde sütun olarak label kullanılan sütun adı, s tanımlamak Featureiçin kullanılan kaynak tablolarda bulunmamalıdır.
• API'de create_feature sınırlı bir işlev listesi (UDAFs) desteklenir. Bkz . Desteklenen işlevler.
• Varlık sütunları DATE veya TIMESTAMP türünde olamaz.
• RequestSource, ScalarDataType içinde tanımlanan yalnızca skaler veri türlerini destekler: INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, DATE, SHORT. Diziler, haritalar ve yapılar gibi karmaşık türler desteklenmez.
• RequestSource toplama işlevlerini veya zaman pencerelerini desteklemez. Yalnızca ColumnSelection işlevler kullanılabilir.
• Varlık sütunu adları, zaman aralıkları sütun adları ve istek özelliği sütun adları kümesi, bir eğitim kümesindeki veya hizmet uç noktasındaki tüm kaynaklar arasında genel olarak benzersiz olmalıdır.

• score_batch sunucusuz işlemde başarılı olmayabilir. Databricks Runtime 17.0 ML veya üzerini çalıştıran klasik bir işlem kümesi kullanarak bu sorunu geçici olarak düzeltin.

Gerçekleştirmeye özgü sınırlamalar için bkz. Sınırlamalar.