API-referentie voor Feature Views

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 om een feature in een schema te creëren. 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 feature-metadata te lezen. get_feature, create_training_set, en list_materialized_features vereisen READ FEATURE op de eigenschap. Dit privilege verleent geen toegang tot featuregegevens in bron- of gematerialiseerde uitvoertabellen. Om die gegevens voor training of dienst te lezen, moet je ook de relevante tabellen hebben SELECT . READ FEATURE verleend aan een schema of catalogus is van toepassing op alle huidige en toekomstige functies die het bevat.
• MANAGE: Vereist om de levenscyclus en toekenningen van een functie te beheren. Het verwijderen van een feature met delete_feature, en het materialiseren van een feature met materialize_features, vereist MANAGE op de feature. Het verwijderen van een gematerialiseerd kenmerk met delete_materialized_feature wordt niet geregeld door MANAGE: alleen de maker van de gematerialiseerde functie kan het verwijderen.

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

FeatureEngineeringClient.register_feature() 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, RequestSource, or FeatureViewSource
    function: Union[AggregationFunction, ColumnSelection, CustomUDF], # Required for all features
    catalog_name: str,                                     # Required: The catalog name for the feature
    schema_name: str,                                      # Required: The schema name for the feature
    entity: Optional[List[str]] = None,                    # Required for DeltaTableSource and StreamSource
    timeseries_column: Optional[str] = None,               # Required for DeltaTableSource and StreamSource
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

Parameters:

• source: De databron die wordt gebruikt bij feature-computation (DeltaTableSource, StreamSource, RequestSource, of FeatureViewSource).
• function: An AggregationFunction die een operator en een tijdsvenster bundelt, ColumnSelection("column_name") voor doorvoerkenmerken of CustomUDF voor rijgewijze transformaties. Zie Ondersteunde functies voor compatibele brontypes.
• 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 de aggregatie- of opzoeksleutels (primaire sleutels) worden gedefinieerd. Vereist voor DeltaTableSource en StreamSource. Bijvoorbeeld ["user_id"] aggregaties of opzoeken per gebruiker. Laat en RequestSourceFeatureViewSourceweg.
• timeseries_column: De tijdstempelkolom die wordt gebruikt voor tijdvensteraggregatie of selectie van de meest recente waarde. Vereist voor DeltaTableSource en StreamSource. Laat en RequestSourceFeatureViewSourceweg.
• 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. Een functie kan niet worden verwijderd zolang er nog gematerialiseerde functies zijn. Verwijder eerst de gematerialiseerde features en verwijder daarna de feature. 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

FirstN, LastN, , FirstDistincten LastDistinct gebruik de features timeseries_column om invoerrijen te ordenen en een array terug te geven met tot wel waarden n . De n parameter moet een positief geheel getal zijn. FirstN en FirstDistinct selecteer waarden van vroeg tot laatste. LastN en LastDistinct selecteer waarden van het laatste tot het vroegste, en geef vervolgens de geselecteerde waarden in tijdstempelvolgorde terug. FirstDistinct en LastDistinct verwijder dubbele waarden bij het selecteren van waarden in die richting.

Als bijvoorbeeld de bronrijen voor een entiteit geordend zijn als event_time , ["A", "A", "B", "C", "B", "B"]geven de volgende functies terug:

FirstN, LastN, , FirstDistincten LastDistinct vereisen databricks-feature-engineering versie 0.17.0 of later.

CustomUDF

CustomUDFpast een geregistreerde Unity Catalog Python user-defined function (UDF) toe op elke rij. Gebruik het om verzoekinvoer te transformeren of feature-waarden te combineren. Het aggregeert geen rijen en definieert geen tijdsvenster.

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

input_bindings koppelt elke UDF-parameternaam aan een invoer. Voor RequestSource, is de invoer een bronkolomnaam. Voor FeatureViewSource, is het een upstream feature-referentie. Bind elke UDF-parameter, inclusief parameters met standaardinstellingen. Invoertypen moeten exact overeenkomen met de UDF-parametertypen, zonder impliciete numerieke casts. Gebruik scalair input- en returntypen.

Delta-ondersteunde CustomUDF functies kunnen niet online worden gematerialiseerd of aangeboden. Om tabel-ondersteunde featurewaarden te transformeren voor training en servering, definieer je een Delta-ondersteunde aggregatie- of kolomselectiefunctie en verwijs je ernaar via FeatureViewSource.

CustomUDF wordt niet ondersteund met StreamSource. Om de output van een streamingfunctie te transformeren, verwijs je naar die functie via FeatureViewSource.

