Functieoverzichten

Important

Deze functie bevindt zich in openbare preview-versie. Werkruimtebeheerders kunnen de toegang tot deze functie beheren vanaf de pagina Previews . Zie Azure Databricks previews beheren.

Met functieweergaven kunt u functies van gegevensbronnen definiëren en berekenen. Functies kunnen worden gedefinieerd met behulp van verschillende bronnen (Delta-tabel, Kafka Stream en aanvraagtijdgegevens) en berekeningen (tijdvensteraggregaties, eenvoudige kolomselecties en meer). In deze handleiding worden de volgende werkstromen behandeld:

  • Werkstroom voor functieontwikkeling
    • Hiermee create_feature definieert u functieobjecten voor Unity Catalog die kunnen worden gebruikt in modeltraining en het leveren van werkstromen.
    • U kunt objecten ook lokaal maken Feature en gebruiken register_feature om ze later in Unity Catalog op te slaan. Lokaal samengestelde functies kunnen worden gebruikt met create_training_set vóór de registratie.
  • Werkstroom voor modeltraining
    • Gebruik create_training_set deze functie om geaggregeerde functies van een bepaald tijdstip voor machine learning te berekenen. Raadpleeg Modellen trainen met Feature Views voor gedetailleerde documentatie over het trainen met Feature Views.
  • Kenmerk materialisatie en voorziening werkstroom
    • Na het definiëren van een kenmerk met create_feature of het ophalen ervan met behulp van get_feature, kunt u het materialize_features kenmerk of de set kenmerken materialiseren naar een offline-winkel voor efficiënt hergebruik of naar een online-winkel voor online-aanbieding.
    • Gebruik create_training_set met de gemaakte weergave om een gegevensset voor offline batchtraining voor te bereiden.

Zie de API-naslaginformatie voor functieweergaven voor API-details.

Requirements

  • Serverloze berekening of een klassiek rekencluster met Databricks Runtime 17.0 ML of hoger.

  • U moet het aangepaste Python-pakket installeren. Voer de volgende regels code uit telkens wanneer je een notebook uitvoert:

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

Quickstart-voorbeeld

Voor een uitvoerbaar quickstart-notitieblok, zie Voorbeeldnotitieblok.

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

voorbeeldnotitieblok

Snelstartnotebook voor featureweergaven

Notebook krijgen

Streaming-functies

Naast batchfuncties uit Delta-tabellen kunt u functies uit streamingbronnen definiëren voor realtime gebruiksvoorbeelden. Streamingfuncties gebruiken dezelfde functieklasse als batchfuncties , dezelfde Feature constructors, dezelfde aggregatiefuncties, dezelfde trainings- en dienstwerkstromen, dus voor het upgraden van batch naar realtime zijn minimale codewijzigingen vereist. Zodra ze zijn gematerialiseerd, leveren streamingfuncties een end-to-end actualiteit van minder dan een seconde (p99-latentie van 200 ms) rechtstreeks naar de endpoints waarop uw model wordt aangeboden.

Als u streamingfuncties wilt gebruiken, stelt u eerst een Stream in en verwijst u ernaar met behulp van een StreamSource. Stream-bronnen ondersteunen Kafka als invoer en onderhouden automatisch een opnametabel (Delta) als een historische kopie van de gegevens voor training.

Een streamingfunctie definiëren

Een StreamSource verwijst naar een Stream op basis van de driedelige naam (catalog.schema.stream_name). Een Stream is geen beveiligbaar object voor Unity Catalog, maar is beperkt tot een Unity Catalog-schema en toegang wordt beheerd door de opnametabel van Stream. Kolomverwijzingen in entiteits-, tijdreeks- en functiedefinities moeten worden voorafgegaan door value. of key. om aan te geven welk deel van het Kafka-bericht moet worden gelezen. Geneste velden worden ondersteund met punt notatie (bijvoorbeeld 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)),
    ),
)

Filtervoorwaarden op StreamSource

Gebruik filter_condition deze functie om rijen uit de stroom te filteren voordat aggregatie wordt uitgevoerd, net als op DeltaTableSource.

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

Kolomselectie uit streams

ColumnSelection functies werken met streamingbronnen. De geselecteerde kolom vertegenwoordigt de meest recente waarde uit de Stream voor elke entiteit, waarbij de nauwkeurigheid van een bepaald tijdstip wordt gerespecteerd.

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

Toegang krijgen tot geneste velden

U kunt geneste JSON-velden openen met behulp van punt notatie (bijvoorbeeld value.nested_field.amount). Tijdens het uitvoeren van de aanvraag gebruiken de payload en de respons de namen van leaf-knooppunten (bijvoorbeeld amount in plaats van value.amount). Bladknooppuntnamen moeten uniek zijn voor alle entiteits-, tijdreeks- en functieuitvoerkolommen binnen een model of functiespecificatie, omdat het ondersteunende eindpunt bladnamen gebruikt om waarden te routeren.

Tijdvensters voor streamingfuncties

Streamingfuncties bieden alleen RollingWindow ondersteuning voor aggregaties. Schuivende vensters worden voortdurend opnieuw berekend op basis van de meest recente gegevens, wat aansluit bij het realtimekarakter van streaminggegevensbronnen. TumblingWindow en SlidingWindow zijn ontworpen voor batchberekening via vaste historische intervallen.

