Funkciónézetek API-referencia

Hozzáférés-kezelés

A funkciók szabályozható Unity Catalog-objektumok. A szolgáltatáshoz való hozzáférést a CREATE FEATUREUnity READ FEATUREMANAGE Catalog jogosultságai vezérlik. A teljes leírásért tekintse meg a Unity Catalog jogosultságokra vonatkozó hivatkozását.

• CREATE FEATURE — Szolgáltatás létrehozásához szükséges egy sémában. create_feature és register_feature a szülőséma megkövetelése CREATE FEATURE . A minimális jogosultság elvét követve adja meg CREATE FEATURE a sémaszinten; egy katalógusban is megadhatja, hogy a katalógus bármely sémájában létre lehessen hozni a szolgáltatásokat.
• READ FEATURE — Egy funkció és adatainak olvasásához szükséges. get_feature, create_training_setés a materializált funkciók adatainak betanításhoz vagy kiszolgáláshoz való olvasása megköveteli READ FEATURE a funkciót. READ FEATURE a sémán vagy katalóguson megadott összes jelenlegi és jövőbeli funkcióra vonatkozik.
• MANAGE — Egy szolgáltatás életciklusának és támogatásainak kezeléséhez szükséges. Egy funkció törlése a funkcióvaldelete_feature, és egy funkció megvalósítása a szolgáltatással materialize_featuresMANAGE vagy delete_materialized_featureazzal együtt.

Minden funkcióművelethez szükség van USE CATALOG a szülőkatalógusra és USE SCHEMA a szülősémára is. A materializálás módjáról MANAGE és READ FEATURE alkalmazásáról az Engedélyek című témakörben olvashat.

Funkciónézet API

Feature konstruktor és register_feature()

Az ajánlott módszer egy Feature objektum helyi létrehozása és a register_feature Unity Catalogban való megőrzése. Ez a kétlépéses munkafolyamat lehetővé teszi, hogy a regisztráció előtt kísérletezzen a funkciókkal (beleértve azokat is create_training_set).

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

FeatureEngineeringClient.register_feature() regisztrál egy helyileg létrehozott egységet Feature a Unity Catalogban.

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() egyetlen lépésben ellenőrzi, felépíti és azonnal regisztrálja a funkciót a Unity Katalógusban. Ezt akkor használja, ha nem kell helyileg kísérleteznie a funkcióval.

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

Paraméterek:

• source: A funkciószámításban (DeltaTableSourcevagy StreamSourceRequestSource) használt adatforrás.
• function: Az AggregationFunction operátort (például Sum(input="amount")), a bemeneti oszlopot és az időablakot összecsomagolva. Vagy ColumnSelection("column_name") átmenő funkciókhoz.
• catalog_name: A funkció Unity Catalog-katalógusának neve.
• schema_name: A funkció Unity Catalog-sémaneve.
• entity: Az összesítési szintet (elsődleges kulcsokat) meghatározó oszlopnevek listája. Az összesítési funkciókhoz szükséges. Például ["user_id"] a felhasználónkénti összesítések.
• timeseries_column: Az időablak-összesítéshez használt időbélyeg-oszlop. Az összesítési funkciókhoz szükséges.
• name: Választható funkciónév. Ha nincs megadva, automatikusan létrejön a bemeneti oszlopból, függvényből és ablakból (például amount_avg_rolling_7d).
• description: A funkció opcionális leírása.

Visszatér: Érvényesített funkciópéldány

Felveti: ValueError, ha bármelyik ellenőrzés sikertelen

delete_feature()

Egy funkció törlése a Unity Katalógusból a teljes név alapján.

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

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

A szolgáltatás törlése előtt távolítsa el vagy frissítse az arra hivatkozó modelleket vagy szolgáltatásjellemzőket. Ha a funkció materializált, először törölje a materializált funkciót. Tekintse meg a materializált funkciók törlését ismertető témakört.

Automatikusan létrehozott nevek

Ha name nincs megadva, a rendszer automatikusan létrehoz egy nevet. A létrehozott nevek a következő mintát követik: {column}_{function}_{window}. Például:

• price_avg_rolling_1h (1 órás átlagos ár)
• transaction_count_rolling_30d_1d (30 napos tranzakciószám az esemény időbélyegétől számított 1d késleltetéssel)

Támogatott függvények

Összesítő függvények

ColumnSelection (átmenő)

