Materializowanie widoków funkcji

Po utworzeniu definicji Feature View, które są przechowywane w Unity Catalog, możesz generować dane cech z tabeli źródłowej na podstawie definicji cech. Ten proces nazywa się materializacją cech. Azure Databricks tworzy potoki Lakeflow i zarządza nimi, aby zasilać tabele w Unity Catalog na potrzeby trenowania modeli oraz skoringu wsadowego lub udostępniania online.

Aby uzyskać informacje na temat obsługi widoków funkcji, zobacz Obsługa widoków funkcji.

Aby ponownie obliczyć zmaterializowane wartości po poprawieniu lub dodaniu historycznych danych źródłowych, zobacz Backfill Feature Views.

Aby usunąć zmaterializowane wartości dla wybranych obiektów, zobacz Wyczyść wartości cech dla obiektów.

Requirements

• Cechy należy tworzyć jako Feature Views i przechowywać w Unity Catalog.
• Wymagania dotyczące wersji, zobacz Wymagania.

Obsługa materializacji według typu funkcji

To, czy i gdzie cecha może zostać zmaterializowana, zależy od jej typu:

• ColumnSelection funkcje mogą być zmaterializowane w sklepach online. Zobacz materializację ColumnSelection.
• Funkcje wsadowe Last z a RollingWindow mogą być realizowane w sklepach internetowych. Zobacz Materializacja najnowszych wartości ograniczonych czasem aktualności.
• RequestSource, FeatureViewSource i CustomUDF funkcje nie mogą zostać zmaterializowane. Zobacz Funkcje na żądanie i materializacja.

Funkcje, które pojawiają się tylko w sklepach internetowych, są nadal użyteczne offline: create_training_set i compute_features obliczają swoje wartości punktowe bezpośrednio ze źródła, więc nie jest potrzebna materializacja offline.

Materializacja elementu ColumnSelection

ColumnSelection funkcje wybierają najnowszą wartość pojedynczej kolumny dla klucza jednostki bez agregacji. Można je urzeczywistnić wyłącznie w sklepach internetowych. W przypadku przypadków użycia w trybie offline (trenowanie i wnioskowanie wsadowe) ColumnSelection funkcje są pobierane bezpośrednio z danych źródłowych w czasie wykonywania zapytania, więc materializacja offline nie jest wymagana.

Właściwości materializacji

• Potok przetwarzania zapisuje najnowszy wiersz dla klucza encji w tabeli online bez okna agregacji.
• Materializacja online wypełnia tabelę online najnowszą bieżącą wartością dla każdego klucza jednostki.

Przykład

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    DeltaTableSource, Feature, ColumnSelection, TableTrigger, OnlineStoreConfig,
)

fe = FeatureEngineeringClient()

delta_source = DeltaTableSource(
    catalog_name="catalog",
    schema_name="schema",
    table_name="transactions",
)

amount_feature = Feature(
    source=delta_source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    name="latest_transaction_amount",
)

# Register before materializing
amount_feature = fe.register_feature(
    feature=amount_feature,
    catalog_name="catalog",
    schema_name="schema",
)

mfs = fe.materialize_features(
    features=[amount_feature],
    online_config=OnlineStoreConfig(
        catalog_name="catalog",
        schema_name="feats_online",
        table_name_prefix="txn_",
        online_store_name="lb_usw2"
    ),
    trigger=TableTrigger(),
)

ColumnSelection funkcjonalności używają TableTrigger, który uruchamia pipeline za każdym razem, gdy źródłowa tabela Delta otrzymuje nowy commit. Nie offline_config jest wymagane, ponieważ ColumnSelection funkcje są odczytywane bezpośrednio ze źródła dla przypadków użycia w trybie offline (trenowanie i wnioskowanie wsadowe).

Materializuj najnowsze wartości ograniczone świeżością

Agregacja wsadowa Last z RollingWindow nadaje funkcji online czas życia (TTL): jej wartość wygasa i jest usuwana z magazynu online, gdy żadna wartość źródłowa nie mieści się w tym oknie czasowym. Jest to przydatne, gdy wiek wartości decyduje o tym, czy jest ona bezpieczna do serwisowania. Na przykład stan urządzenia z ostatniej godziny może zostać zwrócony, podczas gdy starszy stan zwraca wartość null zamiast pozostawać dostępny bezterminowo.

