API-referens för Feature Views

Åtkomstkontroll

Funktioner är styrningsbara Unity Catalog-objekt. Åtkomst till en funktion styrs av behörigheterna CREATE FEATURE, READ FEATUREoch MANAGE Unity Catalog. Fullständiga beskrivningar finns i Referens för behörigheter för Unity Catalog.

• CREATE FEATURE: Krävs för att skapa en funktion i ett schema. create_feature och register_feature kräver CREATE FEATURE på det överordnade schemat. Efter principen om minsta behörighet beviljar CREATE FEATURE du på schemanivå. Du kan också bevilja den i en katalog så att du kan skapa funktioner i alla scheman i katalogen.
• READ FEATURE: Krävs för att läsa funktionsmetadata. get_feature, create_training_set, och list_materialized_features kräver READ FEATURE på funktionen. Denna privilegie ger inte tillgång till funktionsdata i käll- eller materialiserade utmatningstabeller. För att läsa dessa data för utbildning eller tjänst måste du också ha SELECT på de relevanta tabellerna. READ FEATURE som beviljats för ett schema eller en katalog gäller för alla aktuella och framtida funktioner som den innehåller.
• MANAGE: Krävs för att hantera en funktions livscykel och bidrag. Att ta bort en funktion med delete_feature, och materialisera en funktion med materialize_features, kräver MANAGE på funktionen. Att ta bort en materialiserad funktion med delete_materialized_feature styrs inte av MANAGE: endast skaparen av den materialiserade funktionen kan ta bort den.

Alla funktionsåtgärder kräver USE CATALOG också i den överordnade katalogen och USE SCHEMA i det överordnade schemat. Information om hur MANAGE och READ FEATURE gäller för materialisering finns i Behörigheter.

API för funktionsvy

Feature konstruktor och register_feature()

Den rekommenderade metoden är att konstruera ett Feature objekt lokalt och använda register_feature för att spara det i Unity Catalog. Med det här tvåstegsarbetsflödet kan du experimentera med funktioner (inklusive create_training_set) innan du registrerar dem.

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() registrerar en lokalt konstruerad Feature i 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() validerar, skapar och registrerar omedelbart en funktion i Unity Catalog i ett enda steg. Använd detta när du inte behöver experimentera med funktionen lokalt först.

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: Datakällan som används vid funktionsberäkning (DeltaTableSource, StreamSource, , RequestSourceeller FeatureViewSource).
• function: An AggregationFunction som buntar en operator och ett tidsfönster, ColumnSelection("column_name") för genomgångsfunktioner eller CustomUDF för radvisa transformationer. Se Stödda funktioner för kompatibla källkodstyper.
• catalog_name: Unity Catalog-katalognamnet för funktionen.
• schema_name: Unity Catalog-schemanamnet för funktionen.
• entity: Lista över kolumnnamn som definierar aggregerings- eller uppslagsnycklarna (primära nycklar). Krävs för DeltaTableSource och StreamSource. Till exempel ["user_id"] aggregerar eller söker efter per användare. Utelämna för RequestSource och FeatureViewSource.
• timeseries_column: Tidsstämpelkolumnen som används för aggregering av tidsfönster eller val av senaste värde. Krävs för DeltaTableSource och StreamSource. Utelämna för RequestSource och FeatureViewSource.
• name: Valfritt funktionsnamn. Om det utelämnas genereras automatiskt från indatakolumnen, funktionen och fönstret (till exempel amount_avg_rolling_7d).
• description: Valfri beskrivning av funktionen.

Returnerar: En validerad funktionsinstans

Höjer: ValueError om verifieringen misslyckas

delete_feature()

Tar bort en funktion från Unity Catalog med dess fullständigt kvalificerade namn.

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

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

Innan du tar bort en funktion tar du bort eller uppdaterar modeller eller funktionsspecifikationer som refererar till den. En funktion kan inte tas bort medan den fortfarande har materialiserade funktioner. Ta bort de materialiserade funktionerna först, sedan ta bort funktionen. Se Så här tar du bort en materialiserad funktion.

Namn som genereras automatiskt

När name utelämnas genereras ett namn automatiskt. Genererade namn följer mönstret: {column}_{function}_{window}. Ett exempel:

• price_avg_rolling_1h (Genomsnittspris på 1 timme)
• transaction_count_rolling_30d_1d (30 dagars antal transaktioner med 1d fördröjning från händelsetidsstämpeln)

Funktioner som stöds

Sammansättningsfunktioner

