API-referentie voor Feature Views

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.

Toegangsbeheer

Functies zijn beheerbare Unity Catalog-objecten. Toegang tot een functie wordt beheerd door de CREATE FEATUREbevoegdheden en READ FEATUREMANAGE unity-catalogus. Zie de naslaginformatie over unity-catalogusbevoegdheden voor volledige beschrijvingen.

• CREATE FEATURE — Vereist voor het maken van een functie in een schema. create_feature en register_feature vereisen CREATE FEATURE op het bovenliggende schema. Als u het principe van minimale bevoegdheden volgt, verleent CREATE FEATURE u het op schemaniveau. U kunt het ook verlenen aan een catalogus om het maken van functies in elk schema in die catalogus toe te staan.
• READ FEATURE — Vereist om een functie en de bijbehorende gegevens te lezen. get_feature, create_training_seten het lezen van gerealiseerde functiegegevens voor training of het leveren van functies is vereist READ FEATURE voor de functie. READ FEATURE verleend aan een schema of catalogus is van toepassing op alle huidige en toekomstige functies die het bevat.
• MANAGE — Vereist voor het beheren van de levenscyclus en subsidies van een functie. Het verwijderen van een functie met delete_feature, en het materialiseren van een functie met materialize_features of delete_materialized_feature, vereisen MANAGE voor de functie.

Alle functiebewerkingen vereisen USE CATALOG ook de bovenliggende catalogus en USE SCHEMA het bovenliggende schema. Zie MANAGE voor informatie over hoe READ FEATURE en van toepassing is op materialisatie.

Api voor functieweergave

Feature constructor en register_feature()

De aanbevolen methode is om een Feature object lokaal te maken en te gebruiken register_feature om het te behouden in Unity Catalog. Met deze werkstroom in twee stappen kunt u experimenteren met functies (inclusief create_training_set) voordat u ze registreert.

Feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
)

FeatureEngineeringClient.register_feature() registreert een lokaal samengestelde Feature in Unity Catalog.

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() valideert, maakt en registreert onmiddellijk een functie in Unity Catalog in één stap. Gebruik deze optie wanneer u niet eerst hoeft te experimenteren met de functie.

FeatureEngineeringClient.create_feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    catalog_name: str,                                     # Required: The catalog name for the feature
    schema_name: str,                                      # Required: The schema name for the feature
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

Parameters:

• source: de gegevensbron die wordt gebruikt in functieberekeningen (DeltaTableSource, StreamSourceof RequestSource).
• function: Een AggregationFunction die de operator bundelt (bijvoorbeeld Sum(input="amount")), de invoerkolom en het tijdvenster samen. Of ColumnSelection("column_name") voor passthrough-functies.
• catalog_name: de naam van de Unity Catalog-catalogus voor de functie.
• schema_name: de schemanaam van de Unity-catalogus voor de functie.
• entity: Lijst met kolomnamen waarmee het aggregatieniveau (primaire sleutels) wordt gedefinieerd. Vereist voor aggregatiefuncties. Bijvoorbeeld ["user_id"] aggregaties per gebruiker.
• timeseries_column: De tijdstempelkolom die wordt gebruikt voor tijdvensteraggregatie. Vereist voor aggregatiefuncties.
• name: Optionele functienaam. Als u dit weglaat, wordt automatisch gegenereerd op basis van de invoerkolom, functie en het venster (bijvoorbeeld amount_avg_rolling_7d).
• description: Optionele beschrijving van de functie.

Retourneert: Een gevalideerd Feature-instance

Verhoogt: ValueError als validatie mislukt

delete_feature()

Hiermee verwijdert u een functie uit Unity Catalog op basis van de volledig gekwalificeerde naam.

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

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

Voordat u een functie verwijdert, moet u alle modellen of functiespecificaties die ernaar verwijzen, verwijderen of bijwerken. Als de functie is gerealiseerd, verwijdert u eerst de gerealiseerde functie. Zie Een gerealiseerde functie verwijderen.

Automatisch gegenereerde namen

Wanneer name u dit weglaat, wordt automatisch een naam gegenereerd. Gegenereerde namen volgen het patroon: {column}_{function}_{window}. Voorbeeld:

• price_avg_rolling_1h (gemiddelde prijs van 1 uur)
• transaction_count_rolling_30d_1d (Aantal transacties van 30 dagen met een vertraging van 1d uit gebeurtenistijdstempel)

