Feature Views API referansı

Erişim denetimi

Özellikler yönetilebilir Unity Kataloğu nesneleridir. Bir özelliğe erişim, , CREATE FEATUREve READ FEATURE Unity Kataloğu ayrıcalıkları tarafından MANAGEdenetlenmektedir. Tam açıklamalar için bkz. Unity Kataloğu ayrıcalık başvurusu.

• CREATE FEATURE — Şemada özellik oluşturmak için gereklidir. create_feature ve register_feature üst şemada gerektirir CREATE FEATURE . En az ayrıcalık ilkesine uyarak şema düzeyinde verin CREATE FEATURE ; ayrıca bu kataloğa bir katalogdaki herhangi bir şemada özellik oluşturmaya izin vermek için bunu bir katalogda da vekleyebilirsiniz.
• READ FEATURE — Bir özelliği ve verilerini okumak için gereklidir. get_feature, create_training_setve eğitim veya sunum READ FEATURE için gerçekleştirilmiş özellik verilerini okuma özelliği gerektirir. READ FEATURE şema veya katalog üzerinde verilen, içerdiği tüm geçerli ve gelecekteki özellikler için geçerlidir.
• MANAGE — Bir özelliğin yaşam döngüsünü ve izinlerini yönetmek için gereklidir. ile delete_featurebir özelliği silme ve veya materialize_featuresdelete_materialized_feature ile MANAGE bir özelliği gerçekleştirme özelliği gerektirir.

Tüm özellik işlemleri üst katalogda ve USE CATALOG üst şemada da gereklidirUSE SCHEMA. Gerçekleştirmenin nasıl MANAGE ve READ FEATURE uygulanacağı için bkz. İzinler.

Özellik Görünümü API'si

Feature oluşturucu ve register_feature()

Önerilen yaklaşım, yerel olarak bir Feature nesne oluşturmak ve bunu Unity Kataloğu'na kalıcı hale getirmek için kullanmaktır register_feature . Bu iki adımlı iş akışı, kaydetmeden önce özellikleri (dahil create_training_set) denemenize olanak tanır.

Feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
)

FeatureEngineeringClient.register_feature() Unity Kataloğu'nda yerel olarak inşa edilmiş Feature bir kaydı kaydeder.

FeatureEngineeringClient.register_feature(
    feature: Feature,       # Required: A Feature instance (not already registered)
    catalog_name: str,      # Required: UC catalog name
    schema_name: str,       # Required: UC schema name
) -> Feature

from databricks.feature_engineering.entities import Feature, DeltaTableSource, AggregationFunction, Sum, RollingWindow
from datetime import timedelta

