Dokumentacja referencyjna interfejsu API Feature Views

Kontrola dostępu

Funkcje są zarządzalnymi obiektami wykazu aparatu Unity. Dostęp do funkcji jest kontrolowany przez CREATE FEATUREuprawnienia katalogu , READ FEATUREi MANAGE aparatu Unity. Aby uzyskać pełne opisy, zobacz Dokumentacja uprawnień wykazu aparatu Unity.

• CREATE FEATURE: Wymagane do stworzenia cechy w schemiacie. create_feature i register_feature wymagaj CREATE FEATURE w schemacie nadrzędnym. Zgodnie z zasadą najniższych uprawnień przyznaj CREATE FEATURE na poziomie schematu; możesz również przyznać go w wykazie, aby umożliwić tworzenie funkcji w dowolnym schemacie w tym wykazie.
• READ FEATURE: Wymagane do odczytu metadanych funkcji. get_feature, create_training_set, oraz list_materialized_features wymagają READ FEATURE na cechie. Ten przywilej nie daje dostępu do danych cech w tabelach źródłowych lub materializowanych wyników. Aby odczytać te dane do szkolenia lub serwisowania, musisz mieć SELECT je również w odpowiednich tabelach. READ FEATURE przyznane w schemacie lub wykazie ma zastosowanie do wszystkich bieżących i przyszłych funkcji, które zawiera.
• MANAGE: Wymagany do zarządzania cyklem życia i grantami funkcji. Usunięcie cechy z , delete_featurei materializacja cechy z materialize_features, wymagają MANAGE na tej cechie. Usunięcie zmaterializowanej cechy nie delete_materialized_feature jest regulowane przez MANAGE: tylko twórca zmaterializowanej cechy może ją usunąć.

Wszystkie operacje funkcji wymagają USE CATALOG również w katalogu nadrzędnym i USE SCHEMA w schemacie nadrzędnym. Aby dowiedzieć się, jak MANAGE i READ FEATURE zastosować do materializacji, zobacz Uprawnienia.

Interfejs API widoku funkcji

Feature konstruktor i register_feature()

Zalecanym podejściem Feature jest utworzenie obiektu lokalnie i użycie register_feature go do utrwalania go w wykazie aparatu Unity. Ten dwuetapowy przepływ pracy umożliwia eksperymentowanie z funkcjami (w tym create_training_set) przed ich zarejestrowaniem.

Feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, RequestSource, or FeatureViewSource
    function: Union[AggregationFunction, ColumnSelection, CustomUDF], # Required for all features
    entity: Optional[List[str]] = None,                    # Required for DeltaTableSource and StreamSource
    timeseries_column: Optional[str] = None,               # Required for DeltaTableSource and StreamSource
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
)

FeatureEngineeringClient.register_feature() rejestruje lokalnie skonstruowany Feature w wykazie aparatu 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() weryfikuje, konstruuje i natychmiast rejestruje funkcję w wykazie aparatu Unity w jednym kroku. Użyj tej funkcji, jeśli nie musisz najpierw eksperymentować z funkcją lokalnie.

FeatureEngineeringClient.create_feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, RequestSource, or FeatureViewSource
    function: Union[AggregationFunction, ColumnSelection, CustomUDF], # Required for all features
    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 DeltaTableSource and StreamSource
    timeseries_column: Optional[str] = None,               # Required for DeltaTableSource and StreamSource
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

Parametry:

• source: Źródło danych używane w obliczaniu cech (DeltaTableSource, StreamSource, RequestSource, lub FeatureViewSource).
• function: An AggregationFunction , który łączy operator i okno czasowe, ColumnSelection("column_name") dla cech przechodzących lub CustomUDF dla transformacji wierszowych. Zobacz Obsługiwane funkcje dla kompatybilnych typów źródeł.
• catalog_name: nazwa wykazu wykazu aparatu Unity dla funkcji.
• schema_name: nazwa schematu wykazu aparatu Unity dla funkcji.
• entity: Lista nazw kolumn definiujących klucze agregacji lub wyszukiwania (klucze podstawowe). Wymagane dla DeltaTableSource i StreamSource. Na przykład ["user_id"] agreguje lub wyszukuje poszczególnych użytkowników. Pominąć dla RequestSource i FeatureViewSource.
• timeseries_column: kolumna znacznika czasu używana na potrzeby agregacji przedziału czasu lub wyboru najnowszej wartości. Wymagane dla DeltaTableSource i StreamSource. Pominąć dla RequestSource i FeatureViewSource.
• name: opcjonalna nazwa funkcji. Jeśli pominięto, automatycznie wygenerowane z kolumny wejściowej, funkcji i okna (na przykład amount_avg_rolling_7d).
• description: opcjonalny opis funkcji.

Zwraca: Zweryfikowane wystąpienie funkcji

Zgłasza: ValueError, jeśli sprawdzanie poprawności nie powiedzie się

delete_feature()

Usuwa funkcję z wykazu aparatu Unity przy użyciu w pełni kwalifikowanej nazwy.

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

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

Przed usunięciem funkcji usuń lub zaktualizuj wszystkie modele lub specyfikacje funkcji odwołujące się do tej funkcji. Funkcji nie można usunąć, dopóki wciąż ma zmaterializowane funkcje. Najpierw usuń zmaterializowane cechy, a potem sam element. Zobacz Jak usunąć zmaterializowaną funkcję.

Nazwy generowane automatycznie

Po name pominięciu nazwa jest generowana automatycznie. Wygenerowane nazwy są zgodne ze wzorcem: {column}_{function}_{window}. Przykład:

• price_avg_rolling_1h (średnia cena 1-godzinna)
• transaction_count_rolling_30d_1d (30-dniowa liczba transakcji z 1d opóźnieniem od znacznika czasu zdarzenia)

Obsługiwane funkcje

Funkcje agregacji