FirstN, LastN, , och LastDistinct använder funktionerna timeseries_column för att ordna inmatningsrader och returnera en array med upp till n värdenFirstDistinct. Parametern n måste vara ett positivt heltal. FirstN och FirstDistinct väljer värden från tidigast till senast. LastN och LastDistinct väljer värden från senaste till tidigaste, och returnerar sedan de valda värdena i tidsstämpelordning. FirstDistinct och LastDistinct ta bort dubbletter när man väljer värden i den riktningen.

Till exempel, om källraderna för en entitet är ordnade efter event_time som ["A", "A", "B", "C", "B", "B"], returnerar följande funktioner:

FirstN, LastN, , FirstDistinctoch LastDistinct kräver databricks-feature-engineering version 0.17.0 eller senare.

CustomUDF

CustomUDFapplicerar en registrerad Unity Catalog Python-användardefinierad funktion (UDF) på varje rad. Använd den för att transformera begärda indata eller kombinera funktionsvärden. Den aggregerar inte rader eller definierar något tidsfönster.

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

input_bindings mappar varje UDF-parameternamn till en indata. För RequestSource, är indatan ett källkolumnnamn. För FeatureViewSource, är det en uppströms funktionsreferens. Bind varje UDF-parameter, inklusive parametrar med standardinställningar. Indatatyper måste exakt matcha UDF-parametertyperna, utan implicita numeriska casts. Använd skalära in- och returtyper.

Delta-backade CustomUDF funktioner kan inte materialiseras eller serveras online. För att transformera tabellstödda funktionsvärden för träning och serving, definiera en Delta-backad aggregering eller kolumnvalsfunktion och referera till den genom FeatureViewSource.

CustomUDF stöds inte med StreamSource. För att transformera en streamingfunktion sin utgång, referera till den funktionen genom FeatureViewSource.

CustomUDF kräver RequestSourcedatabricks-feature-engineering version 0.17.0 eller senare.

För att använda en CustomUDF, behöver EXECUTE du privilegiet på UDF, privilegiet USE CATALOG på dess föräldrakatalog och USE SCHEMA privilegiet på dess föräldraschema.

Följande exempel använder NumPy för att beräkna log(1 + amount), vilket minskar skalan för stora transaktionsbelopp. Kör det på serverless compute med anpassade UDF-beroenden aktiverade. Schemat main.ecommerce måste existera.

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

Registrera en funktion som binder förfrågningskolumnen transaction_amount till UDF-parametern 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",
)

UDF:s konfigurerar ENVIRONMENT beroenden för offline-beräkning. För online-servering, deklarera även paketen i create_feature_spec(extra_pip_requirements=...) eller log_model(extra_pip_requirements=...). De kopieras inte automatiskt från UDF. Se Feature Serving ochmodellberoenden.

CustomUDF funktioner kan inte materialiseras. Request-backade och feature-backade UDF:er körs on demand under utbildning och tjänstgöring. Varje UDF i en beroendekedja lägger till beräkningar, så håll funktioner och kedjor små. UDF:er måste hantera saknade ingångar, som kan vara None offline eller NaN online.

För vägledning om hur man hanterar saknade värden, se Hur man hanterar saknade funktionsvärden.

ColumnSelection (genompassning)

ColumnSelection väljer en enskild kolumn från en källa utan att tillämpa någon aggregering. Den omsluts direkt i parametern function (inte inuti AggregationFunction). Returtypen härleds från källschemat.

ColumnSelection stöder följande datakällor:

• DeltaTableSource: Returnerar det senaste värdet per entitetsnyckel via en punkt-i-tid-koppling (ingen lookback-fönsteraggregering).
• StreamSource: Returnerar det senaste värdet per entitetsnyckel från strömmen (ingen aggregering av återblicksfönster).
• RequestSource: Passerar genom värdet som anges vid inferens (eller extraheras från den märkta DataFrame vid träningstillfället).

För en DeltaTableSource, ColumnSelection stöder filter_condition funktioner och transformation_sql, tillämpad före det senaste värdevalet, samma som aggregeringsfunktioner.

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

Exempel: sammansättnings- och kolumnvalsfunktioner

I följande exempel visas funktioner som definierats över samma datakälla.

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

Funktioner med filtervillkor

Parametern filter_condition låter dig filtrera rader från källtabellen innan du beräknar aggregeringar eller väljer det senaste kolumnvärdet. Det här fungerar som en SQL-sats WHERE som tillämpas innan du grupperar och aggregerar data.