# Step 1: Construct the feature locally
feature = Feature(
    source=DeltaTableSource(catalog_name="main", schema_name="store", table_name="transactions"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

# Step 2: Register in Unity Catalog
fe = FeatureEngineeringClient()
registered_feature = fe.register_feature(
    feature=feature,
    catalog_name="main",
    schema_name="store",
)

create_feature()

FeatureEngineeringClient.create_feature() tek bir adımda Unity Kataloğu'nda bir özelliği doğrular, oluşturur ve hemen kaydeder. İlk olarak özelliği yerel olarak denemeniz gerekmeyen durumlarda bunu kullanın.

FeatureEngineeringClient.create_feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    catalog_name: str,                                     # Required: The catalog name for the feature
    schema_name: str,                                      # Required: The schema name for the feature
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

Parametreler:

• source: Özellik hesaplamasında (DeltaTableSource, StreamSourceveya RequestSource) kullanılan veri kaynağı.
• function AggregationFunction: işlecini (örneğin, Sum(input="amount")), giriş sütununu ve zaman penceresini bir araya getiren bir. Veya ColumnSelection("column_name") doğrudan geçiş özellikleri için.
• catalog_name: Özelliğin Unity Kataloğu katalog adı.
• schema_name: Özelliğin Unity Kataloğu şema adı.
• entity: Toplama düzeyini (birincil anahtarlar) tanımlayan sütun adlarının listesi. Toplama özellikleri için gereklidir. Örneğin, ["user_id"] kullanıcı başına toplamalar.
• timeseries_column: Zaman penceresi toplama için kullanılan zaman damgası sütunu. Toplama özellikleri için gereklidir.
• name: İsteğe bağlı özellik adı. Atlanırsa, giriş sütunu, işlevi ve penceresinden (örneğin, amount_avg_rolling_7d) otomatik olarak oluşturulur.
• description: özelliğin isteğe bağlı açıklaması.

Döndürür: Doğrulanmış bir Özellik örneği

Harekete geçiren: Herhangi bir doğrulama başarısız olursa ValueError

delete_feature()

Unity Kataloğu'ndan bir özelliği tam adıyla siler.

FeatureEngineeringClient.delete_feature(
    full_name: str,  # Required: '<catalog>.<schema>.<feature_name>'
) -> None

fe.delete_feature(full_name="main.store.amount_sum_rolling_7d")

Bir özelliği silmeden önce, bu özelliğe başvuran modelleri veya özellik belirtimlerini kaldırın veya güncelleştirin. Özellik gerçekleştirilmişse önce gerçekleştirilmiş özelliği silin. Bkz. Gerçekleştirilmiş bir özelliği silme.

Otomatik oluşturulan adlar

Atlandığında name , otomatik olarak bir ad oluşturulur. Oluşturulan adlar şu desene uyar: {column}_{function}_{window}. Örneğin:

• price_avg_rolling_1h (1 saatlik ortalama fiyat)
• transaction_count_rolling_30d_1d (Olay zaman damgasından 1d gecikmeli 30 günlük işlem sayısı)

Desteklenen işlevler

Toplama işlevleri

ColumnSelection (geçiş)

ColumnSelection herhangi bir toplama uygulamadan kaynaktan tek bir sütun seçer. Doğrudan parametresinde function (içinde AggregationFunctiondeğil) sarmalanır. Dönüş türü kaynak şemadan çıkarılır.

ColumnSelection herhangi bir veri kaynağıyla kullanılabilir:

• DeltaTableSource: Belirli bir noktaya birleştirme yoluyla varlık anahtarı başına en son değeri döndürür (geri arama penceresi toplaması yoktur).
• StreamSource: Akıştan varlık anahtarı başına en son değeri döndürür (geri arama penceresi toplaması yoktur).
• RequestSource: Çıkarım zamanında sağlanan değeri geçirir (veya eğitim zamanında etiketlenmiş DataFrame'den ayıklanır).

from databricks.feature_engineering.entities import (
    ColumnSelection, DeltaTableSource, Feature, FieldDefinition,
    RequestSource, ScalarDataType,
)

delta_source = DeltaTableSource(
    catalog_name="main", schema_name="feature_store", table_name="transactions",
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="session_duration", data_type=ScalarDataType.DOUBLE),
    ]
)

# ColumnSelection from a Delta table
latest_amount = Feature(
    source=delta_source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    name="latest_transaction_amount",
)

# ColumnSelection from a RequestSource
session_feature = Feature(
    source=request_source,
    function=ColumnSelection("session_duration"),
    name="session_duration",
)

Örnek: toplama ve sütun seçimi özellikleri

Aşağıdaki örnek, aynı veri kaynağı üzerinde tanımlanan özellikleri gösterir.

from databricks.feature_engineering.entities import (
    AggregationFunction, Feature, Sum, Avg, ApproxCountDistinct,
    ColumnSelection, RollingWindow,
)
from datetime import timedelta

window = RollingWindow(window_duration=timedelta(days=7))

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Sum(input="amount"), window),
)

avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Avg(input="amount"), window),
)

distinct_count = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(ApproxCountDistinct(input="product_id", relativeSD=0.01), window),
)

# Column selection (no aggregation, no time window)
latest_amount = Feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="event_time",
    name="latest_amount",
)

Filtre koşulları içeren özellikler

parametresi, filter_condition toplamaları hesaplamadan önce kaynak tablodaki satırları filtrelemenize olanak tanır. Bu, verileri gruplandırma ve toplamadan önce uygulanan bir SQL WHERE yan tümcesi olarak işlev görür.

Filtreler, özellik hesaplaması için gereken verilerin üst kümesini içeren büyük kaynak tablolarla çalışırken kullanışlıdır ve bu tabloların üzerinde ayrı görünümler oluşturma gereksinimini en aza indirir.

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow, DeltaTableSource
from datetime import timedelta

# Source with filter applied at the source level
high_value_transactions = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="transactions",
    filter_condition="amount > 100",  # Only transactions over $100
)

high_value_sales = Feature(
    source=high_value_transactions,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30))),
)