Ondersteunde functies

Aggregatiefuncties

Note

Aggregatiefuncties worden samen AggregationFunction met een tijdvenster verpakt, zoals wordt beschreven in tijdvensters. Elke functie gebruikt een input parameter die de bronkolom opgeeft om samen te voegen.

Function	Description	Voorbeeld van een toepassing
Sum(input="column")	Totaal van waarden	Dagelijks app-gebruik per gebruiker in minuten
Avg(input="column")	Gemiddelde van waarden	Gemiddelde transactiebedrag
Count(input="column")	Aantal records	Aantal aanmeldingen per gebruiker
Min(input="column")	Minimumwaarde	Laagste hartslag geregistreerd door een draagbaar apparaat
Max(input="column")	Maximumwaarde	Hoogste transactiebedrag per sessie
StddevPop(input="column")	Standaarddeviatie van populatie	Variabiliteit van dagelijkse transactiehoeveelheid voor alle klanten
StddevSamp(input="column")	Voorbeeld van standaarddeviatie	Variabiliteit van klikfrequenties voor advertentiecampagnes
VarPop(input="column")	Afwijking van populatie	Verspreiding van sensormetingen voor IoT-apparaten in een fabriek
VarSamp(input="column")	Voorbeeldvariantie	Verspreiding van filmbeoordelingen over een steekproefgroep
ApproxCountDistinct(input="column", relativeSD=0.05)	Geschat unieke aantal	Uniek aantal gekochte artikelen
ApproxPercentile(input="column", percentile=0.95, accuracy=100)	Percentiel bij benadering	reactielatentie p95
First(input="column")	Eerste waarde	Eerste tijdstempel voor aanmelding
Last(input="column")	Laatste waarde	Meest recente aankoopbedrag

ColumnSelection (passthrough)

ColumnSelection selecteert één kolom uit een bron zonder aggregatie toe te passen. Het wordt rechtstreeks in de function parameter verpakt (niet binnen AggregationFunction). Het retourtype wordt afgeleid uit het bronschema.

Function	Description	Voorbeeld van een toepassing
ColumnSelection("col")	Meest recente waarde van een kolom (geen aggregatie)	Meest recente leverancierscategorie, passthrough van een aanvraagveld

ColumnSelection kan worden gebruikt met elke gegevensbron:

• DeltaTableSource: retourneert de meest recente waarde per entiteitssleutel via een punt-in-time-join (geen aggregatie van lookbackvensters).
• StreamSource: retourneert de meest recente waarde per entiteitssleutel uit de stream (geen aggregatie van lookbackvensters).
• RequestSource: Geeft de waarde door die tijdens deductietijd is opgegeven (of geëxtraheerd uit het gelabelde DataFrame tijdens de training).

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

Voorbeeld: functies voor aggregatie en kolomselectie

In het volgende voorbeeld ziet u functies die zijn gedefinieerd in dezelfde gegevensbron.

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

Functies met filtervoorwaarden

filter_condition Met de parameter kunt u rijen uit de brontabel filteren voordat u aggregaties rekent. Deze functie fungeert als een SQL-component WHERE die wordt toegepast voordat gegevens worden gegroepeerd en samengevoegd.

Note

filter_condition filtert rijen vóór aggregatie, zoals een SQL-component WHERE die eerder GROUP BYis toegepast. De granulariteit wordt niet gewijzigd. Deze wordt altijd gedefinieerd door entity de functiedefinitie.

Filters zijn handig bij het werken met grote brontabellen met een superset van gegevens die nodig zijn voor functieberekeningen en het minimaliseren van de noodzaak om afzonderlijke weergaven boven op deze tabellen te maken.

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

Gegevensbronnen

DeltaTableSource

DeltaTableSource is een kortstondig Python object dat wordt gebruikt om te definiëren hoe functies worden berekend vanuit een brontabel. Er wordt geen nieuwe tabel gemaakt. Hiermee geeft u de configuratie op voor het lezen van gegevens en het samenvoegen van functies.

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
)

Parameters:

