Referenční příručka k rozhraní API Feature Views

Important

Tato funkce je ve verzi Public Preview. Správci pracovního prostoru můžou řídit přístup k této funkci ze stránky Previews . Viz Manage Azure Databricks preview.

Řízení přístupu

Funkce jsou řídicí objekty katalogu Unity. Přístup k funkci je řízen oprávněními CREATE FEATUREkatalogu , READ FEATUREa MANAGE Unity. Úplný popis najdete v referenčních informacích k oprávněním katalogu Unity.

CREATE FEATURE — Vyžaduje se k vytvoření funkce ve schématu. create_feature a register_feature vyžadovat CREATE FEATURE u nadřazeného schématu. Podle principu nejnižších oprávnění udělte CREATE FEATURE na úrovni schématu. Můžete ho také udělit v katalogu, abyste umožnili vytváření funkcí v jakémkoli schématu v tomto katalogu.
READ FEATURE — Požadováno ke čtení funkce a jejích dat. get_feature, create_training_seta čtení materializovaných dat o funkcích pro trénování nebo obsluhu vyžadují READ FEATURE tuto funkci. READ FEATURE udělené schématu nebo katalogu platí pro všechny aktuální a budoucí funkce, které obsahuje.
MANAGE – Vyžaduje se ke správě životního cyklu a grantů funkce. Odstranění funkce pomocí delete_featurefunkce a materializace funkce s materialize_features nebo delete_materialized_feature, vyžaduje MANAGE u této funkce.

Všechny operace funkcí také vyžadují USE CATALOG nadřazený katalog a USE SCHEMA nadřazené schéma. Informace o tom, jak MANAGE a READ FEATURE jak se vztahují na materializaci, najdete v tématu Oprávnění.

Rozhraní API pro zobrazení funkcí

`Feature` konstruktor a `register_feature()`

Doporučeným přístupem je místně vytvořit Feature objekt a použít register_feature ho k zachování v katalogu Unity. Tento dvoustupňový pracovní postup umožňuje experimentovat s funkcemi (včetně create_training_set) před registrací.

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() zaregistruje místně vytvořený Feature v katalogu Unity.

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() ověří, vytvoří a okamžitě zaregistruje funkci v katalogu Unity v jednom kroku. Tuto možnost použijte, když nemusíte experimentovat s funkcí místně.

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

Parametry:

source: Zdroj dat použitý při výpočtu funkce (DeltaTableSource, StreamSourcenebo RequestSource).
function: Seskupí AggregationFunction operátor (například Sum(input="amount")), vstupní sloupec a časové okno. Nebo ColumnSelection("column_name") pro průchozí funkce.
catalog_name: Název katalogu Unity pro tuto funkci.
schema_name: Název schématu katalogu Unity pro funkci.
entity: Seznam názvů sloupců, které definují úroveň agregace (primární klíče). Vyžaduje se pro agregační funkce. Například ["user_id"] agregace na uživatele.
timeseries_column: Sloupec časového razítka použitý pro agregaci časového intervalu. Vyžaduje se pro agregační funkce.
name: Volitelný název funkce. Pokud tento parametr vynecháte, automaticky se vygeneruje ze vstupního sloupce, funkce a okna (například amount_avg_rolling_7d).
description: Volitelný popis funkce.

Vrátí: Ověřená instance funkce

Vyvolává: ValueError, pokud selže jakékoli ověření

`delete_feature()`

Odstraní funkci z katalogu Unity podle jeho plně kvalifikovaného názvu.

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

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

Před odstraněním funkce odeberte nebo aktualizujte všechny modely nebo specifikace funkcí, které na ni odkazují. Pokud je tato funkce materializovaná, nejprve odstraňte materializovanou funkci. Viz Postup odstranění materializované funkce.

Automaticky generované názvy

Pokud name tento parametr vynecháte, automaticky se vygeneruje název. Vygenerované názvy se řídí vzorem: {column}_{function}_{window}. Příklady:

price_avg_rolling_1h (1hodinová průměrná cena)
transaction_count_rolling_30d_1d (30denní počet transakcí s 1d zpožděním od časového razítka události)

Podporované funkce

Agregační funkce

Note

Agregační funkce jsou zabalené společně AggregationFunction s časovým oknem, jak je popsáno v časových oknech. Každá funkce přebírá input parametr určující zdrojový sloupec, který se má agregovat.