# Multiple conditions
completed_orders_source = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="orders",
    filter_condition="status = 'completed' AND payment_method = 'credit_card'",
)

completed_orders = Feature(
    source=completed_orders_source,
    entity=["user_id"],
    timeseries_column="order_time",
    function=AggregationFunction(Count(input="order_id"), RollingWindow(window_duration=timedelta(days=7))),
)

# Filter on a StreamSource
from databricks.feature_engineering.entities import StreamSource

purchase_stream = StreamSource(
    full_name="main.ecommerce.transactions_stream",
    filter_condition="value.event_type = 'purchase'",
)

purchase_total = Feature(
    source=purchase_stream,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(Sum(input="value.amount"), RollingWindow(window_duration=timedelta(hours=1))),
)

Veri kaynakları

DeltaTableSource

DeltaTableSource, özelliklerin kaynak tablodan nasıl hesapılacağını tanımlamak için kullanılan kısa ömürlü bir Python nesnesidir. Yeni bir tablo oluşturmaz. Verileri okumak ve özellikleri toplamak için yapılandırmayı belirtir.

DeltaTableSource(
    catalog_name: str,                              # Required: Catalog name
    schema_name: str,                               # Required: Schema name
    table_name: str,                                # Required: Table name
    filter_condition: Optional[str] = None,         # Optional: SQL WHERE clause to filter source data
    transformation_sql: Optional[str] = None,       # Optional: SQL SELECT expression for column transformations
    dataframe_schema: Optional[str] = None,         # Required if transformation_sql is set: schema of the resulting DataFrame
)

Parametreler:

• catalog_name, schema_name, table_name: Unity Kataloğu'nda kaynak Delta tablosunu tanımlayın.
• filter_condition: Toplamadan önce uygulanan bir SQL WHERE yan tümcesi. Örnek: "status = 'completed'".
• transformation_sql: Kaynak tabloya uygulanan bir SQL SELECT ifadesi. Toplamadan önce sütunları, türetilmiş sütunları veya işlem türetilmiş sütunları yeniden adlandırmak için bunu kullanın. Atlanırsa, tüm sütunlar seçilir (*). Örnek: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: Dönüştürmelerden sonra elde edilen DataFrame'in Spark StructType JSON biçimindeki şeması (kimden df.schema.json()). Sağlanmışsa transformation_sql gereklidir. Bu, sisteme dönüştürmenizin sonucu olan sütun adlarını ve türlerini bildirir.

Hem hem de filter_conditiontransformation_sql ayarlandığında, sonuçta elde edilen sorgu şöyledir: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

Örnek: Sütun dönüştürmeleri için kullanma transformation_sql

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="raw_events",
    transformation_sql="user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time",
    filter_condition="event_type = 'purchase'",
    dataframe_schema=spark.sql(
        "SELECT user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time FROM main.analytics.raw_events LIMIT 0"
    ).schema.json(),
)

Örnek: PySpark DataFrame'den türetme transformation_sql ve dataframe_schema

Dönüştürmenizi PySpark sorgusu olarak yazabilir ve ardından sonuçta elde edilen DataFrame'den şemayı ayıklayabilirsiniz:

df = spark.sql(f"""
  SELECT user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time
  FROM main.analytics.events
  WHERE event_date >= date_sub(current_date(), 7)
  LIMIT 0
""")

# Use df.schema.json() as the dataframe_schema
source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    transformation_sql="user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time",
    filter_condition="event_date >= date_sub(current_date(), 7)",
    dataframe_schema=df.schema.json(),
)

DeltaTableSource.from_sql()

Kolaylık sağlamak için SQL sorgusundan bir DeltaTableSource oluşturabilirsiniz. yöntemi sorguyu ayrıştırarak tablo adını transformation_sql, ve filter_condition'yi otomatik olarak ayıklar.

DeltaTableSource.from_sql(
    sql: str,                           # Required: SQL SELECT query
    spark: SparkSession,                # Required: active SparkSession (for schema inference)
) -> DeltaTableSource

Yalnızca basit SELECT ... FROM ... [WHERE ...] sorgular desteklenir. Karmaşık SQL (JOIN'ler, alt sorgular, CTEs, UNION'lar) reddedilir. Karmaşık sorgular için doğrudan ve DeltaTableSourceile transformation_sql oluşturmafilter_condition.