FirstN, LastN, , oraz LastDistinct używają cech timeseries_column do porządkowania wierszy wejściowych i zwracania tablicy zawierającej wartości do wartości nFirstDistinct. Parametr musi n być dodatnią liczbą całkowitą. FirstN oraz FirstDistinct wybierać wartości od najwcześniejszego do najpóźniejszego. LastN oraz LastDistinct wybierz wartości od najpóźniejszej do najwcześniejszej i zwróć wybrane wartości w kolejności znaczników czasu. FirstDistinct oraz usuwanie LastDistinct wartości zduplikowanych podczas wybierania wartości w tym kierunku.

Na przykład, jeśli wiersze źródłowe dla danej jednostki są uporządkowane przez event_time , ["A", "A", "B", "C", "B", "B"]następujące funkcje zwracają:

FirstN, LastN, FirstDistinct, oraz LastDistinct wymagają databricks-feature-engineering wersji 0.17.0 lub nowzej.

CustomUDF

CustomUDFdo każdego wiersza stosuje zarejestrowaną funkcję użytkownika Python (UDF) w Unity Catalog. Użyj go do transformacji wejść żądań lub łączenia wartości cech. Nie agreguje wierszy ani nie definiuje okna czasowego.

CustomUDF(
    function_name="main.ecommerce.log_amount_udf",
    input_bindings={"amount": "transaction_amount"},
)

input_bindings mapuje każdą nazwę parametru UDF na wejście. Dla RequestSource, wejście to nazwa kolumny źródłowej. Dla FeatureViewSource, jest to odniesienie cechy upstream. Przypisz każdy parametr UDF, w tym parametry z domyślnymi ustawieniami. Typy wejściowe muszą dokładnie odpowiadać typom parametrów UDF, bez ukrytych liczbowych castów. Używaj skalarnych typów wejściowych i zwrotów.

Funkcje wspierane CustomUDF przez Deltę nie mogą być realizowane ani udostępniane online. Aby przekształcić wartości cech opartych na tabelach do trenowania i serwowania, zdefiniuj agregację lub cechę wyboru kolumn opartą na Delta i odwołuj się do niej przez FeatureViewSource.

CustomUDF nie jest wspierany przez StreamSource. Aby przekształcić wyjście funkcji streamingowej, odwołaj się do tej cechy przez FeatureViewSource.

CustomUDF z wymaga RequestSourcedatabricks-feature-engineering wersji 0.17.0 lub nowzej.

Aby użyć CustomUDF, potrzebujesz EXECUTE uprawnień na UDF, uprawnienia USE CATALOG na jego katalogu nadrzędnym oraz uprawnienia USE SCHEMA na schemacie nadrzędnym.

Poniższy przykład wykorzystuje NumPy do obliczenia log(1 + amount), zmniejszając skalę dużych kwot transakcji. Uruchom go na serwerowym komputerze z włączonymi niestandardowymi zależnościami UDF . Schemat main.ecommerce musi istnieć.

spark.sql("""
CREATE OR REPLACE FUNCTION main.ecommerce.log_amount_udf(amount DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
ENVIRONMENT (
  dependencies = '["numpy==1.26.4"]',
  environment_version = '5'
)
AS $$
import numpy as np

if amount is None or not np.isfinite(amount) or amount < 0:
    return None
return float(np.log1p(amount))
$$
""")

Zarejestruj funkcję, która wiąże kolumnę transaction_amount żądania z parametrem amountUDF :

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

fe = FeatureEngineeringClient()

log_transaction_amount = fe.create_feature(
    source=RequestSource(
        schema=[
            FieldDefinition(name="transaction_amount", data_type=ScalarDataType.DOUBLE),
        ]
    ),
    function=CustomUDF(
        function_name="main.ecommerce.log_amount_udf",
        input_bindings={"amount": "transaction_amount"},
    ),
    catalog_name="main",
    schema_name="ecommerce",
    name="log_transaction_amount",
)

UDF ENVIRONMENT konfiguruje zależności do obliczeń offline. Do podawania online deklaruj paczki w lub create_feature_spec(extra_pip_requirements=...)log_model(extra_pip_requirements=...). Nie są one automatycznie kopiowane z UDF. Zobacz Feature Serving dependences and model dependences.

CustomUDF nie można zmaterializować funkcji. Na żądanie i funkcjonalne UDF działają na żądanie podczas szkoleń i obsługi. Każdy UDF w łańcuchu zależności dodaje obliczenia, więc funkcje i łańcuchy pozostają małe. UDF muszą obsługiwać brakujące wejścia, które mogą być None offline lub NaN online.

Aby uzyskać wskazówki dotyczące obsługi brakujących wartości, zobacz: Jak radzić sobie z brakującymi wartościami cech.

ColumnSelection (przejście)

ColumnSelection wybiera jedną kolumnę ze źródła bez stosowania żadnej agregacji. Jest on owinięty bezpośrednio w parametrze function (a nie wewnątrz AggregationFunction). Typ zwrotu jest wywnioskowany ze schematu źródłowego podczas tworzenia cechy. Zmiana typu kolumny źródłowej po utworzeniu funkcji nie jest obsługiwana.

ColumnSelection Wspierają następujące źródła danych:

• DeltaTableSource: zwraca najnowszą wartość na klucz jednostki za pośrednictwem sprzężenia do punktu w czasie (bez agregacji okna wyszukiwania).
• StreamSource: Zwraca najnowszą wartość na klucz encji ze strumienia (brak agregacji okna wstecznego).
• RequestSource: przechodzi przez wartość podaną w czasie wnioskowania (lub wyodrębniony z oznaczonej ramki danych w czasie trenowania).

Dla , DeltaTableSourceColumnSelection cechy obsługują filter_condition i transformation_sql, zastosowane przed wyborem najnowszej wartości, tak samo jak cechy agregacji.

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

Przykład: funkcje agregacji i wyboru kolumn

W poniższym przykładzie przedstawiono funkcje zdefiniowane w tym samym źródle danych.

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