• catalog_name, , schema_name: table_nameIdentificeer de Delta-brontabel in Unity Catalog.
• filter_condition: Een SQL-component WHERE die vóór de aggregatie wordt toegepast. Voorbeeld: "status = 'completed'".
• transformation_sql: Een SQL-expressie SELECT die is toegepast op de brontabel. Gebruik deze optie om de naam van kolommen, cast-typen of afgeleide berekeningskolommen te wijzigen voordat u deze samenvoegt. Als u dit weglaat, worden alle kolommen geselecteerd (*). Voorbeeld: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: Het schema van het resulterende DataFrame na transformaties, in Spark StructType JSON-indeling (van df.schema.json()). Vereist indien transformation_sql opgegeven. Dit vertelt het systeem de kolomnamen en typen die het resultaat zijn van uw transformatie.

Wanneer beide filter_condition zijn transformation_sql ingesteld, is de resulterende query: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

Note

De timeseries_column (opgegeven in de functiedefinitie, niet op DeltaTableSource) moet van het type TimestampType zijn of DateType. Typen gehele getallen kunnen werken, maar leiden tot verlies in precisie voor tijdvensteraggregaties.

Voorbeeld: Gebruiken transformation_sql voor kolomtransformaties

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

Voorbeeld: Afgeleid transformation_sql en dataframe_schema van een PySpark DataFrame

U kunt uw transformatie schrijven als een PySpark-query en vervolgens het schema extraheren uit het resulterende DataFrame:

df = spark.sql(f"""
  SELECT user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time
  FROM main.analytics.events
  WHERE event_date >= date_sub(current_date(), 7)
  LIMIT 0
""")

# Use df.schema.json() as the dataframe_schema
source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    transformation_sql="user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time",
    filter_condition="event_date >= date_sub(current_date(), 7)",
    dataframe_schema=df.schema.json(),
)

Note

transformation_sql biedt alleen ondersteuning voor rijgewijze expressies (kolomnamen, casts, rekenkundige bewerkingen). Aggregatiefuncties zoals COUNT(*) of SUM() worden niet ondersteund. Gebruik AggregationFunction in plaats daarvan de functiedefinitie.

DeltaTableSource.from_sql()

Als u wilt, kunt u een DeltaTableSource sql-query maken. De methode parseert de query om automatisch de tabelnaam op te halen, transformation_sqlen filter_condition.

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

Alleen eenvoudige SELECT ... FROM ... [WHERE ...] query's worden ondersteund. Complexe SQL (JOIN's, subquery's, CTEs, UNION's) wordt geweigerd. Voor complexe query's maakt DeltaTableSource u rechtstreeks met transformation_sql en 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",
)

Herhalen met to_dataframe()

Gebruik source.to_dataframe() deze functie om een voorbeeld te bekijken van de gegevens die worden gebruikt voor functieberekeningen. Dit is handig voor het herhalen en filter_conditiontransformation_sql totdat ze de verwachte resultaten produceren.

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

Informatie over entiteiten

Entiteitskolommen definiëren het aggregatieniveau voor uw functies. Ze worden opgegeven voor de Feature definitie, niet op DeltaTableSource. Entiteiten bepalen:

• Hoe gegevens worden gegroepeerd: Functies worden geaggregeerd per unieke combinatie van entiteitswaarden (vergelijkbaar met GROUP BY in SQL)
• De primaire-sleutelstructuur: elke unieke entiteitscombinatie resulteert in één rij met berekende functies

Voorbeeld: Functies op klantniveau

Met de volgende code worden functies op klantniveau samengevoegd (één rij per klant):

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

Voorbeeld: Functies op klantarchiefniveau

Als u functies op een gedetailleerder niveau wilt aggregeren (één rij per combinatie van klantarchief), gebruikt u meerdere entiteitskolommen:

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

Wanneer u functies nodig hebt op verschillende niveaus van aggregatie (bijvoorbeeld op klant- en winkelniveau), gebruikt u verschillende entity waarden in uw functiedefinities. Hetzelfde DeltaTableSource kan worden gedeeld tussen functies met verschillende entiteitsconfiguraties.

StreamSource

StreamSource verwijst naar een Stream. De Stream bevat de configuratie voor verbinding, verificatie, schema en opname voor de streamingbron. Voor Kafka moeten kolomverwijzingen in functiedefinities worden voorafgegaan door value. of key. om aan te geven welk deel van het bericht moet worden gelezen.

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

Parameters:

• full_name: De volledige driedelige naam van een Stream (bijvoorbeeld "my_catalog.my_schema.my_stream").
• filter_condition (optioneel): Een SQL-component WHERE die wordt toegepast op het streamen van gegevens vóór de aggregatie, met behulp van punt-voorvoegselkolomverwijzingen (bijvoorbeeld "value.event_type = 'purchase'").