CustomUDF Met RequestSource vereist databricks-feature-engineering versie 0.17.0 of later.

Om een CustomUDFte gebruiken, heb je het EXECUTE privilege op de UDF, het USE CATALOG privilege op de oudercatalogus en het USE SCHEMA privilege op het ouderschema nodig.

Het volgende voorbeeld gebruikt NumPy om te berekenen log(1 + amount), waardoor de schaal van grote transactiebedragen wordt verminderd. Draai het op serverless compute met aangepaste UDF-afhankelijkheden ingeschakeld. Het main.ecommerce schema moet bestaan.

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

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

Registreer een functie die de verzoekkolom transaction_amount bindt aan de UDF-parameter amount:

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

fe = FeatureEngineeringClient()

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

De UDF ENVIRONMENT configureert afhankelijkheden voor offline berekeningen. Voor online servering kun je ook de pakketten in create_feature_spec(extra_pip_requirements=...) of log_model(extra_pip_requirements=...)opgeven. Ze worden niet automatisch van de UDF gekopieerd. Zie Feature Serving afhankelijkheden en modelafhankelijkheden.

CustomUDF functies kunnen niet worden gematerialiseerd. Request-backed en feature-backed UDF's draaien on demand tijdens training en dienst. Elke UDF in een afhankelijkheidsketen voegt berekeningen toe, dus houd functies en ketens klein. UDF's moeten ontbrekende invoer verwerken, die offline of NaN online kunnen zijnNone.

Voor richtlijnen voor het omgaan met ontbrekende waarden, zie Hoe je ontbrekende featurewaarden kunt omgaan.

ColumnSelection (doorgang)

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.

ColumnSelection ondersteunt de volgende gegevensbronnen:

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

Voor een DeltaTableSource, ColumnSelection ondersteunt filter_condition features en transformation_sql, toegepast vóór de laatste waarde selectie, hetzelfde als aggregatie-features.

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

De filter_condition parameter stelt je in staat om rijen uit de brontabel te filteren voordat je aggregaties berekent of de laatste kolomwaarde selecteert. Deze functie fungeert als een SQL-component WHERE die wordt toegepast voordat gegevens worden gegroepeerd en samengevoegd.

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

Parameters:

• catalog_name, , schema_name: table_nameIdentificeer de Delta-brontabel in Unity Catalog.
• filter_condition: Een SQL-clausule WHERE toegepast vóór aggregatie of kolomselectie. Voorbeeld: "status = 'completed'".
• transformation_sql: Een SQL-expressie SELECT die is toegepast op de brontabel. Gebruik dit om kolommen te hernoemen, types te casten of afgeleide kolommen te berekenen vóór aggregatie of kolomselectie. 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.
• lateness: Een SourceLateness object dat beschrijft hoe lang de bron normaal gesproken nodig heeft om compleet te worden in gebeurtenistijd. Als deze wordt weggelaten, wordt de bron onmiddellijk als volledig beschouwd.

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

SourceLateness.settling_delay is de aanbevolen manier om tijdens het trainen een consistente ETL-vertraging te simuleren die de online materialisatie beïnvloedt. Azure Databricks schuift de geschikte trainingsevaluatietijd met deze duur naar achteren, zodat een trainingsvoorbeeld geen data gebruikt die nog online onderweg zou zijn. Tijdens materialisatie wacht Azure Databricks dezelfde duur voordat een voltooid venster wordt gepubliceerd en dient het laatst voltooide venster in de tussenliggende periode.

Stel bijvoorbeeld dat een dagelijkse ETL-taak 8 uur na middernacht voltooid is in een lokale tijdzone waar middernacht overeenkomt met 07:00 UTC. Gebruik een bezinkingsvertraging van 8 uur en een vensterverschuiving van 7 uur:

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

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

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

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

Ondersteunde transformation_sql expressies

Dezelfde regels gelden voor transformation_sql op en DeltaTableSourceStreamSource.

transformation_sql ondersteunt elke rijgewijs uitdrukking; bewerkingen worden onafhankelijk geëvalueerd voor elke rij. Ze veranderen het aantal rijen of de één-op-één correspondentie met de bron niet. Rijgewijze uitdrukkingen omvatten kolomhernoemingen, gieten, rekenkundige bewerkingen en meer.