RollingWindow określa jawnie określony czas trwania, który nadaje funkcji działanie podobne do TTL. Na przykład, jeśli źródło publikuje wartości codziennie, a Ty chcesz zachować tylko wartości z ostatnich siedmiu dni, użyj a RollingWindow o siedmiodniowym czasie trwania. Przy każdym wyzwalaczu materializacja wykonuje pełną wymianę tabeli online, więc wartości wychodzące poza okno są usuwane w przewidywalnym harmonogramie. Daje to taki sam wynik jak strumieniowanie Last z RollingWindow, ale kosztuje mniej, ponieważ nie wykorzystuje stale mocy obliczeniowej.

Ten wzorzec publikowania jest podobny do tabeli cech z TTL publikowanej w trybie migawki. Jeśli utrzymujesz zestaw tabel cech i chcesz używać ich razem z innymi Widokami Cech, możesz je dostosować do frameworka tworzenia Widoków Cech. Użyj partii Last z użyciem RollingWindow i TableTrigger.

Ta kombinacja posiada specjalny tryb materializacji dostępny wyłącznie online z następującymi wymaganiami:

• Źródłem musi być DeltaTableSource.
• Funkcja agregująca musi być Last, a jej okno musi być RollingWindow.
• Materializacja musi zapewniać OnlineStoreConfig, pomijać OfflineStoreConfig i używać TableTrigger.

W poniższym przykładzie latest_device_state_1h jest zarejestrowana cecha spełniająca następujące wymagania:

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import OnlineStoreConfig, TableTrigger

fe = FeatureEngineeringClient()

materialized = fe.materialize_features(
    features=[latest_device_state_1h],
    online_config=OnlineStoreConfig(
        catalog_name="main",
        schema_name="feature_store",
        table_name_prefix="latest_device_state_serving",
        online_store_name="device_state_store",
    ),
    trigger=TableTrigger(),
)

Na potrzeby trenowania offline i oceny wsadowej klient mechanizmu inżynierii cech oblicza wartość dla danego punktu w czasie bezpośrednio ze źródła. Nie odczytuje materializacji w trybie offline dla tej funkcji.

Considerations

• Expiry przesuwa się tylko na wyzwalacz. Wartość w trybie online i termin jej wygaśnięcia przesuwają się do przodu, gdy zatwierdzenie tabeli źródłowej wywołuje odświeżenie materializacji. Samo upływ czasu nie powoduje odświeżenia. Jeśli źródło przestanie publikować, ostatnia zmaterializowana wartość pozostaje w sklepie internetowym aż do późniejszego zatwierdzenia, które wywoła odświeżenie.
• Dopasuj kolumnę szeregów czasowych do czasu publikacji źródła. Kolumna szeregów czasowych musi odzwierciedlać, kiedy źródło opublikowało dane. W przeciwnym razie wartości online i offline się różnią, ponieważ sklep internetowy dołącza w momencie wyzwalania, a offline czyta łączenie w czasie serii czasowej.
• Ustaw czas trwania okna na wielokrotność tempa publikacji. Jeśli czas trwania RollingWindow nie jest wielokrotnością częstotliwości publikacji źródła, niektóre wartości są traktowane jako wygasłe podczas trenowania offline, mimo że nadal są widoczne online.

Funkcje na żądanie i realizacja

RequestSource wartości pochodzą z treningowej ramki danych lub z żądania inferencji, więc nie ma tabeli źródłowej do materializacji. FeatureViewSource Cechy stosują A CustomUDF do wartości cech powyżej poziomu podczas szkolenia lub służby. Nie przechowują one wstępnie obliczonego wyniku. Materializacja również nie jest obsługiwana dla funkcji CustomUDF, które są oparte na tabeli Delta.

W przypadku grafu zależności, takiego jak revenue_sum_7d i cost_sum_7d, zasilających funkcję margin:

• W sprawie szkoleń offline zadzwoń pod create_training_set, korzystając z margin. Uwzględnia funkcje źródłowe i oblicza wartości na dany moment, wykorzystując zgodne zmaterializowane dane offline, gdy są dostępne.
• Do obsługi online wykorzystaj wspierane funkcje przychodów i kosztów w sklepie internetowym. Punkt końcowy je wyszukuje i oblicza margin dla każdego żądania.
• Przekazuj do materialize_features tylko obsługiwane funkcje upstream, a nie do margin ani żadnych funkcji opartych na żądaniach. Materializacja nie powoduje rekurencyjnej materializacji zależności cechy wyliczanej.