from databricks.feature_engineering.entities import StreamSource

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

RequestSource

RequestSource definieert een schema voor gegevens die worden opgegeven bij deductietijd in de nettolading van de aanvraag in plaats van op te zoeken vanuit een vooraf gerealiseerde tabel. Tijdens de training worden deze kolommen geëxtraheerd uit het gelabelde DataFrame dat wordt doorgegeven aan create_training_set. Tijdens het leveren van het model moet de aanroeper deze opnemen in de nettolading van de HTTP-aanvraag.

RequestSource wordt gebruikt met ColumnSelection (om rechtstreeks een waarde door te geven). Het biedt geen ondersteuning voor aggregatiefuncties of tijdvensters.

Het schema definiëren

Definieer het schema als een lijst FieldDefinition met objecten, die elk een kolomnaam en een 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),
    ]
)

Ondersteunde gegevenstypen

RequestSourceondersteunt de scalaire typen die zijn gedefinieerd inScalarDataType: INTEGER, FLOAT, BOOLEAN, STRING, DOUBLELONG, TIMESTAMP. DATESHORT Complexe typen zoals matrices, kaarten en structs worden niet ondersteund.

Hoe aanvraaggegevens worden gehydrateerd

Context	Gedrag
Training (create_training_set)	Kolommen worden geëxtraheerd uit het gelabelde DataFrame. Typen worden gevalideerd op basis van het gedeclareerde schema. Niet-overeenkomsten resulteren in een fout (geen impliciete typeconversie).
Serveren (modeleindpunt)	Kolommen worden opgehaald uit dataframe_records of dataframe_split in de HTTP-aanvraag. JSON-waarden worden omgezet in de gedeclareerde typen (bijvoorbeeld JSON-nummer → DOUBLE).

Modelhandtekening

Wanneer een model wordt geregistreerd met log_model een trainingsset die functies bevat RequestSource , worden de RequestSource kolommen toegevoegd aan de MLflow-modelhandtekening als vereiste invoer. Dit betekent dat het API-schema van het servereindpunt aangeeft welke velden aanroepers moeten opgeven tijdens deductietijd.

Trainings- en deductie-API

create_training_set()

Hiermee maakt u een trainingsgegevensset met de juiste functieberekening naar een bepaald tijdstip. Zie Modellen trainen met functieweergaven voor meer informatie.

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

Registreert een model met functiemetagegevens voor het bijhouden van herkomst en het automatisch opzoeken van functies tijdens deductie. Zie Modellen trainen met functieweergaven voor meer informatie.

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

Voert offline batchdeductie uit met automatische functiezoekactie. Maakt gebruik van de functiemetagegevens die zijn opgeslagen met het model om de juiste functies naar een bepaald tijdstip te berekenen, waardoor consistentie met training wordt gegarandeerd.

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

Het dataframe voor invoer moet de kolommen entiteit en tijdreeksen bevatten die tijdens de training worden gebruikt. Functies worden automatisch berekend op basis van de brongegevens.

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

Tijdvensters

Functieweergaven ondersteunen drie verschillende venstertypen voor het beheren van lookbackgedrag voor op tijdvensters gebaseerde aggregaties: rolling, tumbling en sliding.

• Terugdraaiende vensters kijken terug vanaf de tijd van de gebeurtenis. Duur en vertraging worden expliciet gedefinieerd.
• Tumbling-vensters zijn vaste, niet-overlappende tijdvensters. Elk gegevenspunt behoort tot precies één venster.
• Schuifvensters zijn overlappende en doorlopende tijdvensters met een configureerbaar verschuivingsinterval.

In de volgende afbeelding ziet u hoe ze werken.

Doorlopend venster

Note

RollingWindow was eerder benoemd ContinuousWindow. Als u migreert vanaf een eerdere SDK-versie, werkt u de importbewerkingen dienovereenkomstig bij.

Doorlopende vensters worden up-to-datum- en realtimeaggregaties, die doorgaans worden gebruikt voor streaminggegevens. In streaming-pijplijnen verzendt het rolling venster alleen een nieuwe rij wanneer de inhoud van het venster met vaste lengte verandert, bijvoorbeeld wanneer een gebeurtenis binnenkomt of verlaat. Wanneer een functie voor rolling window wordt gebruikt in trainingspijplijnen, wordt een nauwkeurige functieberekening voor een bepaald tijdstip uitgevoerd op de brongegevens met behulp van de duur van het venster met vaste lengte direct voorafgaand aan de tijdstempel van een specifieke gebeurtenis. Dit helpt het verschil tussen online en offline te voorkomen of lekkage van gegevens te voorkomen. Functies op tijd T aggregeren gebeurtenissen van [T − duur, T).

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