ColumnSelection egyetlen oszlopot jelöl ki egy forrásból összesítés alkalmazása nélkül. Közvetlenül a function paraméterbe van csomagolva (nem belül AggregationFunction). A visszatérési típus a forrássémából származik.

ColumnSelection bármilyen adatforrással használható:

• DeltaTableSource: Az entitáskulcsonkénti legújabb értéket adja vissza egy időponthoz kötött illesztéssel (nincs visszatekintő ablak összesítése).
• StreamSource: Az entitáskulcsonkénti legújabb értéket adja vissza a streamből (nincs visszatekintő ablak összesítése).
• RequestSource: A következtetési időpontban megadott értéket adja át (vagy a betanításkor kinyeri a címkézett DataFrame-ből).

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

Példa: összesítési és oszlopkijelölési funkciók

Az alábbi példa az ugyanazon adatforráson definiált funkciókat mutatja be.

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

Szűrési feltételekkel rendelkező funkciók

A filter_condition paraméter lehetővé teszi a sorok szűrését a forrástáblából az összesítések kiszámítása előtt . Ez az adatok csoportosítása és összesítése előtt alkalmazott SQL-záradékként WHERE működik.

A szűrők akkor hasznosak, ha nagy méretű forrástáblákkal dolgoznak, amelyek a funkciószámításhoz szükséges adatok szuperhalmazát tartalmazzák, és minimálisra csökkentik a különálló nézetek létrehozásának szükségességét ezen táblák tetején.

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

Adatforrások

DeltaTableSource

DeltaTableSource egy rövid ideig Python objektum, amellyel meghatározható, hogyan számíthatók ki a szolgáltatások a forrástáblából. Nem hoz létre új táblát. Megadja az adatok olvasásának és a funkciók összesítésének konfigurációját.

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
)

Paraméterek:

• catalog_name, , schema_nametable_name: Azonosítsa a forrás Delta táblát a Unity Catalogban.
• filter_condition: Az összesítés előtt alkalmazott SQL-záradék WHERE . Példa: "status = 'completed'".
• transformation_sql: A forrástáblára alkalmazott SQL-kifejezés SELECT . Ezzel az összesítés előtt átnevezheti az oszlopokat, az öntött típusokat vagy a számítási származtatott oszlopokat. Ha nincs megadva, az összes oszlop ki van jelölve (*). Példa: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: Az eredményként kapott DataFrame sémája átalakítások után, Spark StructType JSON formátumban (forrás).df.schema.json() Kötelező, ha transformation_sql meg van adva. Ez jelzi a rendszernek az átalakításból eredő oszlopneveket és típusokat.

Ha mindkettő filter_conditiontransformation_sql be van állítva, az eredmény a következő lesz: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

Példa: Oszlopátalakítások használata transformation_sql

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

Példa: PySpark DataFrame származtatása transformation_sql és dataframe_schema származtatása

Az átalakítást PySpark-lekérdezésként is megírhatja, majd kinyerheti a sémát az eredményként kapott DataFrame-ből:

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

DeltaTableSource.from_sql()

Kényelmesen létrehozhat egy DeltaTableSource SQL-lekérdezést. A metódus elemzi a lekérdezést, hogy automatikusan kinyerje a tábla nevét, transformation_sqlés filter_condition.

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

Csak az egyszerű SELECT ... FROM ... [WHERE ...] lekérdezések támogatottak. A rendszer elutasítja az összetett SQL-eket (JOIN-eket, alquerieseket, CTE-ket, UNION-okat). Összetett lekérdezések esetén közvetlenül DeltaTableSource az és transformation_sqla 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",
)

Iterálás a to_dataframe()

A funkciószámításhoz használt adatok előnézetének megtekintésére szolgál source.to_dataframe() . Ez akkor hasznos, ha az iterálás után filter_condition és transformation_sql a várt eredmény eléréséig tart.

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

Az entitások ismertetése

Az entitásoszlopok határozzák meg a funkciók összesítési szintjét. Ezek a Feature definícióban vannak megadva, nem a következőn DeltaTableSource: . Az entitások a következőt határozzák meg:

• Az adatok csoportosítása: A funkciók az entitásértékek egyedi kombinációja alapján vannak összesítve (az SQL-hez GROUP BY hasonlóan)
• Az elsődleges kulcsstruktúra: Minden egyedi entitáskombináció egy sor számítási funkciót eredményez

Példa: Ügyfélszintű funkciók