from databricks.feature_engineering.entities import (
    AggregationFunction,
    DeltaTableSource,
    Feature,
    Sum,
    TumblingWindow,
)

source = DeltaTableSource.from_sql(
    spark=spark,
    sql=f"SELECT customer_id, event_ts, amount * 2 AS doubled_amount, amount FROM {CATALOG}.{SCHEMA}.{TABLE}",
)

feature = Feature(
    source=source,
    function=AggregationFunction(Sum(input="doubled_amount"), time_window=TumblingWindow(window_duration=timedelta(days=7))),
    entity=["customer_id"], timeseries_column="event_ts",
)

ile yinele to_dataframe()

Özellik hesaplaması için kullanılacak verilerin önizlemesini görüntülemek için kullanın source.to_dataframe() . Bu, beklenen sonuçları elde edene kadar ve filter_condition üzerinde transformation_sql yineleme yapmak için yararlıdır.

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    filter_condition="event_type = 'purchase'",
)

# Preview the filtered source data
source.to_dataframe().display()

Varlıkları anlama

Varlık sütunları, özelliklerinizin toplama düzeyini tanımlar. Bunlar tanımda Feature belirtilir, üzerinde DeltaTableSourcebelirtilmez. Varlıklar aşağıdakileri belirler:

• Veriler nasıl gruplandırılır: Özellikler, varlık değerlerinin benzersiz bileşimi başına toplanır (SQL'de olduğu gibi GROUP BY )
• Birincil anahtar yapısı: Her benzersiz varlık bileşimi bir dizi hesaplanan özellikle sonuçlanabilir

Örnek: Müşteri düzeyinde özellikler

Aşağıdaki kod, müşteri düzeyindeki özellikleri toplar (müşteri başına bir satır):

from databricks.feature_engineering.entities import DeltaTableSource

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="user_events",
)

Feature(
    source=source,
    entity=["user_id"],                # Features aggregated per user
    timeseries_column="event_time",    # Timestamp for time windows
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

Örnek: Müşteri-Mağaza düzeyi özellikleri

Özellikleri daha ayrıntılı bir düzeyde (müşteri-mağaza birleşimi başına bir satır) toplamak için birden çok varlık sütunu kullanın:

source = DeltaTableSource(
    catalog_name="main",
    schema_name="retail",
    table_name="transactions",
)

Feature(
    source=source,
    entity=["user_id", "store_id"],  # Features aggregated per user-store pair
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

Farklı toplama düzeylerinde (örneğin, müşteri düzeyinde ve müşteri-mağaza düzeyinde) özelliklere ihtiyacınız olduğunda, özellik tanımlarınızda farklı entity değerler kullanın. DeltaTableSource Aynı durum farklı varlık yapılandırmalarına sahip özellikler arasında paylaşılabilir.

StreamSource

StreamSource bir Stream'e başvurur. Stream, akış kaynağı için bağlantı, kimlik doğrulaması, şema ve alma yapılandırması içerir. Kafka için özellik tanımlarındaki sütun başvuruları, iletinin hangi bölümünün okunması gerektiğini belirtmek için veya value. ön ekine key. eklenmelidir.

StreamSource(
    full_name: str,                       # Required: Three-part Stream name (catalog.schema.stream)
    filter_condition: Optional[str],      # Optional: SQL WHERE clause applied before aggregation
)

Parametreler:

• full_name: Bir Akışın tam üç parçalı adı (örneğin, "my_catalog.my_schema.my_stream").
• filter_condition (isteğe bağlı): Nokta ön ekli sütun başvuruları kullanılarak toplamadan önce akış verilerine uygulanan bir SQL WHERE yan tümcesi (örneğin, "value.event_type = 'purchase'").

from databricks.feature_engineering.entities import StreamSource

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

RequestSource

RequestSource , önceden gerçekleştirilmiş bir tablodan aramak yerine istek yükünde çıkarım zamanında sağlanan veriler için bir şema tanımlar. Eğitim sırasında, bu sütunlar etiketli DataFrame'den ayıklanır ve öğesine create_training_setgeçirilir. Model sunma sırasında çağıranın bunları HTTP isteği yüküne dahil etmesi gerekir.

RequestSource ile ColumnSelection kullanılır (doğrudan bir değerden geçmek için). Toplama işlevlerini veya zaman pencerelerini desteklemez.

Şemayı tanımlama

Şemayı, her biri FieldDefinition bir sütun adı ve ScalarDataTypebir belirten nesnelerin listesi olarak tanımlayın:

from databricks.feature_engineering.entities import (
    FieldDefinition, RequestSource, ScalarDataType,
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="transaction_amount", data_type=ScalarDataType.DOUBLE),
        FieldDefinition(name="vendor_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_time", data_type=ScalarDataType.DATE),
    ]
)