Funkcje z warunkami filtrowania

Parametr pozwala filter_condition filtrować wiersze z tabeli źródłowej przed obliczeniem agregacji lub wybraniem najnowszej wartości kolumny. Ta funkcja jest klauzulą SQL WHERE , która jest stosowana przed grupowaniem i agregowaniem danych.

Filtry są przydatne podczas pracy z dużymi tabelami źródłowymi, które zawierają nadzbiór danych potrzebnych do obliczeń funkcji, i zminimalizować potrzebę tworzenia oddzielnych widoków na podstawie tych tabel.

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

Źródła danych

DeltaTableSource

DeltaTableSource jest efemerycznym obiektem Python używanym do definiowania sposobu obliczania funkcji z tabeli źródłowej. Nie tworzy nowej tabeli. Określa konfigurację odczytywania danych i agregowania funkcji.

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
    lateness: Optional[SourceLateness] = None,       # Optional: Expected source settling time
)

Parametry:

• catalog_name, , schema_nametable_name: zidentyfikuj źródłową tabelę delty w wykazie aparatu Unity.
• filter_condition: Klauzula SQL WHERE stosowana przed agregacją lub wyborem kolumn. Przykład: "status = 'completed'".
• transformation_sql: wyrażenie SQL SELECT zastosowane do tabeli źródłowej. Użyj tego do zmiany nazw kolumn, typów odrzutów lub obliczania kolumn pochodnych przed agregacją lub wyborem kolumn. W przypadku pominięcia wszystkie kolumny są zaznaczone (*). Przykład: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: schemat wynikowej ramki danych po przekształceniach w formacie Spark StructType JSON (z df.schema.json()). Wymagane w przypadku transformation_sql dostarczenia. Informuje system o nazwach kolumn i typach, które wynikają z transformacji.
• lateness: Obiekt SourceLateness opisujący, jak długo zwykle potrzebuje, by źródło stało się kompletne w czasie zdarzenia. Jeśli zostanie pominięty, źródło jest natychmiast uznawane za kompletne.

Po ustawieniu obu filter_condition elementów i transformation_sql wynikowe zapytanie to: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

SourceLateness.settling_delay jest zalecanym sposobem symulacji podczas treningu stałego opóźnienia ETL, które wpływa na materializację online. Azure Databricks cofa czas kwalifikacyjnej oceny treningowej o ten czas, tak aby przykład treningowy nie korzystał z danych, które nadal byłyby w trakcie transmisji online. Podczas materializacji Azure Databricks czeka ten sam czas przed publikacją ukończonego okna i serwuje ostatnie ukończone okno w tym okresie.

Na przykład, załóżmy, że codzienne zadanie ETL kończy się 8 godzin po północy w lokalnej strefie czasowej, gdzie północ odpowiada 07:00 UTC. Stosuj opóźnienie osiadania wynoszące 8 godzin i przesunięcie okna 7 godzin:

from datetime import timedelta
from databricks.feature_engineering.entities import (
    DeltaTableSource,
    SourceLateness,
    TumblingWindow,
)

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    lateness=SourceLateness(settling_delay=timedelta(hours=8)),
)

window = TumblingWindow(
    window_duration=timedelta(days=1),
    offset=timedelta(hours=7),
)

Przykład: Używanie transformation_sql dla przekształceń kolumn

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

Przykład: wyprowadzanie transformation_sql i dataframe_schema z ramki danych PySpark

Przekształcenie można napisać jako zapytanie PySpark, a następnie wyodrębnić schemat z wynikowej ramki danych:

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

Wspierane transformation_sql wyrażenia

Te same zasady dotyczą transformation_sql na i DeltaTableSourceStreamSource.

transformation_sql obsługuje dowolne wyrażenia wierszowe; operacje oceniane niezależnie dla każdego wiersza. Nie zmieniają liczby wierszy ani zgodności jeden do jednego ze źródłami. Wyrażenia wierszowe obejmują zmiany nazw kolumn, rzuty, operacje arytmetyczne i inne.

Operacje zmieniające kształt lub liczbę wierszy nie są obsługiwane, takie jak agregacje takie jak SUM() lub COUNT(). Zamiast tego użyj AggregationFunction definicji funkcji.

DeltaTableSource.from_sql()

Dla wygody możesz utworzyć element DeltaTableSource na podstawie zapytania SQL. Metoda analizuje zapytanie, aby automatycznie wyodrębnić nazwę tabeli, transformation_sqli filter_condition.

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

Obsługiwane są tylko proste SELECT ... FROM ... [WHERE ...] zapytania. Złożone sieci SQL (JOIN, podzapytania, KARTY CT, UNION) są odrzucane. W przypadku złożonych zapytań skonstruuj DeltaTableSource bezpośrednio za pomocą transformation_sql elementów i 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",
)

Iterowanie za pomocą polecenia to_dataframe()

Użyj polecenia source.to_dataframe() , aby wyświetlić podgląd danych, które będą używane do obliczeń funkcji. Jest to przydatne w przypadku iteracji filter_condition i transformation_sql do momentu wygenerowania oczekiwanych wyników.

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

Informacje o jednostkach

Kolumny jednostek definiują poziom agregacji funkcji. Są one określone w Feature definicji, a nie na DeltaTableSource. Jednostki określają:

• Sposób grupowania danych: funkcje są agregowane na unikatową kombinację wartości jednostek (podobnie jak GROUP BY w języku SQL)
• Struktura klucza podstawowego: każda unikatowa kombinacja jednostek powoduje wyświetlenie jednego wiersza obliczonych funkcji

Przykład: funkcje na poziomie klienta

Poniższy kod agreguje funkcje na poziomie klienta (jeden wiersz na klienta):

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

Przykład: funkcje na poziomie sklepu klienta

Aby agregować funkcje na bardziej szczegółowym poziomie (jeden wiersz na kombinację magazynu klientów), użyj wielu kolumn jednostek:

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