Filter är användbara när du arbetar med stora källtabeller som innehåller en supermängd data som behövs för funktionsberäkning och minimerar behovet av att skapa separata vyer ovanpå dessa tabeller.

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

Datakällor

DeltaTableSource

DeltaTableSource är ett tillfälliga Python objekt som används för att definiera hur funktioner beräknas från en källtabell. Den skapar ingen ny tabell. Den anger konfigurationen för att läsa data och aggregeringsfunktioner.

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_name: Identifiera deltatabellen för källan i Unity Catalog.
• filter_condition: En SQL-klausul WHERE som tillämpas före aggregering eller kolumnval. Exempel: "status = 'completed'".
• transformation_sql: Ett SQL-uttryck SELECT som tillämpas på källtabellen. Använd detta för att byta namn på kolumner, kasta typer eller beräkna härledda kolumner innan aggregering eller kolumnval. Om det utelämnas markeras alla kolumner (*). Exempel: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: Schemat för resulterande DataFrame efter transformeringar, i Spark StructType JSON-format (från df.schema.json()). Krävs om transformation_sql anges. Detta talar om för systemet vilka kolumnnamn och typer som är resultatet av omvandlingen.
• lateness: Ett SourceLateness objekt som beskriver hur lång tid källan normalt tar att bli komplett i händelsetid. Om den utelämnas anses källan vara fullständig omedelbart.

När både filter_condition och transformation_sql anges är den resulterande frågan: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

SourceLateness.settling_delay är det rekommenderade sättet att simulera under träning en konsekvent ETL-fördröjning som påverkar onlinematerialisering. Azure Databricks flyttar den kvalificerade träningsutvärderingstiden bakåt med denna tid så att ett träningsexempel inte använder data som fortfarande skulle ha varit under överföring online. Under materialiseringen väntar Azure Databricks samma tid innan det publicerar ett färdigt fönster och tjänar det senast färdiga fönstret under den mellanliggande perioden.

Till exempel, anta att ett dagligt ETL-jobb slutförs 8 timmar efter midnatt i en lokal tidszon där midnatt motsvarar 07:00 UTC. Använd en sättningsfördröjning på 8 timmar och en fönsterförskjutning på 7 timmar:

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

Exempel: Använda transformation_sql för kolumntransformeringar

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

Exempel: Härleda transformation_sql och dataframe_schema från en PySpark DataFrame

Du kan skriva omvandlingen som en PySpark-fråga och sedan extrahera schemat från den resulterande 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(),
)

Stödda transformation_sql uttryck

Samma regler gäller för transformation_sql på DeltaTableSource och StreamSource.

transformation_sql stöder alla radvisa uttryck; operationer utvärderas oberoende för varje rad. De ändrar inte antalet rader eller den en-till-en-korrespondensen med källan. Radvisa uttryck inkluderar kolumnomnamn, avgjutningar, aritmetiska operationer och mer.

Operationer som ändrar formen eller radantalet stöds inte, såsom aggregeringar som SUM() eller COUNT(). Använd AggregationFunction i funktionsdefinitionen i stället.

DeltaTableSource.from_sql()

Som en bekvämlighet kan du skapa en DeltaTableSource från en SQL-fråga. Metoden parsar frågan för att automatiskt extrahera tabellnamnet, transformation_sql, och filter_condition.

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

Endast enkla SELECT ... FROM ... [WHERE ...] frågor stöds. Komplex SQL (JOIN, underfrågor, CTE: er, UNION) avvisas. För komplexa frågor skapar du DeltaTableSource direkt med transformation_sql och 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",
)

Iterera med to_dataframe()

Använd source.to_dataframe() för att förhandsgranska de data som ska användas för funktionsberäkning. Detta är användbart när du itererar på filter_condition och transformation_sql tills de ger de förväntade resultaten.

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

Förstå entiteter

Entitetskolumner definierar aggregeringsnivån för dina funktioner. De anges i Feature definitionen, inte på DeltaTableSource. Entiteter avgör:

• Gruppera data: Funktioner aggregeras per unik kombination av entitetsvärden (ungefär GROUP BY som i SQL)
• Den primära nyckelstrukturen: Varje unik entitetskombination resulterar i en rad med beräknade funktioner

Exempel: Funktioner på kundnivå

Följande kod aggregerar funktioner på kundnivå (en rad per kund):

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

Exempel: Funktioner på kundnivå

Om du vill aggregera funktioner på en mer detaljerad nivå (en rad per kundarkivkombination) använder du flera entitetskolumner:

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