Az alábbi kód az ügyfél szintjén összesíti a funkciókat (ügyfélenként egy sort):

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

Példa: Ügyfél-áruház szintű funkciók

Ha részletesebb szinten szeretné összesíteni a funkciókat (ügyfél-áruház kombinációnként egy sort), használjon több entitásoszlopot:

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

Ha az összesítés különböző szintjein (például ügyfélszintű és ügyfélszintű) funkciókra van szüksége, használjon különböző entity értékeket a funkciódefiníciókban. Ugyanez DeltaTableSource megosztható a különböző entitáskonfigurációkkal rendelkező funkciók között.

StreamSource

StreamSource egy Streamre hivatkozik. A Stream a streamforrás kapcsolati, hitelesítési, séma- és betöltési konfigurációját tartalmazza. A Kafka esetében a funkciódefiníciókban szereplő oszlophivatkozásokat előtaggal value. kell megadni, vagy key. meg kell jelölni, hogy az üzenet melyik részét kell elolvasni.

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

Paraméterek:

• full_name: A Stream teljes háromrészes neve (például "my_catalog.my_schema.my_stream").
• filter_condition (nem kötelező): Az adatok összesítés előtti adatfolyamára alkalmazott SQL-záradék WHERE pontelőtagú oszlophivatkozások használatával (például "value.event_type = 'purchase'").

from databricks.feature_engineering.entities import StreamSource

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

RequestSource

RequestSource A kérelem hasznos adataiban következtetési időben megadott adatok sémáját határozza meg, nem pedig előre materializált táblából való megkeresést. A betanítás során a rendszer kinyeri ezeket az oszlopokat a címkézett DataFrame-ből create_training_set. A modell kiszolgálása során a hívónak tartalmaznia kell őket a HTTP-kérés hasznos adatai között.

RequestSource ColumnSelection használatával (közvetlenül egy érték átengedéséhez). Nem támogatja az összesítő függvényeket és az időablakokat.

A séma meghatározása

Adja meg a sémát objektumok listájáként FieldDefinition , mindegyikben egy oszlopnevet és egy 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),
    ]
)

Támogatott adattípusok

RequestSourcetámogatja a következőben ScalarDataTypedefiniált skaláris típusokat: INTEGER, FLOAT, , BOOLEAN, STRINGDOUBLELONG, . TIMESTAMPDATESHORT Az olyan összetett típusok, mint a tömbök, a térképek és a szerkezetek nem támogatottak.

A kérelemadatok hidratált állapota

Modell aláírása

Ha egy modellt funkciókkal rendelkező log_model betanítási készlettel naplóznakRequestSource, a rendszer kötelező bemenetként hozzáadja az RequestSource oszlopokat az MLflow-modell aláírásához. Ez azt jelenti, hogy a kiszolgáló végpont API-sémája tükrözi, hogy a hívóknak milyen mezőket kell megadniuk következtetési időpontban.

Betanítási és következtetési API

create_training_set()

Létrehoz egy betanítási adatkészletet az időponthoz kötött helyes funkciószámítással. További információ: Modellek betanítása funkciónézetekkel.

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

Naplóz egy modellt a szolgáltatás metaadataival a leállás nyomon követéséhez és az automatikus funkciókereséshez a következtetés során. További információ: Modellek betanítása funkciónézetekkel.

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

Offline kötegelt következtetést hajt végre automatikus funkciókereséssel. A modellben tárolt szolgáltatás-metaadatok használatával pontos funkciókat számít ki, biztosítva a betanítással való konzisztenciát.

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

A bemeneti DataFrame-nek tartalmaznia kell a betanítás során használt entitás- és idősoroszlopokat. A rendszer automatikusan kiszámítja a funkciókat a forrásadatokból.

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

Időablakok

A funkciónézetek három különböző ablaktípust támogatnak az időablak-alapú összesítések visszatekintési viselkedésének szabályozásához: gördülés, csúszás és csúszás.

• A gördülő ablakok visszatekintenek az esemény idejéből. Az időtartam és a késés explicit módon van meghatározva.
• A csúszó ablakok rögzítettek, átfedésmentes időablakok. Minden adatpont pontosan egy ablakhoz tartozik.
• A tolóablakok átfedésben vannak, a gördülő időablakok konfigurálható diaintervallummal.

Az alábbi ábra bemutatja, hogyan működnek.

Gördülő ablak