Function	Popis	Příklad případu použití
`Sum(input="column")`	Součet hodnot	Denní využití aplikace pro jednotlivé uživatele v minutách
`Avg(input="column")`	Průměr hodnot	Střední částka transakce
`Count(input="column")`	Počet záznamů	Počet přihlášení na jednoho uživatele
`Min(input="column")`	Minimální hodnota	Nejnižší zaznamenaná srdeční frekvence nositelným zařízením
`Max(input="column")`	Maximální hodnota	Nejvyšší částka transakce na relaci
`StddevPop(input="column")`	Směrodatná odchylka základního souboru	Variabilita denních objemů transakcí ve všech zákaznících
`StddevSamp(input="column")`	Vzorová směrodatná odchylka	Proměnlivost prokliků reklamních kampaní
`VarPop(input="column")`	Rozptyl populace	Šíření čtení snímačů pro zařízení IoT v továrně
`VarSamp(input="column")`	Výběrový rozptyl	Rozložení hodnocení filmů ve vzorkované skupině
`ApproxCountDistinct(input="column", relativeSD=0.05)`	Přibližný jedinečný počet	Jedinečný počet zakoupených položek
`ApproxPercentile(input="column", percentile=0.95, accuracy=100)`	Přibližný percentil	Latence odpovědi p95
`First(input="column")`	První hodnota	Časové razítko prvního přihlášení
`Last(input="column")`	Poslední hodnota	Poslední nákupní částka

`ColumnSelection` (průchozí)

ColumnSelection vybere jeden sloupec ze zdroje bez použití jakékoli agregace. Zabalí se přímo do parametru function (ne uvnitř AggregationFunction). Návratový typ je odvozen ze zdrojového schématu.

Function	Popis	Příklad případu použití
`ColumnSelection("col")`	Nejnovější hodnota sloupce (bez agregace)	Kategorie posledního dodavatele, předávací pole požadavku

ColumnSelection lze použít s jakýmkoli zdrojem dat:

DeltaTableSource: Vrátí nejnovější hodnotu klíče entity prostřednictvím spojení k určitému bodu v čase (bez agregace okna zpětného vyhledávání).
StreamSource: Vrátí nejnovější hodnotu klíče entity ze streamu (bez agregace okna zpětného vyhledávání).
RequestSource: Předává hodnotu zadanou v době odvozování (nebo extrahuje z označeného datového rámce v době trénování).

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",
)

Příklad: Funkce výběru agregace a sloupce

Následující příklad ukazuje funkce definované ve stejném zdroji dat.

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",
)

Funkce s podmínkami filtru

Tento filter_condition parametr umožňuje filtrovat řádky ze zdrojové tabulky před výpočetními agregacemi. Tato funkce funguje jako klauzule SQL WHERE , která se používá před seskupením a agregací dat.

Note

filter_condition filtruje řádky před agregací, jako je klauzule SQL WHERE použitá před GROUP BY. Nemění členitost, která je vždy definována v entity definici funkce.

Filtry jsou užitečné při práci s velkými zdrojovými tabulkami, které obsahují nadmnožinu dat potřebných pro výpočet funkcí, a minimalizují potřebu vytváření samostatných zobrazení nad těmito tabulkami.

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))),
)

Zdroje dat

`DeltaTableSource`

DeltaTableSource je dočasný objekt Python, který slouží k definování způsobu výpočtu funkcí ze zdrojové tabulky. Nevytvoří novou tabulku. Určuje konfiguraci pro čtení dat a agregaci funkcí.

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
)

Parametry:

catalog_name, , schema_nametable_name: Identifikujte zdrojovou tabulku Delta v katalogu Unity.
filter_condition: Klauzule SQL WHERE použitá před agregací. Příklad: "status = 'completed'".
transformation_sql: Výraz SQL SELECT použitý na zdrojovou tabulku. Tento postup slouží k přejmenování sloupců, typů přetypování nebo výpočetních odvozených sloupců před agregací. Pokud tento argument vynecháte, budou vybrány všechny sloupce (*). Příklad: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
dataframe_schema: Schéma výsledného datového rámce po transformacích ve formátu JSON Sparku StructType (z df.schema.json()). Vyžaduje se, pokud transformation_sql je k dispozici. Tím systému sdělíte názvy a typy sloupců, které jsou výsledkem vaší transformace.

Při nastavení obou filter_condition a transformation_sql nastavení je výsledný dotaz: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

Note

Hodnota timeseries_column (zadaná v definici funkce, nikoli on DeltaTableSource) musí být typu TimestampType nebo DateType. Celočíselné typy můžou fungovat, ale mohou způsobit ztrátu přesnosti pro agregace časových intervalů.

