Zobrazení funkcí

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.

Zobrazení funkcí umožňují definovat a vypočítat funkce ze zdrojů dat. Funkce je možné definovat pomocí různých zdrojů (tabulka Delta, Kafka Stream a data o čase požadavků) a výpočty (agregace s časovým intervalem, jednoduché výběry sloupců a další). Tato příručka se zabývá následujícími pracovními postupy:

• Pracovní postup vývoje funkcí
  • Slouží create_feature k definování objektů funkcí katalogu Unity, které lze použít při trénování modelu a obsluhování pracovních postupů.
  • Alternativně vytvořte Feature objekty místně a použijte register_feature je k jejich uložení do katalogu Unity později. Před registrací lze použít místně vytvořené vlastnosti s create_training_set.
• Pracovní postup trénování modelu
  • Pro výpočet agregovaných atributů k určitému bodu v čase pro strojové učení použijte create_training_set. Podrobnou dokumentaci k trénování pomocí zobrazení funkcí najdete v tématu Trénování modelů pomocí zobrazení funkcí.
• Materializace vlastností a jejich poskytování procesu
  • Po definování funkce pomocí create_feature nebo jejího načtení pomocí get_feature můžete použít materialize_features k materializaci funkce nebo sady funkcí do offline úložiště pro efektivní opakované použití, nebo do online úložiště pro online obsluhu.
  • Pomocí create_training_set materializovaného zobrazení můžete připravit offline dávkovou trénovací datovou sadu.

Podrobnosti o rozhraní API najdete v referenčních informacích k rozhraní API zobrazení funkcí.

Požadavky

• Výpočetní prostředí bez serveru nebo klasický výpočetní cluster s modulem Databricks Runtime 17.0 ML nebo novějším.

• Musíte nainstalovat vlastní balíček Pythonu. Při každém spuštění poznámkového bloku spusťte tyto řádky kódu:

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

Příklad rychlého startu

Spustitelný rychlý startovací poznámkový blok najdete v tomto ukázkovém poznámkovém bloku.

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

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

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

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

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

fe = FeatureEngineeringClient()

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

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

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

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

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

    # training code

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

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

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

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

ukázkový poznámkový blok

Notebook pro rychlý úvod do zobrazení funkcí

Pořiďte si notebook

Funkce streamování

Kromě dávkových funkcí z tabulek Delta můžete definovat funkce ze zdrojů streamování pro případy použití v reálném čase. Funkce pro streamování používají stejnou třídu Feature jako dávkové funkce — stejné konstruktory Feature, stejné agregační funkce, stejné pracovní postupy pro trénování a nasazení — takže přechod z dávkového zpracování na zpracování v reálném čase vyžaduje jen minimální změny kódu. Po materializaci doručují funkce streamování dílčí sekundu koncové aktuálnosti (latence p99 o velikosti 200 ms) přímo do vašeho modelu obsluhující koncové body.

Pokud chcete používat funkce streamování, nejprve nastavte Stream a pak na něj odkazujte StreamSourcepomocí . Zdroje streamů podporují Apache Kafka jako vstup a automaticky udržují ingestní tabulku (Delta) jako historickou kopii dat používanou pro trénink.

Definování funkce streamování

StreamSource odkazuje na datový proud pomocí jeho třídílného názvu (catalog.schema.stream_name). Stream není zabezpečitelný objekt katalogu Unity, ale je vymezen schématem katalogu Unity a přístup se řídí tabulkou příjmu dat streamu. Odkazy na sloupce v entitě, časových řádcích a definicích funkcí musí mít předponu value. nebo key. určit, která část zprávy Kafka se má přečíst. Vnořená pole se podporují pomocí zápisu tečky (například value.user.address.city).

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

client = FeatureEngineeringClient()

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

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

Filtrování podmínek ve Službě StreamSource

Slouží filter_condition k filtrování řádků z datového proudu před agregací, stejně jako u DeltaTableSource.

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

Výběr sloupce ze streamů

ColumnSelection funkce fungují se zdroji streamování. Vybraný sloupec představuje nejnovější hodnotu ze služby Stream pro každou entitu při zachování přesnosti k určitému bodu v čase.

from databricks.feature_engineering.entities import ColumnSelection

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

Přístup k vnořeným polím

K vnořeným polím JSON můžete přistupovat pomocí zápisu tečky (například value.nested_field.amount). Při obsluze používají datová část požadavku i odpověď názvy koncových uzlů (například amount místo value.amount). Názvy listových uzlů musí být jedinečné napříč všemi sloupci entit, časových řad a výstupními sloupci příznaků v rámci modelu nebo specifikace příznaků, protože obslužný koncový bod používá názvy listových uzlů ke směrování hodnot.

Časová okna pro funkce streamování

Funkce streamu podporují pro agregace pouze RollingWindow. Posuvná okna se průběžně přepočítávají na základě nejnovějších dat, což odpovídá charakteru streamovacích zdrojů v reálném čase. TumblingWindow a SlidingWindow jsou navržené pro dávkové výpočty v pevných historických intervalech.