Jeśli potrzebujesz funkcji na różnych poziomach agregacji (na przykład na poziomie klienta i na poziomie magazynu klienta), użyj różnych entity wartości w definicjach funkcji. Te same DeltaTableSource funkcje mogą być współużytkowane przez różne konfiguracje jednostek.

StreamSource

StreamSource odwołuje się do strumienia. Strumień zawiera konfigurację połączenia, uwierzytelniania, schematu i pozyskiwania dla źródła przesyłania strumieniowego. W przypadku platformy Kafka odwołania do kolumn w definicjach funkcji muszą być poprzedzone prefiksem value. lub key. wskazać, która część komunikatu ma być odczytywana.

StreamSource(
    full_name: str,                       # Required: Three-part Stream name (catalog.schema.stream)
    filter_condition: Optional[str] = None,      # Optional: SQL WHERE clause applied before aggregation
    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
    lateness: Optional[SourceLateness] = None,    # Optional: Expected source settling time
)

Parametry:

• full_name: pełna trzyczęściowa nazwa strumienia (na przykład "my_catalog.my_schema.my_stream").
• filter_condition (opcjonalnie): klauzula SQL WHERE stosowana do strumieniowego przesyłania danych przed agregacją przy użyciu odwołań do kolumn z prefiksem kropki (na przykład "value.event_type = 'purchase'").
• transformation_sql (opcjonalnie): Wyrażenie SQL SELECT stosowane przed agregacją lub wyborem kolumn, z użyciem odniesień do key struktur i value z prefiksami kropki. Obsługuje te same wyrażenia wierszowe co DeltaTableSource. Jeśli zostanie pominięty, źródło korzysta ze wszystkich kolumn ().*
• dataframe_schema: Schemat JSON Spark StructType dla prognozowanego wyjścia. Wymagane, jeśli ustawisz transformation_sql.
• lateness: Obiekt SourceLateness opisujący, jak długo strumień zwykle zajmuje ukończenie w czasie zdarzenia. Zobacz: SourceLateness.settling_delay.

from databricks.feature_engineering.entities import StreamSource

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

Wyprowadzamy dataframe_schema , uruchamiając projekcję na tablicy pobierania strumienia, która odsłania key struktury i value .

transformation_sql = (
    "value.amount * value.conversion_rate AS converted_amount, "
    "struct(value.user_id AS user_id, value.event_time AS time) AS event"
)

ingestion_table = "my_catalog.my_schema.events_ingestion"
dataframe_schema = spark.sql(
    f"SELECT {transformation_sql} FROM {ingestion_table} LIMIT 0"
).schema.json()

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

RequestSource

RequestSource Definiuje schemat dla danych, które są udostępniane w czasie wnioskowania w ładunku żądania, a nie wyszukiwania ze wstępnie zmaterializowanej tabeli. Podczas trenowania te kolumny są wyodrębniane z oznaczonej ramki danych przekazanej do elementu create_training_set. Podczas obsługi modelu obiekt wywołujący musi uwzględnić je w ładunku żądania HTTP.

RequestSource można używać funkcji CustomUDF lub ColumnSelection Feature View. Nie obsługuje ona funkcji agregacji ani okien czasowych.

Definiowanie schematu

Zdefiniuj schemat jako listę FieldDefinition obiektów, z których każda określa nazwę kolumny i :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),
    ]
)

Dla schematów zawierających tablice, mapy lub struktury należy przejść dataframe_schema (JSON Spark StructType z df.schema.json()) zamiast , schemalub wywnioskować schemat z DataFrame'u za pomocą RequestSource.from_dataframe:

from databricks.feature_engineering.entities import RequestSource

request_source = RequestSource.from_dataframe(dataframe=labeled_df.select("cart_items", "device_info"))

Obsługiwane typy danych

RequestSourceobsługuje typy skalarne zdefiniowane w : ScalarDataType, , INTEGER, FLOATBOOLEANSTRINGDOUBLELONG, . TIMESTAMPDATESHORT Obsługuje ARRAYtakże , MAP, oraz STRUCT kolumny zbudowane z tych typów, gdy definiujesz schemat z lub dataframe_schemafrom_dataframe. MAP Klucze muszą być STRING.

Jak dane żądań są nawodnione

Podpis modelu

Gdy model jest rejestrowany przy użyciu log_model zestawu szkoleniowego zawierającego RequestSource funkcje, RequestSource kolumny są dodawane do podpisu modelu MLflow jako wymagane dane wejściowe. Oznacza to, że schemat interfejsu API obsługującego punkt końcowy odzwierciedla pola, które obiekty wywołujące muszą podać w czasie wnioskowania.

FeatureViewSource

FeatureViewSourcewykorzystuje wyjścia innych widoków cech jako wejść do .CustomUDF Łączenie cech tworzy skierowany graf acykliczny (DAG). Na przykład funkcja marże może łączyć agregaty przychodów i kosztów, a inna cecha może przekształcać marżę.

Dla . używaj databricks-feature-engineering wersji 0.18.0 lub nowszej FeatureViewSource.

Przekaż listę obiektów Feature do features, a nie do ciągów nazw cech. Pobierz zarejestrowane cechy z .get_feature W input_bindings, używaj każdej zarejestrowanej cechy full_name. Dla lokalnej, niezarejestrowanej funkcji użyj jej name zamiast tego.

Poniższy przykład zakłada dwie zarejestrowane cechy revenue_sum_7d oraz cost_sum_7d, które zwracają DOUBLE wartości przez customer_id i używają event_time do obliczeń punkt-in-time:

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import CustomUDF, FeatureViewSource

fe = FeatureEngineeringClient()
revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d")
cost = fe.get_feature(full_name="main.ecommerce.cost_sum_7d")