Příklad: Použití transformation_sql pro transformace sloupců

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(),
)

Příklad: Odvození transformation_sql datového rámce PySpark a dataframe_schema z něj

Transformaci můžete napsat jako dotaz PySpark a pak extrahovat schéma z výsledného datového rámce:

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(),
)

Note

transformation_sql podporuje pouze výrazy s řádky (přejmenování sloupců, přetypování, aritmetika). Agregační funkce jako COUNT(*) nebo SUM() nejsou podporované. Použijte AggregationFunction místo toho definici funkce.

`DeltaTableSource.from_sql()`

Jako pohodlí můžete vytvořit DeltaTableSource dotaz SQL. Metoda parsuje dotaz tak, aby automaticky extrahovali název tabulky, transformation_sqla filter_condition.

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

Podporují se jenom jednoduché SELECT ... FROM ... [WHERE ...] dotazy. Komplexní kód SQL (JOINs, poddotazy, CTEs, UNIONs) je odmítnut. U složitých dotazů vytvořte DeltaTableSource přímo pomocí transformation_sql a filter_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",
)

Iterovat pomocí `to_dataframe()`

Slouží source.to_dataframe() k zobrazení náhledu dat, která se použijí pro výpočty funkcí. To je užitečné pro iteraci filter_condition a transformation_sql dokud nevygenerují očekávané výsledky.

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()

Principy entit

Sloupce entit definují úroveň agregace pro vaše funkce. Jsou zadány v Feature definici, nikoli v DeltaTableSource. Entity určují:

Seskupování dat: Funkce se agregují na jedinečnou kombinaci hodnot entit (podobně jako GROUP BY v SQL).
Struktura primárního klíče: Každá jedinečná kombinace entit má za následek jeden řádek vypočítaných funkcí.

Příklad: Funkce na úrovni zákazníka

Následující kód agreguje funkce na úrovni zákazníka (jeden řádek na zákazníka):

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))),
)

Příklad: Funkce na úrovni zákaznického úložiště

Pokud chcete agregovat funkce na podrobnější úrovni (jeden řádek na kombinaci zákaznického úložiště), použijte více sloupců entit:

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))),
)

Pokud potřebujete funkce na různých úrovních agregace (například na úrovni zákazníka a na úrovni úložiště zákazníků), použijte v definicích funkcí různé entity hodnoty. Totéž DeltaTableSource se dá sdílet mezi funkcemi s různými konfiguracemi entit.

`StreamSource`

StreamSource odkazuje na Stream. Stream obsahuje konfiguraci připojení, ověřování, schématu a příjmu dat pro zdroj streamování. V případě Kafka musí být odkazy na sloupce v definicích funkcí předponou value. nebo key. určit, která část zprávy se má přečíst.

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

Parametry:

full_name: Úplný třídílný název streamu (například "my_catalog.my_schema.my_stream").
filter_condition (volitelné): Klauzule SQL WHERE použitá pro streamování dat před agregací pomocí odkazů na sloupce s předponou tečky (například "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 definuje schéma pro data, která jsou poskytována v době odvozování v datové části požadavku, a nikoli z předem materializované tabulky. Během trénování se tyto sloupce extrahují z označeného datového rámce předaného create_training_setdo . Během obsluhy modelu musí volající zahrnout do datové části požadavku HTTP.

RequestSource se používá s ColumnSelection (k předání hodnoty přímo). Nepodporuje agregační funkce ani časová okna.

Definování schématu

Definujte schéma jako seznam FieldDefinition objektů, přičemž každý určuje název sloupce a ScalarDataType:

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),
    ]
)

Podporované datové typy

RequestSourcepodporuje skalární typy definované v : ScalarDataTypeINTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, , DATE. SHORT Komplexní typy, jako jsou pole, mapy a struktury, nejsou podporovány.

Jak jsou data požadavku hydratovaná

Kontext	Behavior
Školení (`create_training_set`)	Sloupce se extrahují z označeného datového rámce. Typy se ověřují proti deklarovanému schématu. Neshody vyvolávají chybu (bez implicitního přetypování).
Obsluha (koncový bod modelu)	Sloupce se natahují z `dataframe_records` požadavku HTTP nebo `dataframe_split` z něj. Hodnoty JSON se přetypují na deklarované typy (např. číslo JSON → `DOUBLE`).

Podpis modelu