De volgende tabel bevat de parameters voor een doorlopend venster. De begin- en eindtijden van het venster zijn als volgt gebaseerd op deze parameters:

• Begintijd: evaluation_time - window_duration - delay (inclusief)
• Eindtijd: evaluation_time - delay (exclusief)

Parameter	Constraints
delay (optioneel)	Moet ≥ 0 zijn (hiermee wordt het venster terug in de tijd verplaatst van de tijdstempel van de evaluatie). Gebruik delay dit om rekening te houden met eventuele systeemvertragingen tussen het moment dat de gebeurtenis wordt gemaakt en de tijdstempel van de gebeurtenis om toekomstige lekkage van gebeurtenissen in trainingsgegevenssets te voorkomen. Als er bijvoorbeeld een vertraging is van één minuut tussen de tijd dat gebeurtenissen worden gemaakt en deze gebeurtenissen uiteindelijk terechtkomen in een brontabel waaraan ze een tijdstempel worden toegewezen, is de vertraging dan timedelta(minutes=1).
window_duration	Moet 0 zijn >

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

Definieer een doorlopend venster met vertraging met behulp van onderstaande code.

# Look back 7 days, offset by 1 minute to account for data ingestion delay
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(minutes=1)
)

Voorbeelden van rolling vensters

• window_duration=timedelta(days=7): Hiermee wordt een terugzoekvenster van zeven dagen gemaakt dat eindigt op de huidige evaluatietijd. Voor een evenement om 14:00 uur op dag 7, omvat dit alle gebeurtenissen van 2:00 uur op dag 0 tot (maar niet inclusief) 2:00 uur op dag 7.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Hiermee wordt een terugzoekvenster van 1 uur gemaakt dat eindigt op 30 minuten vóór de evaluatietijd. Voor een evenement om 13:00 uur omvat dit alle gebeurtenissen van 13:30 tot (maar niet inclusief) 2:30 uur. Dit is handig om rekening te houden met vertragingen bij gegevensopname.

tuimelend venster

Voor functies die zijn gedefinieerd door tumbling vensters, worden aggregaties berekend over een vooraf bepaald venster met vaste lengte dat voortgaat met een schuifinterval, waardoor niet-overlappende vensters worden geproduceerd die de tijd volledig opdelen. Als gevolg hiervan draagt elke gebeurtenis in de bron bij aan precies één venster. Functies op het moment t aggregeren gegevens uit vensters die eindigen op of vóór t (exclusief). Windows begint bij het Unix-tijdperk.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

De volgende tabel bevat de parameters voor een tumbling window.

Parameter	Constraints
window_duration	Moet 0 zijn >

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

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

Voorbeeld van een schuivend venster

• window_duration=timedelta(days=5): Hiermee maakt u vooraf ingestelde vaste-lengte vensters van elk 5 dagen. Voorbeeld: Venster 1 omvat dag 0 tot dag 4, venster 2 omvat dag 5 tot dag 9, venster 3 is dag 10 tot dag 14, enzovoort. In het bijzonder bevat Venster 1 alle gebeurtenissen met tijdstempels vanaf 00:00:00.00 dag 0 tot (maar niet inclusief) gebeurtenissen met tijdstempel 00:00:00.00 op dag 5. Elke gebeurtenis behoort tot precies één venster.

Schuifvenster

Voor functies die zijn gedefinieerd met behulp van schuifvensters, worden aggregaties berekend over een vooraf bepaald venster met een vaste lengte dat verschuift met een schuifinterval, waardoor overlappende vensters ontstaan. Elke gebeurtenis in de bron kan bijdragen aan functieaggregatie voor meerdere vensters. Functies op het moment t aggregeren gegevens uit vensters die eindigen op of vóór t (exclusief). Windows begint bij het Unix-tijdperk.

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

De volgende tabel bevat de parameters voor een schuifvenster.

Parameter	Constraints
window_duration	Moet 0 zijn >
slide_duration	Moet 0 zijn > en <window_duration

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

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