Graf korzystający wyłącznie z funkcji opartych na żądanie nie potrzebuje sklepu internetowego. Zobacz Trenowanie z cechami FeatureViewSource i Udostępnianie cech pochodnych.

uprawnienia

Materializacja wymaga uprawnień do funkcji oraz do zasobów źródłowych i docelowych. Pełne opisy uprawnień w Unity Catalog można znaleźć w sekcji Informacje o uprawnieniach w Unity Catalog.

• Materializacja funkcji wymaga MANAGE. Wywołanie materialize_features tworzy i zarządza bazowymi potokami Lakeflow oraz tabelami Unity Catalog, więc jest to operacja zarządzania. Musisz mieć uprawnienie MANAGE do tej funkcji oraz READ FEATURE, aby odczytać zmaterializowaną definicję funkcji.

• Usunięcie zmaterializowanej funkcji jest ograniczone do jej twórcy. Tylko użytkownik, który stworzył zmaterializowaną cechę, może ją usunąć za pomocą delete_materialized_feature. To ograniczenie jest niezależne od uprawnień Unity Catalog: MANAGE funkcja lub jej schemat nadrzędny nie pozwala innemu użytkownikowi jej usunąć.

• Odczyt danych źródłowych wymaga SELECT. W przypadku funkcji korzystającej z tabeli Delta jako źródła musisz mieć SELECT w tabeli źródłowej. W przypadku funkcji korzystającej ze źródła Stream musisz mieć SELECT w tabeli pozyskiwania danych strumienia.

  Informacje o innych uprawnieniach wymaganych przez konfigurację uwierzytelniania strumienia znajdziesz w sekcji uwierzytelnianie w Kafka.

• Tworzenie tabel docelowych wymaga CREATE TABLE. Musisz mieć OfflineStoreConfig w każdym schemacie określonym przez OnlineStoreConfig lub CREATE TABLE. Miejsca offline i online mogą znajdować się w różnych schematach lub katalogach, a materializacja wymaga uprawnień dla każdego z nich.

• Pojawienie się w sklepie internetowym wymaga CAN USE. Musisz mieć CAN USE na instancji lub projekcie Lakebase używanym przez sklep internetowy. Aby uzyskać informacje o uprawnieniach w Lakebase, zobacz Przyznawanie uprawnień projektu.

• Wyświetlanie zmaterializowanych funkcji wymaga ustawienia READ FEATURE w funkcji nadrzędnej. Aby użyć list_materialized_features, musisz mieć READ FEATURE na funkcji, która została zmaterializowana.

• Odczytywanie zmaterializowanych danych wymaga elementu SELECT. Musisz mieć SELECT dla każdej tabeli wyjściowej offline lub online, do której masz dostęp. READ FEATURE w funkcji nadrzędnej nie zapewnia dostępu do tych tabel.

Dla każdego zasobu usługi Unity Catalog zaangażowanego w materializację potrzebujesz także USE CATALOG w katalogu nadrzędnym oraz USE SCHEMA w schemacie nadrzędnym. Ten wymóg dotyczy funkcji, każdego źródła lub tabeli pobierania strumienia oraz każdego skonfigurowanego miejsca docelowego. READ FEATURE i MANAGE przyznane w schemacie lub wykazie mają zastosowanie do wszystkich bieżących i przyszłych funkcji, które zawiera.

Struktury danych interfejsu API

OfflineStoreConfig

Konfiguracja magazynu offline, w którym zostaną zapisane zmaterializowane funkcje. Gdy materialize_features jest wywoływana, zaplecze magazynu funkcji tworzy tabele, korzystając z tego prefiksu. Każde wykonanie pipeline'u materializuje najnowsze wartości cech w tabeli zgodnie z harmonogramem materializacji.

OfflineStoreConfig(
    catalog_name: str,        # Catalog name for the offline table where materialized features will be stored
    schema_name: str,         # Schema name for the offline table
    table_name_prefix: str    # Table name prefix for the offline table. The pipeline may create multiple tables with this prefix, each updated at different cadences
)

from databricks.feature_engineering.entities import OfflineStoreConfig

offline_store = OfflineStoreConfig(
    catalog_name="main",
    schema_name="feature_store",
    table_name_prefix="customer_features"
)