Voorbeeldnotebook voor streamingfuncties

Notebook voor snelstart van streamingfeatureweergaven

Notebook krijgen

Modeltraining en inferentie

Als u modellen wilt trainen en batchdeductie wilt uitvoeren met functieweergaven, waaronder log_model(), score_batch()en create_training_set(), raadpleegt u Modellen trainen met functieweergaven.

Kenmerk-materialisatie

Nadat u functies hebt gedefinieerd, kunt u ze naar offline of online winkels brengen voor efficiënt hergebruik in trainings- en servicewerkstromen. Nadat u functies hebt gerealiseerd, kunt u modellen bedienen met behulp van cpu-modellen. Zie Feature Views materialiseren voor meer informatie.

Beste praktijken

Naamgeving van functies

  • Gebruik beschrijvende namen voor bedrijfskritieke functies.
  • Volg consistente naamconventies tussen teams.
  • Gebruik automatisch gegenereerde namen wanneer u begint met het ontwikkelen van functies.

Tijdvensters

  • Venstergrenzen uitlijnen met bedrijfscycli (dagelijks, wekelijks).
  • Kortere vensters leggen recente trends vast, maar kunnen luidruchtig zijn. Langere vensters produceren stabielere functiedistributies, maar kunnen recente gedragsverschuivingen missen. Kies op basis van hoe snel het onderliggende signaal verandert voor uw use-case. Een venster van 7 dagen verzacht bijvoorbeeld dagelijkse schommelingen en produceert consistente modelinvoer, terwijl een venster van 1 uur snel reageert op gedragswijzigingen, maar kan afwijking veroorzaken die de modelprestaties verslechtert. Als de nauwkeurigheid van uw model verslechtert wanneer de verdeling verschuift, gebruikt u een langer venster om de invoer te stabiliseren.
  • Tumbling en glijdende vensters zijn schaalbaarder dan rollende vensters. Begin met schuifvensters voor de meeste gebruiksvoorbeelden.

Performance

  • Materialiseer functies van dezelfde gegevensbron in één materialize_features aanroep om gegevensscans te minimaliseren.
  • Gebruik dezelfde granulariteit (bijvoorbeeld alle met een duur van 1 uur of 1 dag) voor kenmerken in dezelfde gegevensbron om betere groepering mogelijk te maken tijdens de materialisatie van gegevens.

Entiteitskolommen versus filtervoorwaarden

Gebruik deze handleiding voor beslissingen bij het werken met functies uit dezelfde brontabel:

Gebruik entity (aan create_feature) wanneer u verschillende aggregatieniveaus nodig hebt:

  • Functies op klantniveau (één rij per klant): entity=["customer_id"]
  • Klanten-verkoper functies (meerdere rijen per klant): entity=["customer_id", "merchant_id"]
  • Verschillende aggregatieniveaus kunnen hetzelfde DeltaTableSourcedelen: verschillende entity waarden opgeven voor elke functiedefinitie

Gebruik filter_condition (aan DeltaTableSource) wanneer u rijen op hetzelfde aggregatieniveau wilt filteren:

  • Alleen transacties met hoge waarde: filter_condition="amount > 100" (nog steeds geaggregeerd per klant)
  • Alleen voltooide orders: filter_condition="status = 'completed'" (nog steeds samengevoegd per klant)

Vuistregel: Als uw wijziging resulteert in een ander aantal rijen per entiteitswaarde, gebruikt u verschillende entity waarden voor uw functiedefinities. Als u alleen filtert welke rijen bijdragen aan dezelfde aggregatie, gebruik dan filter_condition op de bron.

Algemene patronen

Klantanalyse

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

Trendanalyse

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

Seizoensgebonden patronen

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

Limitations

  • Namen van entiteits- en tijdreekskolommen moeten overeenkomen tussen de trainingsgegevensset (gelabeld) en de functiedefinities wanneer deze worden gebruikt in de create_training_set API.
  • De kolomnaam die wordt gebruikt als de label kolom in de trainingsgegevensset mag niet voorkomen in de brontabellen die worden gebruikt voor het definiëren van Features.
  • Een beperkte lijst met functies (UDAF's) wordt ondersteund in de create_feature API. Zie Ondersteunde functies.
  • Entiteitskolommen kunnen niet van het type DATE zijn of TIMESTAMP.
  • RequestSourceondersteunt alleen scalaire gegevenstypen die zijn gedefinieerd in ScalarDataType (INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, DATE). SHORT Complexe typen, zoals matrices, kaarten en structs, worden niet ondersteund.
  • RequestSource biedt geen ondersteuning voor aggregatiefuncties of tijdvensters. Alleen ColumnSelection functies kunnen worden gebruikt.
  • De set van kolomnamen van entiteiten, tijdreekskolomnamen en kolomnamen van aanvraagfuncties moet globaal uniek zijn over alle bronnen in een trainingsset of een uitvoerend eindpunt.
  • score_batch kan niet slagen op serverloze berekeningen. U kunt dit omzeilen met behulp van een klassiek rekencluster met Databricks Runtime 17.0 ML of hoger.

Zie Beperkingen voor materialisatiespecifieke beperkingen.

  • Last updated on