När du behöver funktioner på olika aggregeringsnivåer (till exempel på kundnivå och kundlagringsnivå) använder du olika entity värden i dina funktionsdefinitioner. Samma DeltaTableSource sak kan delas mellan funktioner med olika entitetskonfigurationer.

StreamSource

StreamSource refererar till en Stream. Dataströmmen innehåller konfiguration av anslutning, autentisering, schema och inmatning för strömningskällan. För Kafka måste kolumnreferenser i funktionsdefinitioner föregås av eller value. ange vilken del av meddelandet som ska läsaskey..

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: Det fullständiga tredelade namnet på en stream (till exempel "my_catalog.my_schema.my_stream").
• filter_condition (valfritt): En SQL-sats WHERE som används för att strömma data före aggregering med hjälp av kolumnreferenser med punktprefix (till exempel "value.event_type = 'purchase'").
• transformation_sql (valfritt): Ett SQL-uttryck SELECT som appliceras före aggregering eller kolumnval, med prickprefixreferenser till key och value strukturerna. Stöder samma radvisa uttryck som DeltaTableSource. Om den utelämnas använder källkoden alla kolumner (*).
• dataframe_schema: Spark StructType JSON-schemat för den projicerade utgången. Obligatoriskt om du sätter transformation_sql.
• lateness: Ett SourceLateness objekt som beskriver hur lång tid strömmen normalt tar att bli komplett i händelsetid. Se även 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'",
)

Härled dataframe_schema genom att köra projektionen mot strömmens intagningsbord, vilket exponerar och keyvalue strukturerar.

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 definierar ett schema för data som tillhandahålls vid inferens i nyttolasten för begäran i stället för att sökas upp från en fördefinierad tabell. Under träningen extraheras dessa kolumner från den märkta DataFrame som skickas till create_training_set. Under modellservern måste anroparen inkludera dem i nyttolasten för HTTP-begäran.

RequestSource kan användas med funktioner som CustomUDF eller ColumnSelection Feature View. Den stöder inte aggregeringsfunktioner eller tidsfönster.

Definiera schemat

Definiera schemat som en lista över FieldDefinition objekt, där var och en anger ett kolumnnamn och en 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),
    ]
)

Datatyper som stöds

RequestSourcestöder skalära typer som definierats i ScalarDataType: INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, , DATE. SHORT Komplexa typer som matriser, kartor och structs stöds inte.

Hur begärandedata är hydratiserade

Modellsignatur

När en modell loggas med en log_model träningsuppsättning som innehåller RequestSource funktioner läggs kolumnerna RequestSource till i MLflow-modellsignaturen som nödvändiga indata. Det innebär att serverdelsslutpunktens API-schema återspeglar vilka fält som anropare måste ange vid inferens.

FeatureViewSource

FeatureViewSource använder utdata från andra Feature Views som indata till en CustomUDF. Att kedja funktioner skapar en riktad acyklisk graf (DAG). Till exempel kan en marginalfunktion kombinera intäkts- och kostnadsaggregat, och en annan funktion kan omvandla marginalen.

Använd databricks-feature-engineering version 0.18.0 eller senare för FeatureViewSource.

Skicka en lista med Feature objekt till features, inte funktionsnamnsträngar. Hämta registrerade funktioner med get_feature. I input_bindings, använd varje registrerad egenskaps full_name. För en lokal, oregistrerad funktion, använd dess name istället.

Följande exempel antar två registrerade funktioner, revenue_sum_7d och cost_sum_7d, som returnerar DOUBLE värden genom customer_id och använder event_time för tidpunktsberäkning:

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

Följande begränsningar gäller:

• Endast CustomUDF stöds som funktionen. Upstream-funktioner kan vara aggregeringar, kolumnval eller andra CustomUDF funktioner.
• Utelämna entity och timeseries_column på den härledda egenskapen. Varje uppströms funktion behåller sin egen entitet, tidsstämpel och fönsterdefinition.
• En funktion har en källa. För att kombinera ett begäranvärde med en tabellstödd funktion, definiera en RequestSource funktion och referera båda genom FeatureViewSource.
• Varje deklarerad uppströms funktion måste användas i input_bindings. Cykler är inte tillåtna.
• Registrera uppströms funktioner innan du registrerar den härledda egenskapen. Lokala, oregistrerade grafer kan användas för create_training_set experiment.
• För träning eller serving behöver READ FEATURE du eller MANAGE privilegiet på den härledda egenskapen och dess transitiva uppströmsfunktioner. Använd distinkta funktionsnamn över hela grafen för registrering och serv, även över kataloger eller scheman.
• En funktion kan referera till upp till 20 direkta uppströmsfunktioner. Registrerade grafer stöder ett maximalt djup på fem funktioner längs en beroendebana, inklusive basfunktionen.
• FeatureViewSource Egenskaper kan inte materialiseras eller utvärderas med compute_features. Använd create_training_set dem för att utvärdera dem offline. För online-servering, materialisera istället de stödda bordsstödda upstream-funktionerna.