spark.sql("""
CREATE OR REPLACE FUNCTION main.ecommerce.margin_udf(revenue DOUBLE, cost DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
AS $$
import math

if revenue is None or cost is None:
    return None
if not math.isfinite(revenue) or not math.isfinite(cost) or revenue <= 0:
    return None
return (revenue - cost) / revenue
$$
""")

margin = fe.create_feature(
    source=FeatureViewSource(features=[revenue, cost]),
    function=CustomUDF(
        function_name="main.ecommerce.margin_udf",
        input_bindings={"revenue": revenue.full_name, "cost": cost.full_name},
    ),
    catalog_name="main",
    schema_name="ecommerce",
    name="margin",
)

Obowiązują następujące ograniczenia:

• Only CustomUDF jest obsługiwana jako funkcja. Funkcje upstream mogą być agregacjami, wyborami kolumn lub innymi CustomUDF cechami.
• Pomijam entity i timeseries_column na osobie pochodnej. Każda funkcja upstream zachowuje własną jednostkę, znacznik czasu oraz definicję okna.
• Funkcja ma jedno źródło. Aby połączyć wartość żądania z cechą opartą w tabeli, zdefiniuj cechę RequestSource i odwołaj się do obu przez FeatureViewSource.
• Każda zadeklarowana cecha w górze musi być użyta w .input_bindings Cykle nie są dozwolone.
• Rejestruj cechy upstream przed zarejestrowaniem wyprowadzonej cechy. Do eksperymentów można używać create_training_set lokalnych, niezarejestrowanych grafów.
• Do szkolenia lub obsługi potrzebujesz READ FEATURE uprawnienia lub MANAGE na wyprowadzoną funkcję oraz jej przechodniówe funkcje upstream. Do rejestracji i obsługi używaj wyraźnych nazw funkcji na wykresie, nawet w katalogach czy schematach.
• Funkcja może odnosić się do 20 bezpośrednich funkcji w górze. Zarejestrowane grafy obsługują maksymalną głębokość pięciu cech wzdłuż ścieżki zależności, w tym cechę bazową.
• FeatureViewSource Cechy nie mogą być materializowane ani oceniane za pomocą compute_features. Używaj create_training_set do oceny offline. Do obsługi online wykorzystaj wspierane funkcje wspierane przez stół w górnym stopniu.

Aby uzyskać ocenę zależności i wybór wyników, zobacz Train with with FeatureViewSource features. O wdrożeniu zobacz cechy pochodne w Serve.

Interfejs API trenowania i wnioskowania