Při protokolování modelu pomocí log_model trénovací sady, která obsahuje RequestSource funkce, RequestSource se sloupce přidají do podpisu modelu MLflow jako požadované vstupy. To znamená, že schéma rozhraní API koncového bodu obsluhy odráží, která pole volající musí poskytnout v době odvozování.

Rozhraní API pro trénování a odvozování

`create_training_set()`

Vytvoří trénovací datovou sadu se správnými výpočty funkcí k určitému bodu v čase. Podrobnosti najdete v tématu Trénování modelů pomocí zobrazení funkcí.

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()`

Zaznamená model s metadaty funkcí pro sledování rodokmenu a automatické vyhledávání funkcí během odvozování. Podrobnosti najdete v tématu Trénování modelů pomocí zobrazení funkcí.

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()`

Provádí odvozování dávek offline pomocí automatického vyhledávání funkcí. Používá metadata funkcí uložená s modelem k výpočtu správných funkcí k určitému bodu v čase a zajišťují konzistenci s trénováním.

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

Vstupní datový rámec musí obsahovat sloupce entity a timeseries použité během trénování. Funkce se automaticky počítají ze zdrojových dat.

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()

Časová okna

Zobrazení funkcí podporují tři různé typy oken pro řízení chování zpětného vyhledávání pro agregace založené na časových osách: posuvné, přeskakující a posuvné.

Vrácení oken se vrátí zpět z času události. Doba trvání a zpoždění jsou explicitně definovány.
Okna typu 'tumbling' jsou pevná, nepřekrývající se časová okna. Každý datový bod patří přesně do jednoho okna.
Posuvná okna se překrývají, jsou to posunovací časová okna s konfigurovatelným intervalem posunu.

Následující obrázek ukazuje, jak fungují.

Posuvná, přeskakující a posuvná okna zpětného vzhledu.

Posuvné okno

Note

RollingWindow byla dříve pojmenována ContinuousWindow. Pokud migrujete ze starší verze sady SDK, aktualizujte importy odpovídajícím způsobem.

Posuvná okna jsou up-to-date a agregace v reálném čase, které se obvykle používají pro streamovaná data. V kanálech streamování vygeneruje posuvné okno nový řádek pouze v případě, že se změní obsah okna s pevnou délkou, například když událost vstoupí nebo opustí. Při použití funkce posuvného okna v trénovacích kanálech se na zdrojových datech provádí přesný výpočet funkce k určitému bodu v čase pomocí doby trvání časového razítka konkrétní události s pevnou délkou bezprostředně před časovým razítkem konkrétní události. To pomáhá zabránit nesouladu mezi online a offline daty nebo úniku dat. Funkce v čase T agregují události od [T − doba trvání, T).

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

Následující tabulka uvádí parametry pro posuvné okno. Počáteční a koncové časy okna jsou založené na těchto parametrech:

Čas zahájení: evaluation_time - window_duration - delay (včetně)
Koncový čas: evaluation_time - delay (výhradní)

Parameter Omezení

delay (volitelné) Musí být ≥ 0 (posune okno zpět v čase od časového razítka vyhodnocení). Používá delay se k zohlednění jakéhokoli zpoždění systému mezi časem vytvoření události a časovým razítkem události, aby se zabránilo budoucímu úniku událostí do trénovacích datových sad. Pokud je například mezi časem vytvoření událostí zpoždění o jednu minutu a tyto události se nakonec přidají do zdrojové tabulky, kde jsou přiřazeny časové razítko, bude zpoždění timedelta(minutes=1).

window_duration Musí být > 0.

Parameter	Omezení
`delay` (volitelné)	Musí být ≥ 0 (posune okno zpět v čase od časového razítka vyhodnocení). Používá `delay` se k zohlednění jakéhokoli zpoždění systému mezi časem vytvoření události a časovým razítkem události, aby se zabránilo budoucímu úniku událostí do trénovacích datových sad. Pokud je například mezi časem vytvoření událostí zpoždění o jednu minutu a tyto události se nakonec přidají do zdrojové tabulky, kde jsou přiřazeny časové razítko, bude zpoždění `timedelta(minutes=1)`.
`window_duration`	Musí být > 0.

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))

Pomocí následujícího kódu definujte posuvné okno se zpožděním.

# 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)
)

Příklady posuvné okno

window_duration=timedelta(days=7): Vytvoří se 7denní okno pro zpětný pohled, které končí v okamžiku aktuálního vyhodnocení. U události v 2:00 dne 7 se to týká všech událostí od 2:00 v den 0 až do (ale ne) 2:00 dne 7.
window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Vytvoří se 1hodinový interval zpětného vyhledávání, který končí 30 minut před časem vyhodnocení. U události v 17:00 to zahrnuje všechny události od 13:30 do (ale ne) 2:30 pm. To je užitečné při zohlednění zpoždění příjmu dat.