Bewerkingen die de vorm of het aantal rijen veranderen worden niet ondersteund, zoals aggregaties zoals SUM() of COUNT(). 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] = None,      # Optional: SQL WHERE clause applied before aggregation
    transformation_sql: Optional[str] = None,    # Optional: SQL SELECT expression for column transformations
    dataframe_schema: Optional[str] = None,      # Required if transformation_sql is set: schema of the resulting DataFrame
    lateness: Optional[SourceLateness] = None,    # Optional: Expected source settling time
)

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'").
• transformation_sql (optioneel): Een SQL-expressie SELECT toegepast vóór aggregatie of kolomselectie, met gebruik van dot-prefix referenties naar de key en value structs. Ondersteunt dezelfde rij-wijze uitdrukkingen als DeltaTableSource. Als deze wordt weggelaten, gebruikt de bron alle kolommen (*).
• dataframe_schema: Het Spark StructType JSON-schema van de geprojecteerde output. Vereist als je .transformation_sql
• lateness: Een SourceLateness object dat beschrijft hoe lang de stroom normaal gesproken duurt om volledig te worden in de gebeurtenistijd. Zie SourceLateness.settling_delay.

from databricks.feature_engineering.entities import StreamSource

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

Leid dataframe_schema af door de projectie tegen de innametafel van de stroom te laten lopen, waardoor de key en value structuren zichtbaar worden.

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

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

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

RequestSource

RequestSource 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 kan worden gebruikt met CustomUDF of ColumnSelection Feature View-functies. 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

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.

FeatureViewSource

FeatureViewSource gebruikt de uitvoer van andere Feature Views als invoer voor een CustomUDF. Het ketenen van kenmerken creëert een gerichte acyclische grafiek (DAG). Een margefunctie kan bijvoorbeeld omzet- en kostenaggregaten combineren, en een andere functie kan de marge transformeren.

Gebruik databricks-feature-engineering versie 0.18.0 of later voor FeatureViewSource.

Geef een lijst van objecten Feature door aan , niet aan featuresfeature-name strings. Haal geregistreerde kenmerken op met get_feature. In input_bindings, gebruik de van elk geregistreerd kenmerk .full_name Voor een lokale, niet-geregistreerde functie gebruik het name in plaats daarvan.

Het volgende voorbeeld gaat uit van twee geregistreerde kenmerken, revenue_sum_7d en cost_sum_7d, die waarden teruggeven DOUBLE door customer_id en gebruiken event_time voor point-in-time berekening:

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

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

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

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

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

De volgende beperkingen zijn van toepassing:

• Alleen CustomUDF wordt ondersteund als functie. Upstream features kunnen aggregaties, kolomselecties of andere CustomUDF features zijn.
• Laat en timeseries_column over de afgeleide eigenschap wegentity. Elke upstream-functie behoudt zijn eigen entiteit, tijdstempel en vensterdefinitie.
• Een functie heeft één bron. Om een verzoekwaarde te combineren met een table-backed feature, definieer je een RequestSource feature en verwijs je beide via FeatureViewSource.
• Elke gedeclareerde upstream-feature moet worden gebruikt in input_bindings. Cycli zijn niet toegestaan.
• Registreer upstream features voordat je de afgeleide feature registreert. Lokale, niet-geregistreerde grafen kunnen worden gebruikt create_training_set voor experimenten.
• Voor training of serving heb je het READ FEATURE of MANAGE privilege nodig op de afgeleide feature en de transitieve upstream-features. Gebruik verschillende featurenamen over de grafiek heen voor registratie en service, zelfs over catalogi of schema's heen.
• Een kenmerk kan verwijzen naar maximaal 20 directe upstream features. Geregistreerde grafen ondersteunen een maximale diepte van vijf features langs een afhankelijkheidspad, inclusief het basisfeature.
• FeatureViewSource kenmerken kunnen niet worden gematerialiseerd of geëvalueerd met compute_features. Gebruik create_training_set ze offline om ze te evalueren. Voor online serveren materialiseer in plaats daarvan de ondersteunde, op tafel gebaseerde upstream-functies.

Voor afhankelijkheidsevaluatie en outputselectie, zie Trainen met FeatureViewSource-functies. Voor implementatie, zie Serve-afgeleide functies.

Trainings- en deductie-API

create_training_set en score_batch de juiste functiewaarden voor een bepaald tijdstip op aanvraag op basis van de brongegevens. Voor functies die ondersteuning bieden voor offline materialisatie, zoals aggregaties van schuifvensters op deltatabelbronnen, verbetert het materialiseren van functies eerst naar een offlinearchief de prestaties van beide bewerkingen. Wanneer gerealiseerde offlinefuncties beschikbaar zijn, lezen de bewerkingen vooraf samengestelde offlinegegevens in plaats van functiewaarden opnieuw te compileren uit de bron. Zie Functieweergaven materialiseren om functies te materialiseren in een offlinearchief.

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