För beroendeutvärdering och utdataval, se Träna med funktioner i FeatureViewSource. För utrullning, se Serve-härledda funktioner.

API för utbildning och slutsatsdragning

create_training_set och score_batch beräkna korrekta funktionsvärden för tidpunkt på begäran från källdata. För funktioner som stöder offlinematerialisering, till exempel glidande fönsteraggregeringar på deltatabellkällor, förbättrar materialisering av funktioner först till ett offlinearkiv prestandan för båda åtgärderna. När materialiserade offlinefunktioner är tillgängliga läser åtgärderna fördefinierade offlinedata i stället för att omberäkna funktionsvärden från källan. Se Materialisera funktionsvyer för att materialisera funktioner till en offlinebutik.

create_training_set()

Skapar en träningsdatauppsättning med rätt funktionsberäkning vid tidpunkt. Mer information finns i Träna modeller med funktionsvyer.

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

Loggar en modell med funktionsmetadata för ursprungsspårning och automatisk funktionssökning under slutsatsdragning. Mer information finns i Träna modeller med funktionsvyer.

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

Utför batchinferens offline med automatisk funktionssökning. Använder funktionsmetadata som lagras med modellen för att beräkna rätt funktioner vid tidpunkt, vilket säkerställer konsekvens med träningen.

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

Indatadataramen måste innehålla kolumnerna entitet och tidsserie som används under träningen. Funktioner beräknas automatiskt från källdata.

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

Tidsfönster

Feature Views stöder fyra fönstertyper för att styra återblicksbeteende för tidsfönsterbaserade aggregeringar. Vilka fönstertyper som finns tillgängliga beror på funktionens källa: streaming-källkodsfunktioner kan använda rullande och sågtands-fönster, och batch-källfunktioner kan använda rullande, tumbling och skjutande fönster.

• Rullande fönster ser tillbaka från händelsetiden. Varaktighet och fördröjning definieras uttryckligen.
• Rullande fönster är fasta, icke-överlappande tidsfönster. Varje datapunkt tillhör exakt ett fönster.
• Glidande fönster är överlappande, glidande tidsfönster med ett konfigurerbart glidintervall.
• Sawtooth-fönster håller ett långt tillbakablicksfönster fräscht över en strömningskälla med en hybrid batch- och strömningsväg. Se Sawtooth-fönstret.

Följande illustration visar typerna av tumbling-, skjut-, rull- och sågtandsfönster.

Tidsfönstertiming

Använd delay för att utvärdera ett fönster vid en tidigare analytisk tidpunkt. Till exempel beräknar ett 30-dagarsfönster med 7-dagars fördröjning ett 30-dagarsvärde en vecka före utvärderingstiden. delay är oberoende av källans ankomsttid. För att modellera tiden som källdata tar att anlända, konfigurera SourceLateness.settling_delay istället.

När båda miljöerna finns med, komponerar de. Azure Databricks behandlar fönstret som komplett efter source settling-fördröjningen och utvärderar det med hjälp av den analytiska fördröjningen.

Använd offset för att ändra justeringen av fasta fönstergränser. Som standard är tumbling-fönster och glidande fönster justerade mot midnatt UTC. Till exempel justerar en offset på 22 timmar en daglig gräns till 22:00 UTC. För att approximera gränser i en lokal tidszon, konfigurera en statisk förskjutning i förhållande till UTC. Offset justerar inte för sommartid, förskjuter inte utvärderingstiden eller modellerar data som anländer sent.

Följande tabell sammanfattar stödet för dessa fält:

*Punkt: För ett tumblingfönster måste förskjutningen vara kortare än window_duration. För ett skjutfönster måste det vara kortare än slide_duration.

Starttid

Använd start_time för att sätta den tidigaste händelse-tidsgränsen i UTC där en funktion kan sända ut en utdata. Gränsen är inkluderande. start_time Gates utgångar. Den begränsar inte de historiska källrader som ett fönster kan läsa, och den ändrar inte fönsterjusteringen. Om start_time den ligger mellan två justerade gränser är den första berättigade utdata med fast fönster nästa gräns.