Přeskakující okno

U funkcí definovaných pomocí pohybujících se oken se agregace počítají přes předem určené okno s pevnou délkou, které postupuje posuvným intervalem a vytváří nepřekrývající se okna, která plně rozdělují čas. V důsledku toho každá událost ve zdroji patří přesně do jednoho okna. Funkce v čase t agregují data z oken končících na nebo před t (výhradní). Systém Windows se spouští v unixové epochě.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

Následující tabulka uvádí parametry pro přeskakující okno.

Parameter	Omezení
`window_duration`	Musí být > 0.

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

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

Příklad přeskakujícího okna

window_duration=timedelta(days=5): Vytvoří se předem určená okna s pevnou délkou 5 dní. Příklad: Okno č. 1 zahrnuje den 0 až den 4, okno #2 zahrnuje 5. den až 9, okno č. 3 zahrnuje 10. den a další. Konkrétně okno č. 1 obsahuje všechny události s časovými razítky začínajícími dne 0 na 00:00:00.00 až (ale ne včetně) jakékoli události s časovým razítkem 00:00:00.00 dne 5. Každá událost patří přesně do jednoho okna.

Posuvné okno

U funkcí definovaných pomocí posuvných oken se agregace počítají přes předem určené okno s pevnou délkou, které postupuje posunovým intervalem a okna se překrývají. Každá událost ve zdroji může přispět k agregaci funkcí pro více oken. Funkce v čase t agregují data z oken končících na nebo před t (výhradní). Systém Windows se spouští v unixové epochě.

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

Následující tabulka uvádí parametry posuvného okna.

Parameter	Omezení
`window_duration`	Musí být > 0.
`slide_duration`	Musí být > 0 a <`window_duration`

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

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

Příklad posuvného okna

window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Tím se vytvoří překrývající se 5denní intervaly, které se posunou každý den o 1 den. Příklad: Okno č. 1 zahrnuje den 0 až den 4, okno č. 2 zahrnuje 1. den až den 5, okno č. 3 zahrnuje 2. den až den 6 atd. Každé okno obsahuje události od 00:00:00.00 na počátku až po (ale nepokryje) 00:00:00.00 na konci dne. Vzhledem k tomu, že se okna překrývají, může jedna událost patřit do více oken (v tomto příkladu každá událost patří až do 5 různých oken).

Aktivační události materializace

Aktivuje řízení při spuštění kanálu materializace. Typ triggeru závisí na typu funkce.

`CronSchedule`

Používá se CronSchedule pro funkce agregace (AggregationFunction). Kanál běží podle pevného plánu definovaného výrazem Quartz cron.

from databricks.feature_engineering.entities import CronSchedule

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

`TableTrigger`

Používá se TableTrigger pro ColumnSelection funkce zálohované funkcí DeltaTableSource. Kanál se spustí vždy, když nadřazená tabulka Delta obdrží nové potvrzení.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

`StreamingMode`

Používá se StreamingMode pro funkce zálohované funkcí StreamSource. Kanál běží jako kanál průběžného streamování.

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(),
)

Výběr triggeru

Typ funkce	Spouštěč	Když se spustí
Agregace (`AggregationFunction`) z `DeltaTableSource`	`CronSchedule`	Na pevném plánu cron
`ColumnSelection` (z `DeltaTableSource`)	`TableTrigger`	Při každém potvrzení zdrojové tabulky
Funkce z `StreamSource`	`StreamingMode`	Průběžné streamování

V jednom materialize_features volání nelze materializovat funkce, které vyžadují různé typy aktivačních událostí. Místo toho proveďte samostatná volání.

Migrace beta funkcí do verze Public Preview

Zobrazení funkcí Ve verzi Public Preview zavádí entity funkcí první třídy v katalogu Unity, které se řídí oprávněními CREATE FEATURE a READ FEATURE které vyžadují databricks-feature-engineering verzi 0.16.0 nebo novější. Funkce vytvořené během beta verze (s verzí 0.15.0) se ukládají jako funkce katalogu Unity a nepodporují všechny funkce verze Public Preview. Pokud chcete získat dlouhodobou podporu verze Public Preview, znovu vytvořte beta funkce s verzí 0.16.0. Funkce musí být odstraněny a znovu vytvořeny, nejen znovu materializovány.