A gördülő ablakok up-todátum- és valós idejű összesítések, amelyek általában streamelési adatokon keresztül használhatók. A streamelési folyamatokban a gördülő ablak csak akkor bocsát ki új sort, ha a rögzített hosszúságú ablak tartalma megváltozik, például amikor egy esemény belép vagy távozik. Ha egy gördülőablak-funkciót használnak a betanítási folyamatokban, a rendszer pontos időponthoz kötött funkciószámítást végez a forrásadatokon az adott esemény időbélyegét közvetlenül megelőző rögzített hosszúságú időtartam használatával. Ez segít megelőzni az online kapcsolat nélküli eltérést vagy az adatszivárgást. Az időponthoz kapcsolódó T funkciók összesítik az eseményeket [T − időtartam, T).

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

Az alábbi táblázat egy gördülő ablak paramétereit sorolja fel. Az ablak kezdési és befejezési ideje az alábbi paramétereken alapul:

• Kezdési idő: evaluation_time - window_duration - delay (beleértve)
• Befejezési idő: evaluation_time - delay (kizárólagos)

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

Az alábbi kód használatával határozzon meg egy gördülő ablakot késleltetéssel.

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

Példák a gördülő ablakra

• window_duration=timedelta(days=7): Ez egy 7 napos visszatekintési ablakot hoz létre, amely az aktuális kiértékelési időpontban fejeződik be. A 7. napon 14:00-kor kezdődő esemény esetén ez magában foglalja az összes eseményt a 0. napon 14:00 órától a 7. napon 14:00-ig, de ezt az időpontot nem beleértve.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Ez egy 1 órás visszatekintési ablakot hoz létre, amely 30 perccel a kiértékelési idő előtt ér véget. Egy 15:00-kor kezdődő esemény esetén ez magában foglalja az összes eseményt 13:30-tól egészen (de nem beleértve) 14:30-ig. Ez hasznos az adatbetöltési késések figyelembe vételéhez.

Csúszó ablak

Az időt átfedésmentesen felosztó, fix hosszúságú ablakkal definiált funkciók esetében az aggregációk előre meghatározott időtartamú ablakokra vonatkoznak, amelyek egy csúszási intervallummal lépnek tovább. Ennek eredményeképpen a forrás minden eseménye pontosan egy ablakhoz járul hozzá. Az t időpontnál a funkciók összesítik az adatokat az olyan ablakokból, amelyek t előtt végződnek (kizárva). A Windows a Unix-korszaktól indul.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

Az alábbi táblázat egy gördülő ablak paramétereit sorolja fel.

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

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

Guruló ablak példa

• window_duration=timedelta(days=5): Ez előre meghatározott rögzített hosszúságú, egyenként 5 napos ablakokat hoz létre. Példa: Az 1. ablak a 0. naptól a 4. napig, a 2. ablak az 5. naptól a 9. napig, a 3. ablak a 10. naptól a 14. napig terjed stb. Az 1. ablak az összes olyan eseményt tartalmazza, amelynek időbélyegei a 0. napon kezdődnek 00:00:00.00 , legfeljebb (de nem beleértve) az 5. napon beállított időbélyeggel 00:00:00.00 rendelkező eseményeket. Minden esemény pontosan egy ablakhoz tartozik.

Csúszó ablak

A tolóablakokkal definiált funkciók esetében az aggregációk egy előre meghatározott rögzített hosszúságú ablakban vannak kiszámítva, amely egy diaintervallummal halad előre, átfedésben lévő ablakokat hozva létre. A forrás minden eseménye hozzájárulhat több ablak funkcióösszesítéséhez. Az t időpontnál a funkciók összesítik az adatokat az olyan ablakokból, amelyek t előtt végződnek (kizárva). A Windows a Unix-korszaktól indul.

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

Az alábbi táblázat egy tolóablak paramétereit sorolja fel.

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

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

CSúszóablak példa

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Ez átfedésben lévő 5 napos ablakokat hoz létre, amelyek minden alkalommal 1 nappal haladnak előre. Példa: Az 1. ablak a 0. naptól a 4. napig, a 2. ablak az 1. naptól az 5. napig, a 3. ablak a 2. naptól a 6. napig terjed stb. Minden ablak tartalmazza az eseményeket a kezdő naptól 00:00:00.00 a záró napig (de nem beleértve) 00:00:00.00 is. Mivel az ablakok átfedésben vannak, egyetlen esemény több ablakhoz is tartozhat (ebben a példában minden esemény legfeljebb 5 különböző ablakhoz tartozik).