Feature Views ondersteunen vier venstertypen om het terugkijkgedrag voor tijdvenstergebaseerde aggregaties te beheersen. De beschikbare raamtypes hangen af van de bron van de functie: streaming-bron-functies kunnen rol- en zaagtandramen gebruiken, en batch-bron-functies kunnen rolling-, tumbling- en schuiframen gebruiken.

• 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.
• Zaagtandramen houden een lange terugkijk op een streamingbron vers met een hybride batch- en streamingpad. Zie Sawtooth-raam.

De volgende illustratie toont de typen rol-, schuif-, rol- en zaagtandraam.

Tijdvenstertiming

Gebruik delay het om een venster op een eerder analytisch moment te evalueren. Bijvoorbeeld, een venster van 30 dagen met een vertraging van 7 dagen berekent een waarde van 30 dagen per een week vóór de evaluatietijd. delay is onafhankelijk van de aankomsttijd van de bron. Om de tijd te modelleren die de brondata nodig heeft om binnen te komen, configureer SourceLateness.settling_delay je in plaats daarvan.

Wanneer beide settings aanwezig zijn, componeren ze. Azure Databricks behandelt het venster als voltooid na de source settling delay en evalueert het met behulp van de analytische vertraging.

Gebruik offset om de uitlijning van vaste raamgrenzen te veranderen. Standaard zijn tumbling windows en sliding windows uitgelijnd op middernacht UTC. Bijvoorbeeld, een offset van 22 uur brengt een dagelijkse grens op 22:00 UTC. Om grenzen in een lokale tijdzone te benaderen, configureer je een statische offset ten opzichte van UTC. De offset corrigeert geen zomertijd, verschuivt de evaluatietijd niet, en modelleert niet de gegevens die laat aankomen.

De volgende tabel vat de ondersteuning voor deze velden samen:

*Punt: Voor een tumbling-venster moet de offset korter zijn dan window_duration. Voor een schuifvenster moet het korter zijn dan slide_duration.

Begintijd

Gebruik start_time om de vroegste gebeurtenis-tijdgrens in UTC vast te stellen waarop een feature een output kan uitzenden. De grens is inclusief. start_time gates outputs geven. Het beperkt de historische bronrijen die een venster kan lezen niet, en het verandert de uitlijning van het venster niet. Als start_time tussen twee uitgelijnde grenzen valt, is de eerste geschikte vaste-venster-uitgang de volgende grens.

Met start_time, kunnen vensters met vaste duur uitzenden voordat een volledige vensterduur in de bron is verstreken. Deze vroege uitvoer gebruiken de beschikbare brongeschiedenis. Neem bijvoorbeeld een schuifvenster met een éénjaarstermijn window_duration en één dag slide_duration, over een bron waarvan de data begint op 1 januari 2024:

• Zonder start_time, wordt de functie voor het eerst uitgezonden op 1 januari 2025, zodra er een volledig venster van één jaar kan worden gevormd.
• Met start_time de start op 21 augustus 2024 wordt de feature voor het eerst uitgezonden op 21 augustus 2024. Die output bestrijkt alleen de brongeschiedenis die tot nu toe beschikbaar is, vanaf 1 januari 2024. Het venster bereikt zijn volledige éénjaarsperiode op 1 januari 2025 en levert vanaf dat moment volledige outputs op.

Omdat start_time de vensteruitlijning niet verandert, creëert een waarde tussen twee uitgelijnde grenzen geen nieuwe grens. Voor een tumbling window met dagelijkse grenzen om middernacht UTC, wordt een start_time van 06:00 UTC eerst geëmponeerd bij de volgende middernachtgrens. A start_time die precies op een grens landt, emitteert op die grens, omdat de grens inclusief is.

Als start_time niet is ingesteld, kunnen tumbling-vensters en schuifvensters met vaste duur eerst worden gevormd bij een uitgelijnde grens nadat een volledig venster is gevormd. Levenslange schuifvensters en rollende vensters worden verzonden zodra in aanmerking komende brongegevens bestaan.

Voorbeeld:

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

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

Rolraam

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
    start_time: Optional[datetime.datetime] = 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)

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.