Med start_time, kan fönster med fast duration sända ut innan en full fönstervaraktighet har förflutit i källan. Dessa tidiga utdata använder den källhistorik som finns tillgänglig. Till exempel, betrakta ett glidande fönster med ett år window_duration och ett dag slide_duration, över en källa vars data börjar den 1 januari 2024:

• Utan start_time, sänds funktionen först den 1 januari 2025, när ett helt ettårsfönster kan bildas.
• Med speltid till den 21 augusti 2024 släpps start_time funktionen första gången den 21 augusti 2024. Den informationen täcker endast den källhistorik som hittills finns tillgänglig, från och med 1 januari 2024. Fönstret når sin fulla ettårsperiod den 1 januari 2025 och producerar fullständiga resultat därefter.

Eftersom start_time inte ändrar fönsterjustering, skapar ett värde mellan två justerade gränser inte en ny gräns. För ett tumblingfönster med dagliga gränser vid midnatt UTC sänder en start_time klocka 06:00 UTC först vid nästa midnattsgräns. A start_time som landar exakt på en gräns emitterar vid den gränsen, eftersom gränsen är inkluderande.

Om start_time är osatt, kan tumblingfönster och fastvaraktiga glidande fönster först sändas ut vid en justerad gräns efter att ett helt fönster bildats. Livstidsglidande och rullande fönster sänder ut så snart berättigad källdata finns.

Ett exempel:

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

Rullande fönster

Rullande fönster är up-to- datum- och realtidsaggregeringar, som vanligtvis används över strömmande data. I strömmande pipelines genererar det rullande fönstret endast en ny rad när innehållet i fönstret med fast längd ändras, till exempel när en händelse kommer in eller lämnar. När en rullande fönsterfunktion används i träningspipelines utförs en korrekt funktionsberäkning för tidpunkt på källdata med varaktigheten för fönstret med fast längd omedelbart före en viss händelses tidsstämpel. Detta hjälper till att förhindra online-offline skevhet eller dataläckage. Funktioner vid tidpunkten T aggregerar händelser från [T − varaktighet, T).

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

I följande tabell visas parametrarna för ett rullande fönster. Start- och sluttiderna för fönstret baseras på följande parametrar:

• Starttid: evaluation_time - window_duration - delay (inklusive)
• Sluttid: evaluation_time - delay (exklusiv)

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

Definiera ett rullande fönster med fördröjning med hjälp av koden nedan.

# Compute a 7-day value as of one day before the evaluation time
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(days=1)
)

Exempel på rullande fönster

• window_duration=timedelta(days=7): Detta skapar ett 7-dagars tillbakablicksfönster som slutar vid den aktuella utvärderingstiden. För ett evenemang kl. 14:00 på dag 7 inkluderar detta alla händelser från 14:00 på dag 0 upp till (men inte inklusive) 14:00 på dag 7.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Detta skapar ett 1-timmars lookback-fönster som slutar 30 minuter före utvärderingstiden. För ett evenemang kl. 15:00 inkluderar detta alla händelser från 13:30 till (men inte inklusive) 14:30.

Använd Last för att begränsa färskheten för ett senaste värde

Kombinera Last med RollingWindow när ett senaste värde är giltigt endast under en begränsad tid. Vid en utvärderingstid returnerar funktionen värdet från raden med den senaste tidsstämpeln i detta intervall:

[evaluation_time - delay - window_duration, evaluation_time - delay)

Om den senaste raden i intervallet innehåller ett nollvärde, returnerar funktionen null. Om du vill utesluta nollindata, sätt a filter_condition på källan.

Denna kombination skiljer sig från ColumnSelection. ColumnSelection returnerar det senaste observerade icke-nollvärdet utan att det går ut baserat på ålder.

För batchfunktioner har denna kombination ett speciellt materialiseringsläge endast online. Den stöder endast DeltaTableSource, Last, RollingWindow, och TableTrigger. Se Materialize freshness-bounded senaste värden.

Rullande fönster

För funktioner som definieras med fallande fönster beräknas aggregeringar över ett förutbestämt fönster med fast längd som avancerar med ett förskjutningsintervall, vilket ger icke-överlappande fönster som delar upp tiden fullt ut. Därför bidrar varje händelse i källan till exakt ett fönster. Vid tidpunkten t aggregeras data från fönster som avslutas vid eller före t (exklusivt). Windows börjar vid Unix-epoken.

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