OnlineStoreConfig

Konfiguracja sklepu online, w którym są przechowywane funkcje używane przez obsługę modelu. Materializacja tworzy tabele delty za pomocą catalog.schema.table_name_prefix, a następnie przesyła strumieniowo te tabele do magazynu funkcji online o tej samej nazwie.

from databricks.feature_engineering.entities import OnlineStoreConfig

online_store = OnlineStoreConfig(
    catalog_name="main",
    schema_name="feature_store",
    table_name_prefix="customer_features_serving",
    online_store_name="customer_features_store"
)

MaterializedFeature

Reprezentuje widok cech, który został zmaterializowany, to znaczy taki, dla którego w Unity Catalog jest dostępna wstępnie obliczona reprezentacja. Istnieją oddzielne zmaterializowane funkcje tabeli offline i tabeli online. Zazwyczaj nie tworzysz instancji MaterializedFeature bezpośrednio.

Pole latest_backfill_operation to nosi nazwę najnowszego backfillu zmaterializowanej cechy. Zobacz Backfill Feature Views.

Wywołania funkcji interfejsu API

materialize_features()

Materializuje listę widoków funkcji do tabeli delty trybu offline lub do magazynu funkcji online. Funkcjonalności muszą być zarejestrowane w katalogu Unity przed wywołaniem tej funkcji (na przykład przy użyciu create_feature lub register_feature). Lokalnie skonstruowane funkcje, które nie zostały zarejestrowane, nie będą działać.

FeatureEngineeringClient.materialize_features(
    *,                                                                     # Arguments are keyword-only
    features: List[Feature],                                               # List of Feature Views to materialize
    offline_config: Optional[OfflineStoreConfig] = None,                   # Offline store config (aggregation features only)
    online_config: Optional[OnlineStoreConfig] = None,                     # Online store config
    trigger: Union[CronSchedule, TableTrigger, StreamingMode],             # Materialization trigger
    tags: Optional[Dict[str, str]] = None,                                 # Custom tags for cost attribution
    budget_policy_id: Optional[str] = None,                                # Serverless usage policy for cost attribution
) -> List[MaterializedFeature]:

Metoda zwraca listę zmaterializowanych cech, które zawierają metadane dotyczące aktualizacji wartości cech i tabel katalogu Unity, w których cechy są zmaterializowane.

Jeśli podano zarówno OnlineStoreConfig, jak i OfflineStoreConfig, zostaną zwrócone dwie zmaterializowane cechy na każdą podaną cechę, po jednej dla każdego typu sklepu.

Parametr trigger określa, kiedy uruchamiany jest potok materializacji.

• CronSchedule: Działa według harmonogramu wywodzącego się z czasowania funkcji lub według harmonogramu Quartz cron dostarczanego przez wywołującego. Obsługuje funkcje agregacji wsadowej (AggregationFunction z DeltaTableSource).
• TableTrigger: Uruchamia się, gdy tabela Delta upstream otrzymuje commit. Obsługiwane w przypadku funkcji ColumnSelection oraz funkcji agregacji (AggregationFunction) opartych na DeltaTableSource. Dla cech agregacji potok jest ograniczany do uruchomienia co najwyżej raz na połowę szczegółowości funkcji (czas trwania przesuwania dla okna przesuwnego lub długość okna dla okna przewracającego), ograniczony do 1 godziny i nigdy częściej niż co 5 minut. Na przykład maksymalnie raz na 30 minut dla 1 godziny szczegółowości, lub co najwyżej raz na godzinę dla szczegółowości 2 godzin lub więcej. Interwał jest liczony od poprzedniego uruchomienia, więc zatwierdzenie, które nadejdzie po jego upływie, nadal niezwłocznie wyzwala uruchomienie.
• StreamingMode: działa jako ciągły potok strumieniowy. Wymagane w przypadku funkcji wspieranych przez element StreamSource.

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

Aby przypisać koszt materializacji, przekaż tags, budget_policy_id lub oba te parametry. Azure Databricks przypisuje je do tworzonego zadania lub potoku, dzięki czemu powiązane z nimi koszty zawierają informacje o przypisaniu w tabeli systemowej rozliczanego użycia. Oba są stosowane przy tworzeniu, więc przypisanie istniejącej materializacji w inny sposób oznacza stworzenie nowej. budget_policy_id przyjmuje identyfikator polityki użycia serverless. Aby ją utworzyć i uzyskać jej identyfikator, zobacz Tworzenie zasad użycia bezserwerowego. Aby poznać limity tagów, zasoby osiągane przez każdą wartość oraz sposób zapytania o wydatki przypisane, zobacz zarządzanie kosztami Feature Store.