Desteklenen veri türleri

RequestSourceiçinde tanımlanan ScalarDataTypeskaler türleri destekler: INTEGER, FLOAT, BOOLEAN, STRING, , DOUBLE, LONGTIMESTAMPDATE, SHORT. Diziler, haritalar ve yapılar gibi karmaşık türler desteklenmez.

İstek verileri nasıl sulanır?

Model imzası

Bir model, özellikler içeren log_model bir eğitim kümesi kullanılarak RequestSource günlüğe kaydedildiğinde, RequestSource sütunlar MLflow modeli imzasına gerekli girişler olarak eklenir. Bu, hizmet veren uç noktanın API şemasının, çıkarım zamanında hangi alanları çağıranların sağlaması gerektiğini yansıtdığı anlamına gelir.

Eğitim ve çıkarım API'si

create_training_set()

Belirli bir noktaya doğru özellik hesaplaması ile bir eğitim veri kümesi oluşturur. Ayrıntılar için bkz. Özellik Görünümleri ile modelleri eğitma.

FeatureEngineeringClient.create_training_set(
    df: DataFrame,                                # DataFrame with training data
    features: Optional[List[Feature]],            # List of Feature objects
    label: Union[str, List[str], None],           # Label column name(s)
    exclude_columns: Optional[List[str]] = None,  # Optional: columns to exclude
) -> TrainingSet

log_model()

Çıkarım sırasında köken izleme ve otomatik özellik araması için özellik meta verilerine sahip bir modeli günlüğe kaydeder. Ayrıntılar için bkz. Özellik Görünümleri ile modelleri eğitma.

FeatureEngineeringClient.log_model(
    model,                                    # Trained model object
    artifact_path: str,                       # Path to store model artifact
    flavor: ModuleType,                       # MLflow flavor module (e.g., mlflow.sklearn)
    training_set: TrainingSet,                # TrainingSet used for training
    registered_model_name: Optional[str],     # Optional: register model in Unity Catalog
)

score_batch()

Otomatik özellik arama ile çevrimdışı toplu çıkarım gerçekleştirir. Belirli bir noktaya doğru özellikleri hesaplamak ve eğitimle tutarlılık sağlamak için modelle birlikte depolanan özellik meta verilerini kullanır.

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model (e.g., "models:/catalog.schema.model/1")
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

Giriş DataFrame'i eğitim sırasında kullanılan varlık ve zaman aralıkları sütunlarını içermelidir. Özellikler kaynak verilerden otomatik olarak hesaplanır.

fe = FeatureEngineeringClient()

# Batch scoring with automatic feature lookup
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

Zaman aralıkları

Özellik Görünümleri, zaman penceresi tabanlı toplamalar için geri arama davranışını denetlemek için üç farklı pencere türünü destekler: yuvarlanma, yuvarlanma ve kayan.

• Sıralı pencereler olay zamanından geriye bakar. Süre ve gecikme açıkça tanımlanır.
• Atlayan pencereler sabit, örtüşmeyen zaman pencereleridir. Her veri noktası tam olarak bir pencereye aittir.
• Kayan pencereler çakışıyor, sıralı zaman pencereleri yapılandırılabilir bir slayt aralığıyla.

Aşağıdaki çizimde nasıl çalıştıkları gösterilmektedir.

Sıralı pencere

Sıralı pencereler, genellikle akış verileri üzerinde kullanılan up-totarih ve gerçek zamanlı toplamalardır. Akış işlem hatlarında, sıralı pencere yalnızca sabit uzunluktaki pencerenin içeriği değiştiğinde (örneğin, bir olay girdiğinde veya ayrıldığında) yeni bir satır yayar. Eğitim işlem hatlarında sıralı pencere özelliği kullanıldığında, belirli bir olayın zaman damgasından hemen önce sabit uzunluktaki pencere süresi kullanılarak kaynak verilerde doğru bir zaman noktası özellik hesaplaması gerçekleştirilir. Bu, çevrimiçi çevrimdışı dengesizliği veya veri sızıntısını önlemeye yardımcı olur. Zamandaki özellikler, [T − süre, T) aralığındaki olayları T birleştirir.