create_training_set i score_batch obliczanie prawidłowych wartości funkcji punktu w czasie na żądanie z danych źródłowych. W przypadku funkcji obsługujących materializację w trybie offline, takich jak agregacje okien przesuwnych w źródłach tabel różnicowych, materializacja funkcji najpierw w magazynie offline zwiększa wydajność obu operacji. Gdy zmaterializowane funkcje trybu offline są dostępne, operacje odczytują wstępnie skompilowane dane offline zamiast ponownie skompilować wartości funkcji ze źródła. Zobacz Materialize feature Views to materialize features to materialize features to an offline store (Materialize feature Views to materialize features to an offline store

create_training_set()

Tworzy zestaw danych szkoleniowych z prawidłowymi obliczeniami funkcji do punktu w czasie. Aby uzyskać szczegółowe informacje, zobacz Trenowanie modeli za pomocą widoków funkcji.

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

Rejestruje model z metadanymi funkcji na potrzeby śledzenia pochodzenia i automatycznego wyszukiwania funkcji podczas wnioskowania. Aby uzyskać szczegółowe informacje, zobacz Trenowanie modeli za pomocą widoków funkcji.

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

Wykonuje wnioskowanie wsadowe w trybie offline z automatycznym wyszukiwaniem funkcji. Używa metadanych funkcji przechowywanych w modelu do obliczania prawidłowych funkcji w punkcie w czasie, zapewniając spójność z trenowania.

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

Ramka danych wejściowych musi zawierać kolumny jednostki i czasowników używane podczas trenowania. Funkcje są automatycznie obliczane na podstawie danych źródłowych.

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

Okna czasu

Widoki cech obsługują cztery typy okien, aby kontrolować zachowanie retrospektywy dla agregacji opartych na oknach czasowych. Dostępne typy okien zależą od źródła funkcji: funkcje źródłowe streamingu mogą korzystać z okien rolowanych i piłozębnych, a funkcje wsadowego mogą korzystać z okien rolling , rolling i suwak.

• Okna stopniowe wyglądają wstecz od czasu zdarzenia. Czas trwania i opóźnienie są jawnie zdefiniowane.
• Okna ustalane są stałe, nienakładające się okna czasowe. Każdy punkt danych należy do dokładnie jednego okna.
• Okna przesuwne to nakładające się, kroczące okna czasowe z konfigurowalnym interwałem przesunięcia.
• Okna piłozębne utrzymują długie okno retroaktywne świeże nad strumieniem strumieniowym, korzystając z hybrydowej partii i ścieżki strumieniowania. Zobacz okno Sawtooth.

Poniższa ilustracja przedstawia typy okien typu obrotowego, ślizgowego, rolowanego i ząbkowego.

Czasowe okno czasowe

Użyj delay do oceny okna w wcześniejszym momencie analitycznym. Na przykład 30-dniowe okno z 7-dniowym opóźnieniem oblicza wartość 30-dniową na tydzień przed terminem oceny. delay jest niezależny od czasu przybycia źródła. Aby zmodelować czas, jaki potrzebuje źródłowe dane na dotarcie, skonfiguruj SourceLateness.settling_delay zamiast tego.

Gdy oba ustawienia są obecne, tworzą się. Azure Databricks traktuje okno jako kompletne po opóźnieniu osiadania źródłowego i ocenia je za pomocą opóźnienia analitycznego.

Używa się offset do zmiany wyrównania stałych granic okien. Domyślnie okna obrotowe i przesuwane są wyrównane do północy UTC. Na przykład przesunięcie o 22 godziny wyrównuje dzienną granicę do 22:00 UTC. Aby przybliżyć granice w lokalnej strefie czasowej, skonfiguruj statyczny offset względem UTC. Przesunięcie nie koryguje o czas letni, nie przesuwa czasu oceny ani nie modeluje danych o opóźnionym przybyciu.

Poniższa tabela podsumowuje wsparcie dla tych pól:

*Kropka: Dla okna przewracającego się przesunięcie musi być krótsze niż window_duration. Dla okna przesuwnego musi być krótsze niż slide_duration.

Godzina rozpoczęcia

Używa start_time się do ustawienia najwcześniejszej granicy czasu zdarzenia w UTC, przy której cecha może emitować wyjście. Granica jest inkluzywna. start_time Wyjścia bramki. Nie ogranicza historycznych wierszy źródłowych, które okno może odczytać, ani nie zmienia wyrównania okna. Jeśli start_time znajduje się pomiędzy dwoma wyrównanymi granicami, pierwszym uprawnionym wyjściem w stałym oknie jest następna granica.

Przy start_time, okna o stałym czasie trwania mogą emitować przed upływem pełnego czasu trwania okna w źródle. Te wczesne wyniki wykorzystują dostępną historię źródeł. Na przykład rozważmy okno przesuwne z rocznym window_duration i jednodniowym slide_duration, nad źródłem, którego dane zaczynają się od 1 stycznia 2024 roku:

• start_timeBez , funkcja ta zostanie po raz pierwszy wyemitowana 1 stycznia 2025 roku, gdy można utworzyć pełne roczne okno.
• Film start_time zaplanowany jest na 21 sierpnia 2024 roku, a po raz pierwszy zostanie wyemitowany 21 sierpnia 2024 roku. Ten wynik obejmuje jedynie historię źródeł dostępną do tej pory, od 1 stycznia 2024 roku. Okno to osiąga pełny roczny okres 1 stycznia 2025 roku i od tego momentu generuje kompletne produkcje.

Ponieważ start_time nie zmienia wyrównania okien, wartość między dwoma wyrównanymi granicami nie tworzy nowej granicy. Dla okna obrotowego z codziennymi granicami o północy UTC, najpierw o start_time 06:00 UTC emituje się na następnej granicy o północy. A, która ląduje dokładnie na granicy, emituje start_time na tej granicy, ponieważ granica jest inkluzywna.

Jeśli start_time jest nieustawiony, okna przewracające i okna przesuwające o stałej długości trwania emitują najpierw na wyrównanej granicy po utworzeniu pełnego okna. Okna przesuwane na całe życie i okna ruchome emitują się zaraz po pojawieniu się odpowiednich danych źródłowych.

Przykład:

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

window = SlidingWindow(
    window_duration=timedelta(days=365),
    slide_duration=timedelta(days=1),
    start_time=datetime(2024, 8, 21),
)

Okno ruchome

Okna stopniowe są up-to—agregacje daty i czasu rzeczywistego, zwykle używane za pośrednictwem danych przesyłanych strumieniowo. W potokach przesyłania strumieniowego okno stopniowe emituje nowy wiersz tylko wtedy, gdy zawartość okna o stałej długości zmienia się, na przykład gdy zdarzenie wchodzi lub opuszcza. Gdy funkcja okna kroczącego jest używana w potokach trenowania, dokładne obliczenie funkcji punktu w czasie jest wykonywane na danych źródłowych przy użyciu czasu trwania okna o stałej długości bezpośrednio poprzedzających znacznik czasu określonego zdarzenia. Pomaga to zapobiec niesymetryczności danych w trybie online lub wycieku danych w trybie offline. Cechy w czasie T agregują zdarzenia z przedziału [T − długość trwania, T).

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

W poniższej tabeli wymieniono parametry okna kroczącego. Czasy rozpoczęcia i zakończenia okna są oparte na następujących parametrach:

• Godzina rozpoczęcia: evaluation_time - window_duration - delay (włącznie)
• Godzina zakończenia: evaluation_time - delay (wyłączność)

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

Zdefiniuj okno stopniowe z opóźnieniem przy użyciu poniższego kodu.

# Compute a 7-day value as of one day before the evaluation time
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(days=1)
)

Przykłady okien kroczących

• window_duration=timedelta(days=7): To utworzy 7-dniowe okno retrospektywne kończące się w momencie bieżącej oceny. W przypadku wydarzenia o godzinie 14:00 w dniu 7, obejmuje to wszystkie wydarzenia od 23:00 w dniu 0 do (ale nie w tym) 2:00 w dniu 7.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Utworzy to 1-godzinne okno retrospektywne kończące się 30 minut przed czasem oceny. W przypadku wydarzenia o godzinie 15:00 obejmuje to wszystkie zdarzenia od 13:30 do (ale nie obejmuje) 2:30.

Użyj Last do ograniczenia świeżości nowej wartości

Spójrzmy Last z , RollingWindow gdy najnowsza wartość jest ważna tylko przez ograniczony czas. W czasie oceny cecha zwraca wartość z wiersza z najpóźniejszym znacznikiem czasu w tym przedziale:

[evaluation_time - delay - window_duration, evaluation_time - delay)

Jeśli najnowszy wiersz w przedziale zawiera wartość null, cecha zwraca wartość null. Jeśli chcesz wykluczyć wartości wejściowe null, ustaw a filter_condition na źródle.

Ta kombinacja różni się od .ColumnSelection ColumnSelection zwraca ostatnią zaobserwowaną wartość niezerową bez jej wygaśnięcia ze względu na wiek.

Dla funkcji wsadowych ta kombinacja posiada specjalny tryb materializacji dostępny wyłącznie online. Obsługuje tylko DeltaTableSource, Last, RollingWindow, oraz TableTrigger. Zobacz Materializuj najnowsze wartości ograniczone świeżością.

Okno skokowe