Ukázkový poznámkový blok s funkcemi streamování

Notebook s rychlým úvodem do Streaming Feature Views

Pořiďte si notebook

Trénování a odvozování modelů

Chcete-li trénovat modely a spouštět dávkové odvozování pomocí zobrazení funkcí, včetně log_model()score_batch(), a create_training_set(), viz trénování modelů se zobrazeními funkcí.

Materializace vlastností

Jakmile definujete funkce, můžete je materializovat do offline nebo online obchodů pro efektivní opakované použití při trénování a obsluhování pracovních postupů. Po materializaci funkcí můžete modely nasadit pomocí CPU. Podrobnosti najdete v tématu Materializace zobrazení funkcí.

Osvědčené postupy

Pojmenování funkcí

• Používejte popisné názvy pro důležité obchodní funkce.
• Dodržujte konzistentní konvence vytváření názvů napříč týmy.
• Při zahájení vývoje funkcí používejte automaticky generované názvy .

Časová okna

• Zarovnejte hranice oken s obchodními cykly (denně, týdně).
• Kratší okna zachycují nedávné trendy, ale mohou být hlučné. Delší okna vytvářejí stabilnější distribuce funkcí, ale můžou chybět nedávné změny chování. Zvolte na základě toho, jak rychle se změní základní signál pro váš případ použití. Například 7denní okno vyhladí denní výkyvy a vytváří konzistentní vstupy modelu, zatímco 1hodinový interval reaguje rychle na změny chování, ale může způsobit odchylku, která snižuje výkon modelu. Pokud se přesnost modelu sníží při posunu distribuce, použijte k stabilizaci vstupů delší interval.
• Přeskakující a posuvná okna jsou škálovatelná než posuvná (souvislá) okna. Začněte s posuvnými okny pro většinu případů použití.

Performance

• Materializujte funkce ze stejného zdroje dat v jednom materialize_features volání, abyste minimalizovali prohledávání dat.
• Pro funkce ze stejného zdroje dat použijte stejnou úroveň podrobnosti (například vždy 1hodinové nebo vždy 1denní trvání) k usnadnění lepšího seskupování během materializace.

Sloupce entit vs. podmínky filtru

Tento průvodce rozhodováním použijte při práci s funkcemi ze stejné zdrojové tabulky:

Použijte entity (zapnuto create_feature) v případě, že potřebujete různé úrovně agregace:

• Funkce na úrovni zákazníka (jeden řádek na zákazníka): entity=["customer_id"]
• Funkce pro obchodníky se zákazníky (více řádků na zákazníka): entity=["customer_id", "merchant_id"]
• Různé úrovně agregace můžou sdílet stejné DeltaTableSource: zadejte různé entity hodnoty pro každou definici funkce.

Použijte filter_condition (zapnuto DeltaTableSource) v případě, že potřebujete filtrovat řádky na stejné úrovni agregace:

• Pouze transakce s vysokou hodnotou: filter_condition="amount > 100" (stále agregované na zákazníka)
• Pouze dokončené objednávky: filter_condition="status = 'completed'" (stále agregované podle zákazníka)

Empirické pravidlo: Pokud by změna způsobovala jiný počet řádků na hodnotu entity, použijte u definic funkcí různé entity hodnoty. Pokud jenom filtrujete řádky, které přispívají ke stejné agregaci, použijte ve zdroji filter_condition.

Obvyklé scénáře

Analýzy zákazníků

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

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

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

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

Rozbor tendence

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

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

Sezónní vzory

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

Omezení

• Názvy sloupců entit a časových intervalů se musí shodovat mezi trénovací (označenou) datovou sadou a definicemi funkcí při použití v create_training_set rozhraní API.
• Název sloupce použitý jako label sloupec v trénovací datové sadě by neměl existovat ve zdrojových tabulkách používaných k definování Feature.
• Rozhraní create_feature API podporuje omezený seznam funkcí (UDAF). Viz Podporované funkce.
• Sloupce entity nemohou být typu DATE nebo TIMESTAMP.
• RequestSourcepodporuje pouze skalární datové typy definované v ScalarDataType (INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, , DATE). SHORT Komplexní typy, jako jsou pole, mapy a struktury, nejsou podporovány.
• RequestSource nepodporuje agregační funkce ani časová okna. Lze použít pouze ColumnSelection funkce.
• Sada názvů sloupců entit, názvů sloupců časových řad a názvů sloupců funkcí požadavku musí být globálně jedinečná pro všechny zdroje v trénovací sadě nebo koncový bod obsluhy.

• score_batch nemusí být úspěšné na výpočetních prostředcích bez serveru. Tento problém můžete obejít pomocí klasického výpočetního clusteru s modulem Databricks Runtime 17.0 ML nebo novějším.

Informace o omezeních specifických pro materializaci najdete v tématu Omezení.