class RollingWindow(TimeWindow):
    window_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None

Aşağıdaki tabloda sıralı pencere parametreleri listelemektedir. Pencere başlangıç ve bitiş saatleri aşağıdaki parametreleri temel alır:

• Başlangıç saati: evaluation_time - window_duration - delay (dahil)
• Bitiş saati: evaluation_time - delay (özel)

from databricks.feature_engineering.entities import RollingWindow
from datetime import timedelta

# Look back 7 days from evaluation time
window = RollingWindow(window_duration=timedelta(days=7))

Aşağıdaki kodu kullanarak gecikmeli bir sıralı pencere tanımlayın.

# Look back 7 days, offset by 1 minute to account for data ingestion delay
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(minutes=1)
)

Sıralı pencere örnekleri

• window_duration=timedelta(days=7): Bu, geçerli değerlendirme zamanıyla biten 7 günlük bir geriye dönük inceleme penceresi oluşturur. 7. Gün 14:00'teki bir etkinlik için, 0. Gün 14:00'ten başlayarak (ancak 7. Gün 14:00 dahil olmamak üzere) 7. Gün 14:00'e kadar olan tüm etkinlikler buna dahildir.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Bu, değerlendirme zamanından 30 dakika önce biten 1 saatlik bir geri arama penceresi oluşturur. Saat 15:00'te gerçekleşen bir etkinlik için, 13:30'dan 14:30'a kadar (14:30 dahil olmadan) tüm etkinlikleri içerir. Bu, veri alımı gecikmelerini hesaba eklemek için yararlıdır.

Atlayan pencere

Sabit pencereler kullanılarak tanımlanan özellikler için toplamalar, bir kaydırma aralığı ile ilerlerken, zamanı tam olarak bölen ve örtüşmeyen pencereler üreten, önceden belirlenmiş sabit uzunluklu bir pencere üzerinden hesaplanır. Sonuç olarak, kaynaktaki her olay tam olarak bir pencereye katkıda bulunur. t zamanındaki özellikler, t veya öncesinde biten pencerelerden gelen verileri (hariç) toplar. Windows, Unix dönemiyle başlar.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

Aşağıdaki tabloda dönmeli pencerenin parametreleri listeleniyor.

from databricks.feature_engineering.entities import TumblingWindow
from datetime import timedelta

window = TumblingWindow(
    window_duration=timedelta(days=7)
)

Atlayan pencere örneği

• window_duration=timedelta(days=5): Bu, her biri 5 günlük önceden belirlenmiş sabit uzunlukta pencereler oluşturur. Örnek: Pencere #1, 0. Günü 4. Güne, Pencere #2 5. Günü 9. Güne, Pencere #3 ise 10. Günü 14. Güne yayıyor vb. Özellikle, Pencere #1, 0. Gün'de 00:00:00.00 zaman damgasıyla başlayan ve 5. Gün'de 00:00:00.00 zaman damgasına kadar olan (ancak bu 00:00:00.00 damgasına sahip olanlar hariç) tüm olayları içerir. Her olay tam olarak bir pencereye aittir.

Kayan pencere

Kayan pencereler kullanılarak tanımlanan özellikler için toplamalar, slayt aralığına göre ilerleyerek çakışan pencereler oluşturan önceden belirlenmiş bir sabit uzunluklu pencere üzerinden hesaplanır. Kaynaktaki her olay birden çok pencere için özellik toplamaya katkıda bulunabilir. t zamanındaki özellikler, t veya öncesinde biten pencerelerden gelen verileri (hariç) toplar. Windows, Unix dönemiyle başlar.

class SlidingWindow(TimeWindow):
    window_duration: datetime.timedelta
    slide_duration: datetime.timedelta

Aşağıdaki tabloda kayan pencere için parametreler listelemektedir.

from databricks.feature_engineering.entities import SlidingWindow
from datetime import timedelta

window = SlidingWindow(
    window_duration=timedelta(days=7),
    slide_duration=timedelta(days=1)
)