W przypadku funkcji zdefiniowanych przy użyciu niezachodzących okien stałoczasowych, agregacje są obliczane w wstępnie określonym oknie o stałej długości, które przesuwa się według zdefiniowanego interwału, generując nienakładające się okna, które w pełni podzielają czas. W rezultacie każde zdarzenie w źródle przypisuje się dokładnie do jednego okna. Funkcje w momencie t agregują dane z okien kończących się na lub przed t (wyłącznie). System Windows zaczyna się od epoki systemu Unix.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None
    offset: Optional[datetime.timedelta] = None
    start_time: Optional[datetime.datetime] = None

W poniższej tabeli wymieniono parametry okna przesuwnego.

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

window = TumblingWindow(
    window_duration=timedelta(days=1),
    delay=timedelta(hours=2),
    offset=timedelta(hours=22),
)

Przykład okna przeskakującego

• window_duration=timedelta(days=5): Powoduje to utworzenie wstępnie określonych okien o stałej długości 5 dni. Przykład: okno nr 1 obejmuje od 0 do dnia 4, okno #2 obejmuje od 5 do dnia 9, okno #3 obejmuje dzień 10 do 14 dnia itd. W szczególności okno nr 1 zawiera wszystkie zdarzenia ze znacznikami czasu rozpoczynającymi się od 00:00:00.00 dnia 0 do (ale z wyłączeniem) wszystkich zdarzeń ze znacznikami 00:00:00.00 czasu w dniu 5. Każde zdarzenie należy do dokładnie jednego okna.

Okno przesuwane

Dla cech definiowanych za pomocą okien przesuwanych agregacje są obliczane nad oknem przesuwającym się o przedział przesuwania. Przesuwne okno może mieć stały czas trwania lub cały okres życia. Okna o stałym czasie trwania nakładają się, więc każde zdarzenie źródłowe może przyczynić się do agregacji cech dla wielu okien. Okno życiowe obejmuje wszystkie zdarzenia źródłowe przed końcem okna. Funkcje w momencie t agregują dane z okien kończących się na lub przed t (wyłącznie). Windows jest dostosowany do ery Uniksa.

class SlidingWindow(TimeWindow):
    window_duration: Optional[datetime.timedelta]
    slide_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None
    offset: Optional[datetime.timedelta] = None
    start_time: Optional[datetime.datetime] = None

W poniższej tabeli wymieniono parametry okna przesuwanego.

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

window = SlidingWindow(
    window_duration=timedelta(days=7),
    slide_duration=timedelta(days=1),
    delay=timedelta(hours=2),
    offset=timedelta(hours=22),
)

Przykład okna przesuwanego

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Powoduje to nakładanie się 5-dniowych okien, które przesuwają się o 1 dzień każdego dnia. Przykład: okno #1 obejmuje od 0 do 4 dnia, okno #2 obejmuje dzień 1 do dnia 5, okno #3 obejmuje dzień 2 do 6. dnia itd. Każde okno zawiera zdarzenia od 00:00:00.00 dnia rozpoczęcia do dnia końcowego, z wyłączeniem 00:00:00.00. Ponieważ okna nakładają się na siebie, pojedyncze zdarzenie może należeć do wielu okien (w tym przykładzie każde zdarzenie należy do maksymalnie 5 różnych okien).

Okno życia

Ustaw window_duration=None tak, aby tworzyć okno na całe życie. Na każdej granicy slajdu cecha agreguje wszystkie zdarzenia źródłowe dla jednostki z znacznikami czasowymi wyprzedzającymi tę granicę. Na przykład jednodniowy slajd generuje wartość skumulowaną raz dziennie.

Okna życia są obsługiwane wyłącznie przez SlidingWindow. RollingWindow i TumblingWindow wymagają skończonego window_duration.

from datetime import timedelta
from databricks.feature_engineering.entities import (
    AggregationFunction,
    DeltaTableSource,
    Feature,
    SlidingWindow,
    Sum,
)

lifetime_spend = Feature(
    source=DeltaTableSource(
        catalog_name="main",
        schema_name="store",
        table_name="transactions",
    ),
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(
        Sum(input="amount"),
        SlidingWindow(
            window_duration=None,
            slide_duration=timedelta(days=1),
        ),
    ),
    name="lifetime_spend",
)

Okno piłozębne

Okno piłozębne to agregacja wspierająca bardzo świeże aktualizacje ostatnich wydarzeń, wraz z codziennym zagęszczaniem danych historycznych. Jego boczna (starsza) krawędź przesuwa się w stałych, codziennych krokach, podczas gdy krawędź prowadząca pozostaje na bieżąco z najnowszymi wydarzeniami, więc efektywna długość okna "przesuwa się" w ciągu każdego dnia. Większość okna jest obsługiwana z danych w tabeli pobierania strumienia, a tylko dwa najnowsze dni pochodzą z transmisji na żywo. To kompromis, który efektywnie oblicza długoterminowe okna (skalujące się do lat), jednocześnie pozostając responsywnym na świeże aktualizacje.

Okna piłozębne są realizowane na hybrydowej ścieżce partii i strumienia. Potok wsadowy utrzymuje większość okna, podczas gdy potok strumieniowy utrzymuje najnowsze dane świeże w czasie rzeczywistym. Oba są łączone podczas odczytu, więc dla modelu lub obsługującego konsumenta jest to jedna funkcja.

Ponieważ historyczna część okna jest obliczana przez pipeline wsadowy, funkcja piłowata jest gotowa do obsługi krótko po rozpoczęciu materializacji, nawet gdy okno obejmuje miesiące lub lata. Okno ruchome jest ukończone dopiero po upływie pełnego czasu trwania okna. Minimum window_duration musi przekraczać dwa dni (obowiązująca dolna granica). Databricks zaleca okno piłozębne na okresy dłuższe niż 7 dni. Dla okien dłuższych niż dwa dni i do siedmiu dni można wybrać między precyzją o stałej długości okna obrotowego a szybszą gotowością produkcyjną okna piłozębnego.