Další informace ofunkcích

Co potřebujete udělat

Upgradujte na verzi 0.16.0. Toto je požadovaná verze klienta pro funkce Public Preview (dávkové a streamování).
Znovu vytvořte funkce. Beta zobrazení funkcí musí být odstraněna a znovu vytvořena, ne znovu materializována, protože nepodporují všechny funkce Public Preview.
Před zavření okna proveďte migraci. Stávající beta funkce musí být migrovány před 22. červencem 2026.

Identifikace funkcí beta verze a verze Public Preview

Funkce Public Preview se v katalogu Unity zobrazují jako objekt funkce , například v Průzkumníku katalogu. Funkce beta verze se zobrazují jako funkce s definicí YAML. Každá funkce reprezentovaná jako funkce je beta funkce, kterou potřebujete migrovat.

Migrace beta funkcí

Migrace funkce beta verze má tři části:

Znovu vytvořte funkci jako funkci Public Preview.
Znovu materializovat tuto funkci, aby se její offline a online tabulky znovu sestavily v rámci nové funkce.
Po ověření migrovaných funkcí odstraňte beta funkce a jejich materializace.

Opětovné vytvoření funkcí

Umožňuje list_beta_feature_views najít beta funkce, Feature.clone() vytvořit neregistrovanou kopii a register_feature znovu zaregistrovat každou kopii jako funkci Public Preview. Klonování vymaže registraci, katalog a schéma, aby bylo možné funkci znovu zaregistrovat.

Aby nedocházelo ke kolizím názvů, zaregistrujte migrované funkce s jiným názvem nebo v jiném schématu než beta funkce. Následující příklad znovu zaregistruje každou funkci v původním schématu s příponou _migrated názvu.

# 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))

Opětovné materializace migrovaných funkcí

Pokud byla beta funkce materializována, znovu materializujte její protějšek ve verzi Public Preview, aby se jeho offline a online tabulky znovu sestavily pod novou funkcí. Zadejte konfigurace offline a online úložiště pro migrovanou funkci a rekonstruujte trigger z existující materializace funkce beta.

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,
    )

Note

Materializace jednotlivých funkcí ve vlastním materialize_features volání vytvoří samostatný kanál. Pokud chcete snížit náklady na výpočetní prostředky, seskupte funkce, které sdílejí offline a online cíl, a aktivujte je do jednoho materialize_features volání tím, že je předáte dohromady features.

Odstranění beta funkcí

Výstraha

Funkce beta verze a jejich materializace odstraňte až po ověření správnosti migrovaných funkcí a jejich materializovaných dat. Odstranění je nevratné.

Po ověření migrovaných funkcí odstraňte materializace jednotlivých beta funkcí a pak samotnou beta funkci.

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)

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-14

Referenční příručka k rozhraní API Feature Views

Řízení přístupu

Rozhraní API pro zobrazení funkcí

Feature konstruktor a register_feature()

create_feature()

delete_feature()

Automaticky generované názvy

Podporované funkce

Agregační funkce

ColumnSelection (průchozí)

Příklad: Funkce výběru agregace a sloupce

Funkce s podmínkami filtru

Zdroje dat

DeltaTableSource

DeltaTableSource.from_sql()

Iterovat pomocí to_dataframe()

Principy entit

StreamSource

RequestSource

Definování schématu

Podporované datové typy

Jak jsou data požadavku hydratovaná

Podpis modelu

Rozhraní API pro trénování a odvozování

create_training_set()

log_model()

score_batch()

Časová okna

Posuvné okno

Příklady posuvné okno

Přeskakující okno

Příklad přeskakujícího okna

Posuvné okno

Příklad posuvného okna

Aktivační události materializace

CronSchedule

TableTrigger

StreamingMode

Výběr triggeru

Migrace beta funkcí do verze Public Preview

Co potřebujete udělat

Identifikace funkcí beta verze a verze Public Preview

Migrace beta funkcí

Opětovné vytvoření funkcí

Opětovné materializace migrovaných funkcí

Odstranění beta funkcí

Váš názor

Další materiály

`Feature` konstruktor a `register_feature()`

`create_feature()`

`delete_feature()`

`ColumnSelection` (průchozí)

`DeltaTableSource`

`DeltaTableSource.from_sql()`

Iterovat pomocí `to_dataframe()`

`StreamSource`

`RequestSource`

`create_training_set()`

`log_model()`

`score_batch()`

`CronSchedule`

`TableTrigger`

`StreamingMode`