Materializálási triggerek

Aktiválja a materializálási folyamat futásának vezérlését. Az eseményindító típusa a funkció típusától függ.

CronSchedule

Összesítési funkciókhoz használható CronSchedule (AggregationFunction). A folyamat egy quartz cron kifejezés által meghatározott rögzített ütemezésen fut.

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Olyan funkciókhoz használható TableTrigger , ColumnSelection amelyeket egy DeltaTableSource. A folyamat akkor fut, amikor a felsőbb rétegbeli Delta-tábla új véglegesítést kap.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

Olyan funkciókhoz használható StreamingMode , amelyeket egy StreamSource. A folyamat folyamatos streamelési folyamatként fut.

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

Eseményindító kiválasztása

Nem lehet olyan funkciókat létrehozni, amelyek különböző eseményindító-típusokat igényelnek egyetlen materialize_features hívásban. Ehelyett külön hívásokat kell kibocsátanak.

Bétaverziós funkciók migrálása nyilvános előzetes verzióra

A Funkciónézetek nyilvános előzetes verziója első osztályú funkcióentititásokat vezet be a Unity Katalógusban, amelyeket a CREATE FEATURE jogosultságok szabályoznakREAD FEATURE, és a 0.16.0-s vagy újabb verziót igényli.databricks-feature-engineering A bétaverzió során létrehozott funkciók (a 0.15.0-s verzióval) Unity Catalog-függvényekként vannak tárolva, és nem támogatják az összes nyilvános előzetes verziójú funkciót. A hosszú távú nyilvános előzetes verzió támogatásához hozza létre újra a bétaverziós funkciókat a 0.16.0-s verzióval. A szolgáltatásokat nem csak újra kell törölni és újra létrehozni.

A funkciókról további információt a Funkciónézetek című témakörben talál.

Mit kell tenned

• Frissítsen a 0.16.0-ra. Ez a nyilvános előzetes verziójú funkciókhoz (köteg és streamelés) szükséges ügyfélverzió.
• Hozza létre újra a funkciókat. A bétaverziós funkciónézeteket törölni kell és újra létre kell hozni, nem kell újra materializálni, mert nem támogatják az összes nyilvános előzetes verziójú funkciót.
• Migrálás az ablak bezárása előtt. A meglévő bétaszolgáltatásokat 2026. július 22-e előtt kell áttelepíteni.

Bétaverziós és nyilvános előzetes verziójú funkciók azonosítása

A nyilvános előzetes verziójú funkciók funkcióobjektumként jelennek meg a Unity Catalogban, például a Katalóguskezelőben. A bétaszolgáltatások YAML-definícióval rendelkező függvényként jelennek meg. A függvényként ábrázolt összes funkció egy bétaverziós funkció, amelyet át kell telepítenie.

Bétaverziós funkciók migrálása

A bétaverziós funkció migrálása három részből áll:

• Hozza létre újra a funkciót nyilvános előzetes verziójú funkcióként.
• A funkció újravalósítása, így offline és online táblái újraépülnek az új funkció alatt.
• A migrált funkciók ellenőrzése után törölje a bétaszolgáltatásokat és azok materializációit.

A funkciók ismételt létrehozása

A list_beta_feature_views bétaverziós funkciók megkeresésére, Feature.clone() a nem regisztrált példányok létrehozására és register_feature az egyes példányok újbóli regisztrálására használható nyilvános előzetes verziójú funkcióként. A klónozás törli a regisztrációt, a katalógust és a sémát, hogy a funkció újra regisztrálható legyen.

A névütközések elkerülése érdekében regisztrálja a migrált szolgáltatásokat más néven vagy a bétaverziós funkcióktól eltérő sémában. Az alábbi példa újra regisztrálja az eredeti sémában lévő összes funkciót egy _migrated név utótaggal.

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

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

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

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

A migrált funkciók újravalósítása

Ha egy bétaverziós funkció megvalósult, a nyilvános előzetes verzió megfelelőjét újra kell létrehoznia, hogy offline és online táblái újraépülhessenek az új funkció alatt. Adja meg a migrált funkció offline és online áruházbeli konfigurációit, és rekonstruálja az eseményindítót a bétaverzió meglévő materializációjából.

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

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

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

A bétaverzió funkcióinak törlése

A migrált funkciók ellenőrzése után törölje az egyes bétafunkciók materializációit, majd magát a béta funkciót.

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