I följande tabell visas parametrarna för ett rullande fönster.

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

Exempel på rullande fönster

• window_duration=timedelta(days=5): Detta skapar förutbestämda fönster med fast längd på 5 dagar vardera. Exempel: Fönster nr 1 sträcker sig över dag 0 till dag 4, fönster 2 sträcker sig över dag 5 till dag 9, fönster 3 sträcker sig över dag 10 till dag 14 och så vidare. Mer specifikt innehåller Fönster nr 1 alla händelser med tidsstämplar som börjar på 00:00:00.00 dag 0 upp till (men inkluderar inte) några händelser med tidsstämpel 00:00:00.00 på dag 5. Varje händelse tillhör exakt ett fönster.

Skjutfönster

För funktioner definierade med glidande fönster beräknas aggregeringar över ett fönster som avancerar med ett glidintervall. Ett glidande fönster kan ha antingen en fast varaktighet eller en livstidsperiod. Fönster med fast duration överlappar, så varje källhändelse kan bidra till funktionsaggregering för flera fönster. Ett livstidsfönster inkluderar alla källhändelser innan fönstret slutar. Vid tidpunkten t aggregeras data från fönster som avslutas vid eller före t (exklusivt). Windows är anpassade till Unix-epoken.

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

I följande tabell visas parametrarna för ett skjutfönster.

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

Exempel på skjutfönster

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Detta skapar överlappande 5-dagars fönster som avancerar med 1 dag varje gång. Exempel: Fönster nr 1 sträcker sig över dag 0 till dag 4, fönster 2 sträcker sig över dag 1 till dag 5, fönster 3 sträcker sig över dag 2 till dag 6 och så vidare. Varje fönster innehåller händelser från 00:00:00.00 startdagen upp till (men inte inklusive) 00:00:00.00 på slutdagen. Eftersom windows överlappar varandra kan en enskild händelse tillhöra flera fönster (i det här exemplet tillhör varje händelse upp till 5 olika fönster).

Livslängdsfönster

Ställ window_duration=None in för att skapa ett livstidsfönster. Vid varje rutschgräns aggregerar funktionen alla källhändelser för enheten med tidsstämplar före den gränsen. Till exempel ger en endagsslide ett kumulativt värde en gång per dag.

Livstidsfönster stöds endast av SlidingWindow. RollingWindow och TumblingWindow kräver en ändlig 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",
)

Sågtandsfönster

Ett sågtandsfönster är en aggregering som stödjer mycket färska uppdateringar för senaste händelser tillsammans med daglig kompaktering av historisk data. Dess bakre (äldre) kant rör sig i fasta, dagliga steg medan den främre (nyare) kanten håller sig uppdaterad med de senaste händelserna, så den effektiva fönsterlängden "sågas" under varje dag. Största delen av fönstret serveras från data i strömmens inmatningstabell, och endast de två senaste dagarna kommer från livesändningen. Detta är en kompromiss som effektivt beräknar långvariga fönster (skalar till år) samtidigt som den är responsiv på färska uppdateringar.

Sågtandsfönster materialiseras på en hybrid batch- och strömningsväg. En batchpipeline upprätthåller huvuddelen av fönstret, medan en streamingpipeline håller den senaste datan färsk i realtid. De två slås ihop vid läsning, så för modellen eller konsumenten är det en enda funktion.

Eftersom den historiska delen av fönstret beräknas av batchpipelinen är en sågtandsfunktion redo att användas kort efter att materialiseringen påbörjats, även när fönstret sträcker sig över månader eller år. Ett rullande fönster är bara färdigt efter att dess fulla fönstertid har löpt ut. Minimikravet window_duration måste vara mer än två dagar (den obligatoriska nedre gränsen). Databricks rekommenderar ett sågtandsfönster som varar längre än 7 dagar. För fönster längre än två dagar och upp till sju dagar, välj mellan den fasta precisionen hos ett rullfönster och den snabbare produktionsberedskapen hos ett sågtandsfönster.

För att se om en sågtandsfunktion är klar, öppna Funktionsvyn i Katalogutforskaren. I avsnittet Materialiserade funktioner är batchfyllningen klar när funktionens sista materialiseringstid har gått framåt och dess status visar framgång. Den strömningsdelen materialiseras av en Lakeflow-deklarativ pipeline. Efter att Feature View har klarat valideringen länkar den materialiserade funktionen till den pipelinen, där du kan övervaka dess körstatus.