Voorbeeld van schuifvenster

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Hiermee worden overlappende 5-daagse vensters gemaakt die elke keer met 1 dag vooruitgaan. Voorbeeld: Venster 1 omvat dag 0 tot dag 4, venster 2 omvat dag 1 tot dag 5, venster 3 is dag 2 tot dag 6, enzovoort. Elk venster bevat gebeurtenissen vanaf 00:00:00.00 de begindag tot (maar niet inclusief) 00:00:00.00 op de einddag. Omdat vensters elkaar overlappen, kan één gebeurtenis tot meerdere vensters behoren (in dit voorbeeld behoort elke gebeurtenis tot maximaal 5 verschillende vensters).

Materialisatietriggers

Triggers bepalen wanneer een materialisatiepijplijn wordt uitgevoerd. Het triggertype is afhankelijk van het functietype.

CronSchedule

Gebruiken CronSchedule voor aggregatiefuncties (AggregationFunction). De pijplijn wordt uitgevoerd volgens een vast schema dat is gedefinieerd door een Kwarts cron-expressie.

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Gebruiken TableTrigger voor ColumnSelection functies die worden ondersteund door een DeltaTableSource. De pijplijn wordt uitgevoerd wanneer de upstream Delta-tabel een nieuwe doorvoering ontvangt.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

Gebruiken StreamingMode voor functies die worden ondersteund door een StreamSource. De pijplijn wordt uitgevoerd als een pijplijn voor continue streaming.

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

Een trigger kiezen

Functietype	Trigger	Wanneer deze wordt uitgevoerd
Aggregatie (AggregationFunction) van DeltaTableSource	CronSchedule	Volgens een vast cron-schema
ColumnSelection (van DeltaTableSource)	TableTrigger	Bij elke brontabeldoorvoering
Functies van StreamSource	StreamingMode	Doorlopend streamen

U kunt geen functies materialiseren waarvoor verschillende triggertypen in één materialize_features aanroep zijn vereist. Gebruik in plaats daarvan afzonderlijke aanroepen.

Bètafuncties migreren naar openbare preview

Openbare preview van functieweergaven introduceert eersteklas functie-entiteiten in Unity Catalog, die worden beheerd door de CREATE FEATURE en READ FEATURE bevoegdheden, en vereist databricks-feature-engineering versie 0.16.0 of hoger. Functies die zijn gemaakt tijdens de bètaversie (met versie 0.15.0) worden opgeslagen als Unity Catalog-functies en bieden geen ondersteuning voor alle openbare preview-functies. Maak uw bètafuncties opnieuw met versie 0.16.0 om langetermijnondersteuning voor openbare preview te krijgen. Functies moeten worden verwijderd en opnieuw worden gemaakt, niet alleen opnieuw gerealiseerd.

Zie Functieweergaven voor meer informatie over functies.

Wat u moet doen

• Voer een upgrade uit naar 0.16.0. Dit is de vereiste clientversie voor openbare preview-functies (batch en streaming).
• Maak uw functies opnieuw. Bètafunctieweergaven moeten worden verwijderd en opnieuw worden gemaakt, niet opnieuw gerealiseerd, omdat ze niet alle openbare preview-functionaliteit ondersteunen.
• Migreren voordat het venster wordt gesloten. Bestaande bètafuncties moeten vóór 22 juli 2026 worden gemigreerd.

Bèta- en openbare preview-functies identificeren

Openbare preview-functies worden weergegeven als een functieobject in Unity Catalog, bijvoorbeeld in Catalog Explorer. Bètafuncties worden weergegeven als een functie met een YAML-definitie. Een functie die wordt weergegeven als een functie, is een bètafunctie die u moet migreren.

Bètafuncties migreren

Het migreren van een bètafunctie bestaat uit drie onderdelen:

• Maak de functie opnieuw als een openbare preview-functie.
• De functie opnieuw materialiseren, zodat de offline- en onlinetabellen opnieuw worden opgebouwd onder de nieuwe functie.
• Nadat u de gemigreerde functies hebt gecontroleerd, verwijdert u de bètafuncties en de bijbehorende materialisaties.

De functies opnieuw maken

Gebruik list_beta_feature_views deze functie om uw bètafuncties te vinden, Feature.clone() om een niet-geregistreerde kopie te maken en register_feature om elke kopie opnieuw te registreren als een openbare preview-functie. Met klonen worden de registratie, catalogus en het schema gewist, zodat de functie opnieuw kan worden geregistreerd.