# Compute a 7-day value as of one day before the evaluation time
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(days=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.

Gebruik Last om de versheid van een laatste waarde te beperken

Combineer Last met RollingWindow wanneer een laatste waarde slechts voor een beperkte tijd geldig is. Tijdens een evaluatietijd retourneert de functie de waarde van de rij met het meest recente tijdstempel in dit interval:

[evaluation_time - delay - window_duration, evaluation_time - delay)

Als de laatste rij in het interval een nullwaarde bevat, geeft de feature null terug. Als je nul-invoerwaarden wilt uitsluiten, zet dan een filter_condition op de bron.

Deze combinatie verschilt van ColumnSelection. ColumnSelection geeft de laatst waargenomen niet-nulwaarde terug zonder deze te laten verlopen op basis van leeftijd.

Voor batchfuncties heeft deze combinatie een speciale materialisatiemodus die alleen online is. Er wordt alleen DeltaTableSourceondersteuning voor , Last, RollingWindow, en TableTrigger. Zie Materialize freshness-bounded laatste waarden.

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
    delay: Optional[datetime.timedelta] = None
    offset: Optional[datetime.timedelta] = None
    start_time: Optional[datetime.datetime] = None

De volgende tabel bevat de parameters voor een tumbling window.

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

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

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 features die met sliding windows zijn gedefinieerd, worden aggregaties berekend over een venster dat met een schuifinterval vordert. Een schuifvenster kan een vaste duur of een levenslange duur hebben. Vensters met vaste duur overlappen, zodat elk bronevent kan bijdragen aan feature-aggregatie voor meerdere vensters. Een levensduurvenster omvat alle brongebeurtenissen vóór het einde van het venster. Functies op het moment t aggregeren gegevens uit vensters die eindigen op of vóór t (exclusief). Windows is uitgelijnd met het Unix-tijdperk.

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

De volgende tabel bevat de parameters voor een schuifvenster.

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

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

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

Levensduurvenster

Stel window_duration=None in om een levenslang venster te creëren. Bij elke schuifgrens aggregeert het kenmerk alle brongebeurtenissen voor de entiteit met tijdstempels van vóór die grens. Een eendaagse slide levert bijvoorbeeld één keer per dag een cumulatieve waarde op.

Levenslange vensters worden alleen ondersteund door SlidingWindow. RollingWindow en TumblingWindow vereisen een eindige window_duration.

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

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

Zaagtandraam

Een zaagtandraam is een aggregatie die zeer verse updates van recente gebeurtenissen ondersteunt, samen met dagelijkse samenstelling van historische gegevens. De achterste (oudere) rand beweegt zich in vaste, dagelijkse stappen, terwijl de voorste (recente) rand actueel blijft met de laatste gebeurtenissen, waardoor de effectieve raamlengte gedurende de dag "zaagt". Het grootste deel van het venster wordt bediend vanuit data in de invoertabel van de Stream, en alleen de twee meest recente dagen komen uit de livestream. Dit is een compromis dat efficiënt de lange looptijden berekent (schaalbaar tot jaren) terwijl het reageert op verse updates.

Zaagtandramen worden gematerialiseerd op een hybride batch- en stroompad. Een batchpijplijn houdt het grootste deel van het venster vast, terwijl een stroompijplijn de meest recente data in realtime vers houdt. De twee worden bij het lezen samengevoegd, dus voor het model of de consument is het één enkele functie.

Omdat het historische deel van het raam wordt berekend door de batchpijpleiding, is een zaagtandfunctie direct na het begin van de materialisatie klaar om te bedienen, zelfs wanneer het raam maanden of jaren beslaat. Een rollend venster is pas voltooid nadat de volledige periode van het venster is verstreken. Het minimum window_duration moet meer dan twee dagen zijn (de verplichte ondergrens). Databricks raadt een zaagtandvenster aan voor duurder dan 7 dagen. Voor ramen langer dan twee dagen en tot zeven dagen kies je tussen de vaste lengte precisie van een rollend raam en de snellere productiegereedheid van een zaagtandraam.

Om te zien of een zaagtandfunctie klaar is, open je de Feature View in Catalog Explorer. In de sectie Gematerialiseerde features is de batch-backfill voltooid zodra de laatste materialisatietijd van de feature vordert en de status succes aangeeft. Het stroomgedeelte wordt gematerialiseerd door een Lakeflow-declaratieve pijpleiding. Nadat de Feature View de validatie heeft doorstaan, koppelt de gematerialiseerde feature aan die pipeline, waar je de runstatus kunt monitoren.

Zaagtandramen vereisen een StreamSource en worden gematerialiseerd met StreamingMode.

class SawtoothWindow(TimeWindow):
    window_duration: datetime.timedelta

De randen van een zaagtandraam bewegen anders dan die van een rollend raam: de voorrand volgt het laatste verschijnsel, terwijl de achterrand eenmaal per dag naar voren beweegt in plaats van continu. Elke dag bij een vaste afkap van 18:00 UTC schuift de achterrand naar voren naar de UTC-middernachtgrens van die dag. Daardoor is het effectieve venster iets langer window_duration dan en groeit het gedurende de dag voordat het bij de volgende dag weer terugkeert. Training en bediening hanteren dezelfde 18:00 UTC-grens, dus offline training en online bediening blijven consistent.

Zaagtandramen ondersteunen de Sum, Avg, Count, Min, , MaxFirst, VarPopVarSampStddevPopLastenStddevSamp aggregatiefuncties.

Voorbeeld van zaagtandraam

Het volgende voorbeeld toont een 7-daagse telling van transacties van een gebruiker. De voorrand volgt het huidige evenement, terwijl de achterrand dag voor dag vooruit gaat. Voor evenementen op 10 maart gaat het venster terug tot ongeveer 3 maart. Naarmate 10 maart vordert, blijft de voorrand verder vorderen terwijl de achterrand blijft staan, waardoor de overdekte overspanning groeit. Dan, begin 11 maart, loopt de achterrand op tot ongeveer 4 maart. Het effectieve venster is altijd iets langer dan zeven dagen. De twee meest recente dagen worden bediend via de livestream, en de eerste dagen worden bediend vanaf de innametafel van de stream.

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

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

Beperkingen van zaagtandramen

• De delay parameter wordt niet ondersteund.
• SourceLateness.settling_delay wordt niet ondersteund.
• Aggregatiefuncties anders dan Sum, Avg, Count, Min, Max, First, Last, VarPop, , VarSamp, , StddevPop, en StddevSamp worden niet ondersteund (bijvoorbeeld ApproxCountDistinct, ApproxPercentile, , FirstN, LastN, FirstDistinct, en LastDistinct).
• Zaagtandramen vereisen een StreamSource. A DeltaTableSource wordt niet ondersteund.

Materialisatietriggers

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

CronSchedule

Gebruik CronSchedule voor batchaggregatiefuncties. Standaard leidt Azure Databricks een schema af uit het aggregatievenster. Een afgeleid schema houdt rekening met de vensterperiode, het venster delay en offsetde bron, settling_delay zodat een run geen venster publiceert voordat de brongegevens naar verwachting voltooid zijn. Afgeleide schema's ondersteunen tumbling en sliding windows.

Om een afgeleide planning aan te vragen, laat je de cron-uitdrukking weg. CronSchedule() en de expliciete CronSchedule(mode=CronScheduleMode.DERIVED) vorm zijn equivalent:

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

trigger = CronSchedule(mode=CronScheduleMode.DERIVED)

Stel niet in quartz_cron_expression met CronScheduleMode.DERIVED. Wanneer je de gematerialiseerde functie ophaalt, kan het teruggegeven schema de cron-expressie bevatten die Azure Databricks heeft berekend.

Om het schema direct te besturen, geef je een Quartz cron-expressie. CronScheduleMode.MANUAL wordt afgeleid wanneer je een uitdrukking geeft:

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Gebruik TableTrigger voor ColumnSelection features of aggregatiefeatures (AggregationFunction) ondersteund door een DeltaTableSource. De pijplijn wordt uitgevoerd wanneer de upstream Delta-tabel een nieuwe doorvoering ontvangt.

Voor aggregatiefuncties wordt de pipeline zo beperkt dat deze niet op elke commit draait. De pipeline draait hooguit één keer per de helft van de speeltijd, maar nooit vaker dan elke 5 minuten. Bijvoorbeeld, een feature met een tumbling-venster van 1 uur loopt maximaal eens in de 30 minuten, of een feature met een venster van 8 uur maximaal eens in de 4 uur. De 5-minuten vloer geldt wanneer de helft van het venster kleiner is dan dat, dus vensters van 10 minuten of minder lopen hooguit eens in de 5 minuten. Aggregatiefuncties waarvan het venster minder dan 5 minuten is, kunnen niet worden gebruikt TableTrigger, gebruik in plaats daarvan een streamingtrigger.

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

Elke functie gebruikt één trigger; De opties per functietype zijn:

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