Materializacja do sklepu offline

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    CronSchedule, OfflineStoreConfig,
)

fe = FeatureEngineeringClient()

materialized = fe.materialize_features(
    features=features,
    offline_config=OfflineStoreConfig(
        catalog_name="main",
        schema_name="feature_store",
        table_name_prefix="customer_features"
    ),
    trigger=CronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # Hourly
        timezone_id="UTC",
    ),
)

Materializowanie do sklepu online

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

fe = FeatureEngineeringClient()

materialized = fe.materialize_features(
    features=features,
    offline_config=OfflineStoreConfig(
        catalog_name="main",
        schema_name="feature_store",
        table_name_prefix="customer_features"
    ),
    online_config=OnlineStoreConfig(
        catalog_name="main",
        schema_name="feature_store",
        table_name_prefix="customer_features_serving",
        online_store_name="customer_features_store"
    ),
    trigger=CronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # Hourly
        timezone_id="UTC",
    ),
)

Materializowanie funkcji przesyłania strumieniowego

Funkcje przesyłania strumieniowego można zmaterializować tylko w sklepach online; offline_config parametr nie jest obsługiwany. Materializacja offline nie jest obsługiwana, ponieważ funkcje strumieniowe wymagają potoku przetwarzania w czasie rzeczywistym, aby zapewnić świeżość poniżej jednej sekundy. W przypadku trenowania lub oceny w trybie offline klient inżynierii funkcji ponownie obliczy wartości funkcji na podstawie każdego ocenianego punktu danych.

Funkcji przesyłania strumieniowego nie można łączyć z funkcjami wsadowymi w ramach tego samego wywołania materialize_features.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    OnlineStoreConfig, StreamingMode,
)

fe = FeatureEngineeringClient()

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

list_materialized_features()

Zwraca materializacje pojedynczej cechy zidentyfikowanej za pomocą jej pełnej nazwy. feature_name jest wymagany i może być przekazywany wyłącznie jako argument nazwany. Aby przejrzeć materializacje dla wielu funkcji, najpierw wypisz funkcje w katalogu lub schemacie, a następnie wywołaj list_materialized_features dla każdej zwróconej funkcji.

Domyślnie zwracanych jest maksymalnie 100 materializacji. Ten limit można zmienić przy użyciu parametru max_results .

FeatureEngineeringClient.list_materialized_features(
    *,                                      # Arguments are keyword-only
    feature_name: str,                      # Required: full name of the feature whose materializations to list
    max_results: int = 100,                 # Maximum number of materializations to return
) -> List[MaterializedFeature]:

delete_materialized_feature()

Przed usunięciem zmaterializowanej funkcji usuń lub zaktualizuj wszystkie modele lub specyfikacje funkcji odwołujące się do tej funkcji.

Usuwa zmaterializowaną cechę. Typ przekazywanej funkcji zależy od rodzaju funkcji.

• Funkcje agregacji: przekaż zmaterializowaną funkcję offline. Jeśli istnieje funkcja zmaterializowana online dla tej samej funkcji, obie zostaną usunięte. Dla funkcji wsadowej Last dostępnej wyłącznie online z RollingWindow, przekaż zmaterializowaną funkcję online.
• ColumnSelection funkcje: Przekaż zmaterializowaną funkcję online. ColumnSelection funkcje są zmaterializowane tylko w sklepie online (zobacz Materialization ColumnSelection), więc nie ma sparowanej funkcji offline.

W ramach materializacji funkcje są grupowane według źródła danych i okna agregacji w celu zwiększenia wydajności. ColumnSelection funkcje nie mają okna agregacji, dlatego są grupowane tylko według źródła danych. Potok materializacji, tabela offline i tabela online nie zostaną usunięte do momentu usunięcia wszystkich zgrupowanych cech. Po usunięciu ostatniej zmaterializowanej funkcji w grupie magazyn funkcji planuje skojarzone zasoby do automatycznego czyszczenia przez proces w tle. Zobacz Oczyszczanie zasobów w tle.