Registreer gemigreerde functies met een andere naam of in een ander schema dan de bètafuncties om naamconflicten te voorkomen. In het volgende voorbeeld wordt elke functie opnieuw geregistreerd in het oorspronkelijke schema met een _migrated naamachtervoegsel.

# Update this to the catalog whose beta Feature Views you want to migrate.
CATALOG_TO_MIGRATE = "main"

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 1. Find every beta Feature View in the catalog. Returns Feature objects,
#    scanned across all schemas in the catalog.
beta_features = fe.list_beta_feature_views(catalog_name=CATALOG_TO_MIGRATE)

# Keep each beta feature paired with its migrated counterpart for the next steps.
migrations = []
for beta_feature in beta_features:
    catalog_name, schema_name, leaf_name = beta_feature.full_name.split(".")
    # 2. Clone the feature as an unregistered copy, renamed with a "_migrated" suffix.
    cloned = beta_feature.clone(new_name=f"{leaf_name}_migrated")
    # 3. Re-register the clone as a Public Preview feature.
    migrated = fe.register_feature(
        feature=cloned,
        catalog_name=catalog_name,
        schema_name=schema_name,
    )
    migrations.append((beta_feature, migrated))

De gemigreerde functies opnieuw materialiseren

Als een bètafunctie is gerealiseerd, materialiseert u de openbare preview-tegenhanger opnieuw, zodat de offline- en onlinetabellen opnieuw worden opgebouwd onder de nieuwe functie. Geef de offline- en online winkelconfiguraties op voor de gemigreerde functie en reconstrueer de trigger op basis van de bestaande materialisatie van de bètafunctie.

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

for beta_feature, migrated in migrations:
    # Inspect the beta feature's existing materializations to see what to rebuild and
    # to reconstruct the same trigger.
    trigger = None
    needs_offline = needs_online = False
    for mf in fe.list_materialized_features(feature_name=beta_feature.full_name):
        needs_online = needs_online or bool(mf.is_online)
        needs_offline = needs_offline or not mf.is_online
        # Rebuild the trigger from the materialized feature.
        if mf.cron_schedule_trigger is not None:
            trigger = CronSchedule(
                quartz_cron_expression=mf.cron_schedule_trigger.cron_expression,
                timezone_id="UTC",  # Materialized schedules run in UTC.
            )
        elif mf.table_trigger is not None:
            trigger = TableTrigger()
        elif mf.streaming_mode is not None:
            # Streaming features use StreamingMode, which can be reused as-is.
            trigger = mf.streaming_mode
    if not (needs_offline or needs_online):
        continue  # The beta feature was never materialized.

    catalog_name, schema_name, _ = migrated.full_name.split(".")
    fe.materialize_features(
        features=[migrated],
        offline_config=OfflineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
        )
        if needs_offline
        else None,
        online_config=OnlineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
            online_store_name="my_online_store",
        )
        if needs_online
        else None,
        trigger=trigger,
    )

Note

Als u elke functie in een eigen materialize_features aanroep materialiseert, wordt er een afzonderlijke pijplijn gemaakt. Als u de rekenkosten wilt verlagen, groepeert u functies die een offline- en onlinebestemming delen en activeert u deze in één materialize_features aanroep door ze door te geven in features.

De bètafuncties verwijderen

Warning

Verwijder bètafuncties en hun materialisaties pas nadat u hebt gecontroleerd of de gemigreerde functies en de gerealiseerde gegevens juist zijn. Verwijdering is onomkeerbaar.

Nadat u de gemigreerde functies hebt gecontroleerd, verwijdert u de materialisaties van elke bètafunctie en vervolgens de bètafunctie zelf.

for beta_feature, _ in migrations:
    # Delete the beta feature's materializations first.
    mfs = list(fe.list_materialized_features(feature_name=beta_feature.full_name))
    offline_mfs = [mf for mf in mfs if not mf.is_online]
    if offline_mfs:
        # Aggregation features pair an offline and online table; deleting the offline
        # materialized feature removes its paired online table too.
        for mf in offline_mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    else:
        # Online-only features (ColumnSelection, streaming) have no offline pair; delete
        # the online materialized feature directly.
        for mf in mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    # Then delete the beta feature definition.
    fe.delete_feature(full_name=beta_feature.full_name)