Aby sprawdzić, czy funkcja piłowata jest gotowa, otwórz widok funkcji w Eksploratorze katalogu. W sekcji Materializowane cechy uzupełnianie wsad kończy się, gdy ostatni czas materializacji funkcji minie i jej status pokazuje sukces. Część strumieniowa jest realizowana przez deklaratywny pipeline Lakeflow. Po przejściu weryfikacji Feature View funkcja zmaterializowana łączy się z tym pipeline'em, gdzie możesz monitorować jej status działania.

Okna piłozębne wymagają a StreamSource a materializują się z .StreamingMode

class SawtoothWindow(TimeWindow):
    window_duration: datetime.timedelta

Krawędzie okna piłowatego poruszają się inaczej niż w szybie rolnej: krawędź prowadząca śledzi najnowsze zdarzenie, podczas gdy krawędź spoczynkowa przesuwa się raz dziennie, a nie ciągle. Każdego dnia, o stałym progu 18:00 UTC, krawędź tylna przesuwa się do granicy UTC-północ tego dnia. W efekcie efektywne okno jest nieco dłuższe niż window_duration i rośnie w ciągu dnia, zanim w następnym momencie wraca do normy. Szkolenia i obsługa korzystają z tego samego progu 18:00 UTC, więc szkolenia offline i obsługa online pozostają spójne.

Okna piłozębne wspierają Sumfunkcje , Avg, Count, MinLastMaxVarPopStddevPopLastNFirstNVarSampFirstorazStddevSamp funkcje agregacji.

Przykład okna piłowatego

Poniższy przykład pokazuje 7-dniowy bilans transakcji użytkownika. Krawędź prowadząca śledzi bieżące zdarzenie, podczas gdy krawędź spływowa przesuwa się do przodu dzień po dniu. Jeśli chodzi o wydarzenia 10 marca, okno sięga około 3 marca. W miarę upływu 10 marca krawędź natarcia ciągle się przesuwa, podczas gdy krawędź spoczynkowa utrzymuje, więc zakryty rozpięty rozpięty rośnie. Następnie, na początku 11 marca, krawędź podkładu przesuwa się do około 4 marca. Skuteczne okno to zawsze nieco dłużej niż siedem dni. Dwa ostatnie dni są serwowane z transmisji na żywo, a wcześniejsze z stołu wlewającego strumienia.

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

# 7-day window kept continuously fresh with streaming data
window = SawtoothWindow(window_duration=timedelta(days=7))

Ograniczenia okna piłowatego

• Parametr delay nie jest obsługiwany.
• SourceLateness.settling_delay nie jest obsługiwana.
• Funkcje agregacji inne niż Sum, Avg, Count, Min, Max, FirstNStddevPopLastNLastFirstVarPopVarSamporaz StddevSamp nie są obsługiwane (na przykład ApproxCountDistinct, ApproxPercentile, FirstDistinct, oraz ).LastDistinct
• Okna piłozębne wymagają .StreamSource A DeltaTableSource nie jest wspierany.

Wyzwalacze materializacji

Wyzwala kontrolkę po uruchomieniu potoku materializacji. Typ wyzwalacza zależy od typu funkcji.

CronSchedule

Zastosowanie CronSchedule do funkcji agregacji wsadowej. Domyślnie Azure Databricks wyprowadza harmonogram z okna agregacji. Harmonogram pochodny uwzględnia okres okna, okno delay oraz offset, oraz źródło settling_delay , tak aby uruchomienie nie publikowało okna przed oczekiwaniem kompletności danych źródłowych. Rozkłady pochodne wspierają przewracanie i przesuwanie okien.

Aby zażądać harmonogramu pochodnego, pomiń wyrażenie crona. CronSchedule() a formy jawne CronSchedule(mode=CronScheduleMode.DERIVED) są równoważne:

from databricks.feature_engineering.entities import (
    CronSchedule,
    CronScheduleMode,
)

trigger = CronSchedule(mode=CronScheduleMode.DERIVED)

Nie ustawiaj quartz_cron_expression z .CronScheduleMode.DERIVED Po odzyskaniu zmaterializowanej cechy, zwrócony harmonogram może zawierać wyrażenie cron, które obliczył Azure Databricks.

Aby bezpośrednio kontrolować harmonogram, podaj wyrażenie Quartz crona. CronScheduleMode.MANUAL jest wnioskowany, gdy podajemy wyrażenie:

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Zastosowanie TableTrigger dla ColumnSelection cech lub cech agregacji (AggregationFunction) wspieranych przez .DeltaTableSource Potok jest uruchamiany za każdym razem, gdy nadrzędna tabela delty otrzymuje nowe zatwierdzenie.

W przypadku cech agregacji pipeline jest ograniczany, aby nie uruchamiał się przy każdym commite. Potok uruchamiany jest co najwyżej raz na połowę długości okna funkcji, ale nigdy częściej niż co 5 minut. Na przykład funkcja z 1-godzinnym oknem obrotowym uruchamia się co najwyżej raz na 30 minut, lub funkcja z oknem 8-godzinnym działa co najwyżej raz na 4 godziny. Minimalna granica 5 minut obowiązuje, gdy połowa okna jest krótsza niż to, więc okna trwające 10 minut lub mniej są uruchamiane maksymalnie raz na 5 minut. Funkcje agregacji, których okno to mniej niż 5 minut, nie mogą używać TableTrigger, używając zamiast tego wyzwalacza strumieniowego.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

Służy StreamingMode do obsługi funkcji wspieranych przez element StreamSource. Potok jest uruchamiany jako potok ciągłego przesyłania strumieniowego.

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

Wybieranie wyzwalacza

Każda funkcja korzysta z jednego wyzwalacza; Opcje według typu funkcji to:

Nie można materializować funkcji, które wymagają różnych typów wyzwalaczy w jednym materialize_features wywołaniu. Zamiast tego wydaj oddzielne wywołania.