Kayan pencere örneği

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Bu, her seferinde 1 gün ilerleyen çakışan 5 günlük pencereler oluşturur. Örnek: Pencere #1, 0. Günü 4. Güne, Pencere #2 1. Günü 5. Güne, Pencere #3 ise 2. Günü 6. Güne yayıyor vb. Her bir pencerede başlangıç gününden bitiş günü hariç olmak üzere 00:00:00.00'dan 00:00:00.00'e kadar olan olaylar bulunur. Pencereler çakıştığı için tek bir olay birden çok pencereye ait olabilir (bu örnekte, her olay en fazla 5 farklı pencereye aittir).

Gerçekleştirme tetikleyicileri

Bir gerçekleştirme işlem hattı çalıştırıldığında denetimi tetikler. Tetikleyici türü özellik türüne bağlıdır.

CronSchedule

Toplama özellikleri (CronSchedule) için kullanınAggregationFunction. İşlem hattı, Quartz cron ifadesi tarafından tanımlanan sabit bir zamanlamaya göre çalışır.

from databricks.feature_engineering.entities import CronSchedule

trigger = CronSchedule(
    quartz_cron_expression="0 0 * * * ?",  # Hourly
    timezone_id="UTC",
)

TableTrigger

tarafından TableTriggeryedeklenen özellikler için ColumnSelection kullanınDeltaTableSource. Yukarı akış Delta tablosu yeni bir işleme aldığında işlem hattı çalışır.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

tarafından StreamingModeyedeklenen özellikler için kullanınStreamSource. İşlem hattı sürekli akış işlem hattı olarak çalışır.

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

fe = FeatureEngineeringClient()

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

streaming_feature = fe.create_feature(
    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)),
    ),
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
)

fe.materialize_features(
    features=[streaming_feature],
    online_config=OnlineStoreConfig(
        catalog_name="my_catalog",
        schema_name="my_schema",
        table_name_prefix="streaming_features_serving",
        online_store_name="feature_store_online",
    ),
    trigger=StreamingMode(),
)

Tetikleyici seçme

Farklı tetikleyici türleri gerektiren özellikleri tek materialize_features bir çağrıda gerçekleştiremezsiniz. Bunun yerine ayrı aramalar yapın.

Beta özelliklerini Genel Önizleme'ye geçirme

Özellik Görünümleri Genel Önizlemesi, Unity Kataloğu'nda ve ayrıcalıkları tarafından CREATE FEATURE yönetilen birinci sınıf READ FEATURE varlıklarını tanıtır ve 0.16.0 veya sonraki bir sürümü gerektirirdatabricks-feature-engineering. Beta sırasında oluşturulan özellikler (sürüm 0.15.0 ile) Unity Kataloğu işlevleri olarak depolanır ve tüm Genel Önizleme işlevlerini desteklemez. Uzun süreli Genel Önizleme desteği almak için beta özelliklerinizi sürüm 0.16.0 ile yeniden oluşturun. Özelliklerin silinmesi ve yeniden oluşturulması, yalnızca yeniden gerçekleştirilmesi değil.

Özellikler hakkında daha fazla bilgi için bkz. Özellik Görünümleri.

Yapmanız gerekenler

• 0.16.0 sürümüne yükseltin. Bu, Genel Önizleme özellikleri (toplu iş ve akış) için gerekli istemci sürümüdür.
• Özelliklerinizi yeniden oluşturun. Tüm Genel Önizleme işlevlerini desteklemediğinden, Beta Özellik Görünümleri silinip yeniden oluşturulmalıdır, yeniden gerçekleştirilmemelidir.
• Pencere kapanmadan önce geçiş yapın. Mevcut beta özelliklerin 22 Temmuz 2026'den önce geçirilmesi gerekir.

Beta ve Genel Önizleme özelliklerini belirleme

Genel Önizleme özellikleri Unity Kataloğu'nda, örneğin Katalog Gezgini'nde bir Özellik nesnesi olarak görünür. Beta özellikleri YAML tanımına sahip bir işlev olarak görünür. İşlev olarak temsil edilen herhangi bir özellik, geçirmeniz gereken bir beta özelliğidir.

Beta özelliklerini geçirme

Beta özelliğini geçirmenin üç bölümü vardır:

• Özelliği Genel Önizleme özelliği olarak yeniden oluşturun.
• Özelliği yeniden oluşturarak çevrimdışı ve çevrimiçi tablolarının yeni özellik altında yeniden oluşturulmasını sağlayın.
• Geçirilen özellikleri doğruladıktan sonra beta özelliklerini ve bunların gerçekleştirilmesini silin.