Sågtandsfönster kräver en StreamSource och materialiseras med StreamingMode.

class SawtoothWindow(TimeWindow):
    window_duration: datetime.timedelta

Kanterna på ett sågtandsfönster rör sig annorlunda än på ett rullande fönster: framkanten följer den senaste händelsen, medan bakkanten rör sig en gång per dag istället för kontinuerligt. Varje dag vid en fast avgränsning klockan 18:00 UTC går bakkanten fram till dagens gräns mellan UTC och midnatt. Som ett resultat är det effektiva fönstret något längre window_duration än och växer under dagen innan det återkommer en dag vid nästa avbrott. Träning och servering använder samma 18:00 UTC-gräns, så offlineutbildning och onlineservering är konsekventa.

Sågtandsfönster stödjer Sum, Avg, Count, Min, Max, , First, VarSampVarPopStddevPopLastochStddevSamp aggregationsfunktionerna.

Exempel med sågtandsfönster

Följande exempel visar en 7-dagars räkning av en användares transaktioner. Framkanten följer den aktuella händelsen medan den bakre kanten går fram en dag i taget. För evenemang den 10 mars sträcker sig fönstret tillbaka till cirka 3 mars. När den 10 mars går fortsätter framkanten att avancera medan bakkanten håller, så det täckta spannet växer. Sedan, i början av den 11 mars, går bakkanten till ungefär den 4 mars. Det effektiva fönstret är alltid lite längre än sju dagar. De två senaste dagarna serveras från livesändningen, och de tidigare dagarna serveras från strömmens intagsbord.

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

Begränsningar för sågtandsfönster

• Parametern delay stöds inte.
• SourceLateness.settling_delay stöds inte.
• Aggregionsfunktioner andra än Sum, Avg, Count, Min, Max, LastFirstVarSampStddevPopVarPopoch StddevSamp stöds inte (till exempel ApproxCountDistinct, ApproxPercentile, , FirstN, LastN, FirstDistinct, och ).LastDistinct
• Sågtandsfönster kräver en StreamSource. A DeltaTableSource stöds inte.

Materialiseringsutlösare

Utlöser kontroll när en materialiseringspipeline körs. Utlösartypen beror på funktionstypen.

CronSchedule

Användning CronSchedule för batchaggregeringsfunktioner. Som standard härleder Azure Databricks ett schema från aggregeringsfönstret. Ett härlett schema tar hänsyn till fönsterperioden, fönstret delay och offset, samt källan settling_delay så att en körning inte publicerar ett fönster innan dess källdata förväntas vara klar. Härledda scheman stödjer tumbling och glidande fönster.

För att begära ett härlett schema, utelämna cron-uttrycket. CronSchedule() och den explicita CronSchedule(mode=CronScheduleMode.DERIVED) formen är ekvivalenta:

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

trigger = CronSchedule(mode=CronScheduleMode.DERIVED)

Sätt inte med quartz_cron_expressionCronScheduleMode.DERIVED. När du hämtar den materialiserade funktionen kan det returnerade schemat innehålla cron-uttrycket som Azure Databricks beräknade.

För att styra schemat direkt, ge ett kvarts-cron-uttryck. CronScheduleMode.MANUAL härleds när du anger ett uttryck:

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Användning TableTrigger för ColumnSelection funktioner eller aggregeringsfunktioner (AggregationFunction) som backas upp av en DeltaTableSource. Pipelinen körs när den överordnade Delta-tabellen tar emot en ny incheckning.

För aggregeringsfunktioner är pipelinen strypt så att den inte körs på varje commit. Pipelinen körs högst en gång per halva långfilmens fönsterlängd, men aldrig oftare än var femte minut. Till exempel körs en film med ett tumbling-fönster på 1 timme högst en gång var 30:e minut, eller en film med 8-timmarsfönster som mest en gång var fjärde timme. 5-minutersgolvet gäller när halva fönstret är mindre än så, så fönster som är 10 minuter eller mindre löper högst var 5:e minut. Aggregeringsfunktioner vars fönster är under 5 minuter kan inte användas TableTrigger, använd istället en strömningstrigger.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

Använd StreamingMode för funktioner som backas upp av en StreamSource. Pipelinen körs som en pipeline för kontinuerlig direktuppspelning.

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

Välja en utlösare

Varje funktion använder en trigger; Alternativen efter funktionstyp är:

Du kan inte materialisera funktioner som kräver olika utlösartyper i ett enda materialize_features anrop. Utfärda separata anrop i stället.