Aby wyczyścić zmaterializowane funkcje, przyjrzyj się tabeli skojarzonej z zmaterializowaną funkcją. Każda cecha w tabeli (jedna na kolumnę) musi zostać usunięta przed wyczyszczeniem zasobów tabel przeliczeniowych i tabeli Delta.

Użyj list_materialized_features(), aby uzyskać materialized_feature argument.

FeatureEngineeringClient.delete_materialized_feature(
    materialized_feature: MaterializedFeature,  # Required: The materialized feature to delete
) -> None

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

feature_names = [
    "main.feature_store.amount_sum_sliding_7d_1d",
    "main.feature_store.amount_sum_sliding_30d_1d",
    "main.feature_store.transaction_count_sliding_7d_1d",
    "main.feature_store.latest_transaction_amount",
    "main.feature_store.latest_user_tier",
]

for name in feature_names:
    mfs = fe.list_materialized_features(feature_name=name)   # required, keyword-only
    offline = [mf for mf in mfs if not mf.is_online]
    for mf in (offline or mfs):
        fe.delete_materialized_feature(materialized_feature=mf)
    fe.delete_feature(full_name=name)

Czyszczenie zasobów w tle

Po usunięciu zmaterializowanej funkcji usługa Databricks natychmiast usuwa metadane funkcji. Powiązana infrastruktura (tabele, potoki przetwarzania i zadania) jest usuwana asynchronicznie przez proces działający w tle.

Ponieważ wiele zmaterializowanych cech może współdzielić te same tabele i potoki przetwarzania, te współdzielone zasoby nie zostaną usunięte, dopóki nie zostaną usunięte wszystkie zmaterializowane cechy, które się do nich odwołują. Gdy zostanie usunięta ostatnia materializowana funkcja współdzieląca zestaw tabel, proces działający w tle automatycznie usunie następujące zasoby:

• Tabele Delta w trybie offline zawierające zmaterializowane dane cech
• Tabele online, jeśli funkcje zostały zmaterializowane w sklepie online
• Potok materializacji
• Zadanie aranżacji

Ten proces działający w tle używa zarządzanej przez Databricks jednostki usługi systemowej do wykonywania w twoim imieniu tych działań związanych z czyszczeniem, w tym usuwania tabel, potoków i zadań w obszarze roboczym. Od Ciebie nie jest wymagana żadna akcja. Czyszczenie jest w pełni zarządzane przez magazyn funkcji.

Wyświetlanie stanu materializacji

Aby wyświetlić status materializacji twoich widoków funkcji w interfejsie Databricks oraz diagnozować błędy materializacji, zobacz Poznawanie widoków funkcji w katalogu Unity.

Sprawdź aktualność end-to-end funkcji RollingWindow

Ta sekcja dotyczy wyłącznie RollingWindow agregacji. Funkcje wsadowe nie ujawniają metadanych świeżości streamingu opisanych tutaj.

Funkcje przesyłania strumieniowego RollingWindow, zmaterializowane do magazynu online Lakebase, zawierają metadane świeżości dla każdego wiersza. Dzięki CDF rejestrujesz każdą aktualizację tabeli online do tabeli historii Delta w Unity Catalog i mierzysz świeżość end-to-end bez uruchamiania dodatkowego pipeline'u. Świeżość end-to-end to opóźnienie między momentem wystąpienia zdarzenia a momentem, gdy jego wartość zagregowana stała się dostępna do obsługi zapytań.

Jak funkcje RollingWindow emitują wiersze

Dla cechy RollingWindow wartość dla każdego podmiotu podsumowuje zdarzenia w określonym oknie czasowym (na przykład liczba transakcji w ostatniej godzinie). Okno przesuwa się ciągle do przodu, więc wiersz jest zapisywany do sklepu internetowego w dwóch sytuacjach:

• Aktualizacja oparta na danych (is_timer = false): pojawia się nowe zdarzenie i aktualizuje agregację. Wiersz odzwierciedla najnowsze wydarzenie w oknie.
• Wygaśnięcie okna (is_timer = true): nie pojawia się żadne nowe zdarzenie, ale czas płynie naprzód, a najstarsze zdarzenia wypadają poza okno lub wygasają. Ponieważ te zdarzenia nie liczą się już do sumy, wartość się zmienia (na przykład liczba maleje), więc funkcja generuje zaktualizowany wiersz na timerze. Wiersz odzwierciedla moment, w którym najstarsze wydarzenie wygasło, czyli koniec okna tego wydarzenia.