Özellikleri yeniden oluşturma

Beta özelliklerinizi bulmak, list_beta_feature_views kaydedilmemiş bir kopya oluşturmak ve Feature.clone() her kopyayı Genel Önizleme özelliği olarak yeniden kaydetmek için kullanınregister_feature. Kopyalama özelliğin yeniden kaydedilebilmesi için kayıt, katalog ve şemayı temizler.

Ad çakışmalarını önlemek için geçirilen özellikleri beta özelliklerden farklı bir ada veya farklı bir şemaya kaydedin. Aşağıdaki örnek, özgün şemasındaki her özelliği bir _migrated ad soneki ile yeniden kaydeder.

# Update this to the catalog whose beta Feature Views you want to migrate.
CATALOG_TO_MIGRATE = "main"

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 1. Find every beta Feature View in the catalog. Returns Feature objects,
#    scanned across all schemas in the catalog.
beta_features = fe.list_beta_feature_views(catalog_name=CATALOG_TO_MIGRATE)

# Keep each beta feature paired with its migrated counterpart for the next steps.
migrations = []
for beta_feature in beta_features:
    catalog_name, schema_name, leaf_name = beta_feature.full_name.split(".")
    # 2. Clone the feature as an unregistered copy, renamed with a "_migrated" suffix.
    cloned = beta_feature.clone(new_name=f"{leaf_name}_migrated")
    # 3. Re-register the clone as a Public Preview feature.
    migrated = fe.register_feature(
        feature=cloned,
        catalog_name=catalog_name,
        schema_name=schema_name,
    )
    migrations.append((beta_feature, migrated))

Geçirilen özellikleri yeniden gerçekleştirme

Bir beta özelliği gerçekleştirilmişse, çevrimdışı ve çevrimiçi tabloların yeni özellik altında yeniden oluşturulması için Genel Önizleme karşılık gelenini yeniden gerçekleştirin. Geçirilen özellik için çevrimdışı ve çevrimiçi mağaza yapılandırmalarını sağlayın ve tetikleyiciyi beta özelliğinin mevcut gerçekleştirilmesinden yeniden oluşturun.

from databricks.feature_engineering.entities import (
    CronSchedule,
    OfflineStoreConfig,
    OnlineStoreConfig,
    TableTrigger,
)

for beta_feature, migrated in migrations:
    # Inspect the beta feature's existing materializations to see what to rebuild and
    # to reconstruct the same trigger.
    trigger = None
    needs_offline = needs_online = False
    for mf in fe.list_materialized_features(feature_name=beta_feature.full_name):
        needs_online = needs_online or bool(mf.is_online)
        needs_offline = needs_offline or not mf.is_online
        # Rebuild the trigger from the materialized feature.
        if mf.cron_schedule_trigger is not None:
            trigger = CronSchedule(
                quartz_cron_expression=mf.cron_schedule_trigger.cron_expression,
                timezone_id="UTC",  # Materialized schedules run in UTC.
            )
        elif mf.table_trigger is not None:
            trigger = TableTrigger()
        elif mf.streaming_mode is not None:
            # Streaming features use StreamingMode, which can be reused as-is.
            trigger = mf.streaming_mode
    if not (needs_offline or needs_online):
        continue  # The beta feature was never materialized.

    catalog_name, schema_name, _ = migrated.full_name.split(".")
    fe.materialize_features(
        features=[migrated],
        offline_config=OfflineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
        )
        if needs_offline
        else None,
        online_config=OnlineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
            online_store_name="my_online_store",
        )
        if needs_online
        else None,
        trigger=trigger,
    )

Beta özelliklerini silme

Geçirilen özellikleri doğruladıktan sonra her beta özelliğinin gerçekleştirilmesini ve ardından beta özelliğinin kendisini silin.

for beta_feature, _ in migrations:
    # Delete the beta feature's materializations first.
    mfs = list(fe.list_materialized_features(feature_name=beta_feature.full_name))
    offline_mfs = [mf for mf in mfs if not mf.is_online]
    if offline_mfs:
        # Aggregation features pair an offline and online table; deleting the offline
        # materialized feature removes its paired online table too.
        for mf in offline_mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    else:
        # Online-only features (ColumnSelection, streaming) have no offline pair; delete
        # the online materialized feature directly.
        for mf in mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    # Then delete the beta feature definition.
    fe.delete_feature(full_name=beta_feature.full_name)