W obu przypadkach event_time to moment, w którym nastąpiła zmiana, która spowodowała powstanie tego wiersza: pojawienie się nowego zdarzenia lub wygaśnięcie najstarszego zdarzenia. Zatem commit_time - event_time to świeżość tego wiersza, niezależnie od sposobu jego emisji.

Metadane świeżości

Każdy wiersz przesyłany strumieniowo online zawiera następujące kolumny:

Po włączeniu CDF każda zmiana jest dopisywana w tabeli Delta lb_<online_table>_history z kolumnami metadanych CDF. Kolumna delete (_pg_change_type, update_postimage, update_preimage lub insert) identyfikuje każdy typ zmian.

Włącz CDF w tabeli online

1. Lakebase CDF wymaga ustawienia tożsamości repliki na FULL dla każdej tabeli źródłowej, aby aktualizacje były rejestrowane z pełnymi obrazami wierszy przed i po. Połóż ją na tablicy online:

   ALTER TABLE <schema>.<online_table> REPLICA IDENTITY FULL;
   

2. Uruchom Lakebase CDF dla sklepu internetowego, wybierając schemat źródłowy oraz docelowy katalog i schemat w Unity Catalog. Szczegóły konfiguracji można znaleźć w Lakebase Change Data Feed.

CDF tworzy lb_<online_table>_history w schemacie docelowym.

Aby zatrzymać rejestrację zmian, zobacz Wyłącz Lakebase CDF. Wyłączenie CDF zatrzymuje strumień danych dla wszystkich schematów w projekcie. Po wyłączeniu CDF możesz zresetować tożsamość repliki tabeli online, aby usunąć dodatkowy wolumen dziennika WAL:

ALTER TABLE <schema>.<online_table> REPLICA IDENTITY DEFAULT;

Oblicz świeżość

Świeżość dotyczy commit_time - event_time zarówno wierszy opartych na danych, jak i wierszy z wygasaniem okna. Uwzględnij is_timer, aby porównać oba typy emitów i ograniczyć zapytanie do okresu, który chcesz zmierzyć.

SELECT
  event_time,
  commit_time,
  is_timer,
  (unix_micros(commit_time) - unix_micros(event_time)) / 1000 AS feature_freshness_ms
FROM <dest_catalog>.<dest_schema>.lb_<online_table>_history
WHERE _pg_change_type IN ('insert', 'update_postimage')
  AND commit_time >= (
    SELECT max(commit_time) - INTERVAL 1 DAY
    FROM <dest_catalog>.<dest_schema>.lb_<online_table>_history
  );

feature_freshness_ms to opóźnienie, w milisekundach, od czasu zdarzenia do zapisu do Online Store. To jest aktualność wartości dostępnej do wyświetlenia.

Interpretacja wyników

• Śledź zarówno p50, jak i p99 z feature_freshness_ms, i porównaj wiersze oparte na danych (is_timer = false) z wierszami z wygaśnięciem okna (is_timer = true).
• Spodziewaj się niewielkiego, okresowego skoku na granicy partii w trybie czasu rzeczywistego (domyślnie 5 minut), gdy pipeline na chwilę zatrzymuje się, aby zatwierdzić checkpoint i commit. Zazwyczaj jest to znikome w domyślnym odstępie czasu.
• Obserwuj rzadko występujący długi ogon (maksimum, powyżej 99. p.). Odzwierciedla ona nadrabianie zaległości przez potok, a nie stan ustalony, szczególnie po restarcie, takim jak restart konserwacyjny, gdy potok przetwarza zdarzenia buforowane. Mierz świeżość w stanie stacjonarnym w oknie wykluczającym restarty, a skoki restartów oceniaj osobno.

Ograniczenia

Funkcje usługi Batch

• Potoki materializacji wsadowej działają jako bezserwerowe potoki Lakeflow.
• Cech wsadowego okna kroczącego nie można materializować, z wyjątkiem szczególnego przypadku dostępnego wyłącznie online Last, opisanego w sekcji Materializowanie najnowszych wartości ograniczonych świeżością. Do trenowania offline lub wnioskowania wsadowego, cechy okna ruchomego są obliczane z danych źródłowych dla każdego punktu w czasie.
• ColumnSelection funkcje mogą być zmaterializowane tylko w sklepach online.
• RequestSource, FeatureViewSource i CustomUDF funkcje nie mogą zostać zmaterializowane. Zobacz Funkcje na żądanie i materializacja.
• Zmaterializowane funkcje można usunąć tylko w obszarze roboczym, w którym zostały utworzone.
• Tylko użytkownik, który stworzył zmaterializowaną funkcję, może ją usunąć, niezależnie od uprawnień Unity Catalog dla tej funkcji lub jej schematu nadrzędnego.
• W przypadku funkcji agregacji typu zmaterializowanego, funkcji zmaterializowanej dostępnej online nie można usunąć bezpośrednio. Usuń sparowaną, zmaterializowaną funkcję offline, a zmiana zostanie rozpropagowana w obu przypadkach.
• W przypadku zmaterializowanych cech agregacyjnych utworzonych przed 20 kwietnia 2026 r. potok materializacji nadal tworzy nowe wartości cech, dopóki wszystkie zmaterializowane cechy w potoku nie zostaną usunięte, co uruchamia proces oczyszczania zasobów. Aby utworzyć zaktualizowany potok obsługujący usuwanie poszczególnych funkcji, usuń i ponownie zmaterializuj tę funkcję.
• W przypadku zmaterializowanych ColumnSelection funkcji potok materializacji kontynuuje tworzenie nowych wartości funkcji do momentu usunięcia wszystkich zmaterializowanych funkcji w potoku, co wyzwala oczyszczanie zasobów.

Funkcje przesyłania strumieniowego

• Funkcje przesyłania strumieniowego można zmaterializować tylko w sklepach online. Materializacja w trybie offline nie jest potrzebna, ponieważ funkcje przesyłania strumieniowego w czasie trenowania są przeznaczone do ponownego skompilowania z historycznych zdarzeń na punkt danych w celu zapewnienia dokładności na poziomie milisekund.
• Funkcje przesyłania strumieniowego nie mogą być mieszane z funkcjami wsadowymi w jednym materialize_features wywołaniu.
• compute_features nie obsługuje funkcji przesyłania strumieniowego.
• Obszar roboczy musi znajdować się w regionie, który obsługuje instancje Lakebase.
• Wiadomości Kafki mogą korzystać z JSON, Avro lub Protobuf. Informacje o wymaganiach dotyczących schematów i ograniczeniach rejestru schematów znajdziesz w rejestrze schematów.
• Obsługiwane są tylko RollingWindow i SawtoothWindow dla funkcji agregacji streamingowej. TumblingWindow i SlidingWindow powinny być używane w połączeniu z funkcjami wsadowymi. Zobacz okna czasowe dla funkcji streamingowych.
• Dla funkcji strumieniowych obsługiwane są tylko funkcje agregacji Count, Sum, Last, First, FirstN, FirstDistinct, StddevPop, Max, Min, LastN, LastDistinct i Avg.
• Funkcje wybierania kolumn w źródłach danych strumieniowych nie obsługują komunikatów dostarczanych poza kolejnością. Wyświetlane jest najnowsze zdarzenie w strumieniu Kafka, nawet jeśli wartość w kolumnie szeregów czasowych jest wcześniejsza niż we wcześniej otrzymanym zdarzeniu.
• Potoki przesyłania strumieniowego są uruchamiane dwa razy w tygodniu. Każde ponowne uruchomienie może powodować opóźnienia przetwarzania i czasy uruchamiania do 1 minuty. Z wyłączeniem ponownych uruchomień, świeżość p99 wynosi 200 ms.
• Funkcji streamingowych nie można uzupełnić wstecz przez backfill_features. Aby dodać dane historyczne do strumienia do treningu, skonfiguruj źródło backfill w strumieniu.
• Funkcja RollingWindow w sklepie internetowym obejmuje tylko zdarzenia, które pojawiają się po rozpoczęciu materializacji, więc jej wartości są kompletne dopiero po upływie pełnego window_duration. Funkcja SawtoothWindow oblicza swoje historyczne wartości na podstawie tabeli ingestii Streamu w momencie rozpoczęcia materializacji. Materializacja nie przelicza tych wartości, jeśli później poprawisz dane w tabeli ingestii.
• Obsługiwany jest tylko Databricks Online Feature Store.
• Potoki materializacji strumieni działają jako bezserwerowe potoki Lakeflow.