Referência da API Feature Views

Controlo de acesso

As funcionalidades são objetos governáveis do Catálogo Unity. O acesso a uma funcionalidade é controlado pelos CREATE FEATUREprivilégios de , READ FEATURE, e MANAGE do Catálogo Unity. Para descrições completas, consulte a referência de privilégios do Unity Catalog.

• CREATE FEATURE — Necessário para criar uma funcionalidade num esquema. create_feature e register_feature exigir CREATE FEATURE no esquema pai. Seguindo o princípio do privilégio mínimo, conceda CREATE FEATURE ao nível do esquema; também pode concedê-lo num catálogo para permitir a criação de funcionalidades em qualquer esquema desse catálogo.
• READ FEATURE — Obrigado a ler uma característica e os seus dados. get_feature, create_training_set, e a leitura de dados materializados de funcionalidades para treino ou serviço necessário READ FEATURE sobre a funcionalidade. READ FEATURE concedido num esquema ou catálogo aplica-se a todas as funcionalidades atuais e futuras que contém.
• MANAGE — Necessário para gerir o ciclo de vida e os subsídios de uma funcionalidade. Eliminar uma funcionalidade com delete_feature, e materializar uma característica com materialize_features ou delete_materialized_feature, requisito MANAGE sobre a funcionalidade.

Todas as operações de funcionalidades também exigem USE CATALOG o catálogo pai e USE SCHEMA o esquema pai. Para saber como MANAGE e READ FEATURE aplicar à materialização, veja Permissões.

API de Visualização de Funcionalidades

Feature construtor e register_feature()

A abordagem recomendada é construir um Feature objeto localmente e usá-lo register_feature para persistir no Unity Catalog. Este fluxo de trabalho em dois passos permite-lhe experimentar funcionalidades (incluindo create_training_set) antes de as registar.

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() regista um localmente construído Feature no 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() valida, constrói e regista imediatamente uma funcionalidade no Unity Catalog num único passo. Usa isto quando não precisares de experimentar a funcionalidade localmente primeiro.

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

Parâmetros:

• source: A fonte de dados usada no cálculo de características (DeltaTableSource, StreamSource, ou RequestSource).
• function: E AggregationFunction que agrupa o operador (por exemplo, Sum(input="amount")), a coluna de entrada e a janela temporal. Ou ColumnSelection("column_name") para funcionalidades de passagem.
• catalog_name: O nome do catálogo do Catálogo Unity para a funcionalidade.
• schema_name: O nome do esquema do Catálogo Unity para a funcionalidade.
• entity: Lista de nomes de colunas que definem o nível de agregação (chaves primárias). Obrigatório para funcionalidades de agregação. Por exemplo, ["user_id"] agregados por utilizador.
• timeseries_column: A coluna de carimbo temporal usada para agregação de janelas temporais. Obrigatório para funcionalidades de agregação.
• name: Nome opcional da funcionalidade. Se omitido, gerado automaticamente a partir da coluna de entrada, função e janela (por exemplo, amount_avg_rolling_7d).
• description: Descrição opcional da funcionalidade.

Retornos: Uma instância de Feature validada

Aumentos: ValueError se alguma validação falhar

delete_feature()

Apaga uma funcionalidade do Unity Catalog pelo seu nome totalmente qualificado.

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

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

Antes de eliminar uma funcionalidade, remova ou atualize quaisquer modelos ou especificações que a referenciam. Se a funcionalidade já foi materializada, elimine-a primeiro. Veja Como eliminar uma funcionalidade materializada.

Nomes gerados automaticamente

Quando name é omitido, um nome é automaticamente gerado. Os nomes gerados seguem o padrão: {column}_{function}_{window}. Por exemplo:

• price_avg_rolling_1h (Preço médio de 1 hora)
• transaction_count_rolling_30d_1d (Contagem de 30 dias de transação com atraso de 1d desde o carimbo temporal do evento)

Funções suportadas

Funções de agregação

ColumnSelection (passagem)

ColumnSelection seleciona uma única coluna de uma fonte sem aplicar qualquer agregação. Está envolto diretamente no function parâmetro (não dentro AggregationFunctionde ). O tipo de retorno é inferido a partir do esquema fonte.

ColumnSelection pode ser usado com qualquer fonte de dados:

• DeltaTableSource: Devolve o valor mais recente por chave de entidade através de uma junção em ponto no tempo (sem agregação de janela de retorno).
• StreamSource: Devolve o valor mais recente por chave de entidade do fluxo (sem agregação de janela de retorno).
• RequestSource: Passa pelo valor fornecido no momento da inferência (ou extraído do DataFrame rotulado no momento do treino).

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

Exemplo: funcionalidades de agregação e seleção de colunas

O exemplo seguinte mostra características definidas sobre a mesma fonte de dados.

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

Funcionalidades com condições de filtro

O filter_condition parâmetro permite filtrar linhas da tabela de origem antes de calcular as agregações. Isto funciona como uma cláusula SQL WHERE que é aplicada antes de agrupar e agregar dados.

Os filtros são úteis quando se trabalha com grandes tabelas de origem que incluem um superconjunto de dados necessários para o cálculo de características, e minimizam a necessidade de criar vistas separadas sobre essas tabelas.

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

Fontes de dados

DeltaTableSource

DeltaTableSource é um objeto Python efémero usado para definir como as características são calculadas a partir de uma tabela fonte. Não cria uma nova tabela. Especifica a configuração para leitura de dados e agregação de características.

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
)

Parâmetros:

• catalog_name, schema_name, table_name: Identifique a tabela Delta de origem no Unity Catalog.
• filter_condition: Uma cláusula SQL WHERE aplicada antes da agregação. Exemplo: "status = 'completed'".
• transformation_sql: Uma expressão SQL SELECT aplicada à tabela de origem. Use isto para renomear colunas, tipos de cast ou calcular colunas derivadas antes da agregação. Se omitido, todas as colunas são selecionadas (*). Exemplo: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: O esquema do DataFrame resultante após transformações, no formato JSON Spark StructType (de df.schema.json()). Necessário se transformation_sql for fornecido. Isto indica ao sistema os nomes das colunas e tipos que resultam da sua transformação.

Quando tanto filter_condition como transformation_sql estão definidos, a consulta resultante é: SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

Exemplo: Usar transformation_sql para transformações de coluna

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

Exemplo: derivar transformation_sql e dataframe_schema a partir de um DataFrame PySpark

Pode escrever a sua transformação como uma consulta PySpark e depois extrair o esquema do DataFrame resultante:

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

Por conveniência, pode criar um DeltaTableSource a partir de uma consulta SQL. O método analisa a consulta para extrair automaticamente o nome da tabela, transformation_sql, e filter_condition.

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

Apenas consultas simples SELECT ... FROM ... [WHERE ...] são suportadas. SQL complexo (JOINs, subconsultas, CTEs, UNIONs) é rejeitado. Para consultas complexas, constrói-se DeltaTableSource diretamente com transformation_sql e 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",
)

Iterar com to_dataframe()

Use source.to_dataframe() para pré-visualizar os dados que serão usados para o cálculo de características. Isto é útil para iterar sobre filter_condition e transformation_sql até que produzam os resultados esperados.

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

Compreensão das entidades

As colunas de entidades definem o nível de agregação das suas funcionalidades. São especificados na Feature definição, não em DeltaTableSource. As entidades determinam:

• Como os dados são agrupados: As funcionalidades são agregadas por combinação única de valores de entidade (semelhante ao GROUP BY SQL)
• A principal estrutura de chaves: Cada combinação única de entidades resulta numa linha de características computadas

Exemplo: Funcionalidades ao nível do cliente

O seguinte código agrega funcionalidades ao nível do cliente (uma linha por cliente):

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

Exemplo: Funcionalidades ao nível do cliente-loja

Para agregar funcionalidades a um nível mais detalhado (uma linha por combinação cliente-loja), use múltiplas colunas de entidade:

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

Quando precisar de funcionalidades em diferentes níveis de agregação (por exemplo, ao nível do cliente e ao nível do cliente-loja), use valores diferentes entity nas suas definições de funcionalidades. O mesmo DeltaTableSource pode ser partilhado entre funcionalidades com diferentes configurações de entidades.

StreamSource

StreamSource referência a um Stream. O Stream contém a configuração de ligação, autenticação, esquema e ingestão para a fonte de streaming. Para Kafka, as referências às colunas nas definições de características devem ser precedidas por value. ou key. para indicar qual parte da mensagem ler.

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

Parâmetros:

• full_name: O nome completo em três partes de um Stream (por exemplo, "my_catalog.my_schema.my_stream").
• filter_condition (opcional): Uma cláusula SQL WHERE aplicada a dados de fluxo antes da agregação, usando referências a colunas com prefixo de pontos (por exemplo, "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 define um esquema para dados que são fornecidos no momento da inferência na carga útil do pedido, em vez de serem consultados a partir de uma tabela pré-materializada. Durante o treino, estas colunas são extraídas do DataFrame rotulado passado para create_training_set. Durante o serviço de modelos, o chamador deve incluí-los na carga útil de pedido HTTP.

RequestSource é usado com ColumnSelection (para passar diretamente por um valor). Não suporta funções de agregação nem janelas temporais.

Definição do esquema

Defina o esquema como uma lista de FieldDefinition objetos, cada um especificando um nome de coluna e um 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),
    ]
)

Tipos de dados suportados

RequestSourcesuporta os tipos escalares definidos em ScalarDataType: INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, DATE, . SHORT Tipos complexos como arrays, mapas e structs não são suportados.

Como os dados dos pedidos são hidratados

Modelo de assinatura

Quando um modelo é registado com log_model um conjunto de treino que inclui RequestSource características, as RequestSource colunas são adicionadas à assinatura do modelo MLflow como entradas necessárias. Isto significa que o esquema da API do endpoint de serviço reflete quais os campos que os chamadas devem fornecer no momento da inferência.

API de treino e inferência

create_training_set()

Cria um conjunto de dados de treino com cálculo de características corretas em ponto no tempo. Para mais detalhes, veja Modelos de Comboios com Vistas de Características.

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

Regista um modelo com metadados de características para rastreio de linhagem e pesquisa automática de características durante a inferência. Para mais detalhes, veja Modelos de Comboios com Vistas de Características.

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

Realiza inferência em lote offline com pesquisa automática de funcionalidades. Utiliza os metadados das funcionalidades armazenadas com o modelo para calcular as funcionalidades de correção pontual no tempo, garantindo consistência no treino.

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

O DataFrame de entrada deve conter as colunas da entidade e da série temporal usadas durante o treino. As funcionalidades são calculadas automaticamente a partir dos dados de origem.

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

Janelas temporais

As Feature Views suportam três tipos diferentes de janelas para controlar o comportamento de lookback em agregações baseadas em janelas temporais: rolamento, tumbling e deslizamento.

• Janelas rolantes olham para trás da hora do evento. A duração e o atraso são explicitamente definidos.
• As janelas de rotação são janelas temporais fixas e não sobrepostas. Cada ponto de dados pertence exatamente a uma janela.
• As janelas deslizantes são janelas de tempo sobrepostas e rolantes com um intervalo de deslizamento configurável.

A ilustração seguinte mostra como funcionam.

Janela de correr

As janelas rolantes são agregados up-todata e tempo real, normalmente usados sobre dados em streaming. Em pipelines de streaming, a janela rolante emite uma nova linha apenas quando o conteúdo da janela de comprimento fixo muda, como quando um evento entra ou sai. Quando uma funcionalidade de janela rolante é usada em pipelines de treino, é realizado um cálculo preciso de uma funcionalidade em ponto no tempo sobre os dados de origem usando a duração da janela de comprimento fixo imediatamente anterior ao carimbo temporal de um evento específico. Isto ajuda a prevenir o desvio online-offline ou fuga de dados. As características no tempo T agregam eventos a partir de [T − duração, T).

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

A tabela seguinte lista os parâmetros para uma janela rolante. Os tempos de início e fim das janelas baseiam-se nestes parâmetros da seguinte forma:

• Hora de início: evaluation_time - window_duration - delay (inclusivo)
• Hora de término: evaluation_time - delay (exclusivo)

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

Defina uma janela rolante com atraso usando o código abaixo.

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

Exemplos de janelas rolantes

• window_duration=timedelta(days=7): Isto cria uma janela de retrospetiva de 7 dias que termina no momento atual da avaliação. Para um evento às 14:00 do Dia 7, isto inclui todos os eventos desde as 14:00 do Dia 0 até (mas não incluindo) as 14:00 do Dia 7.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): Isto cria uma janela de retrospetiva de 1 hora que termina 30 minutos antes da hora da avaliação. Para um evento às 15:00, isto inclui todos os eventos das 13:30 até (mas não incluindo) 14:30. Isto é útil para compensar atrasos na ingestão de dados.

Janela de tombo

Para características definidas usando janelas de rotação, as agregações são calculadas sobre uma janela de comprimento fixo pré-determinada que avança por um intervalo de deslizamento, produzindo janelas não sobrepostas que particionam totalmente o tempo. Como resultado, cada evento na fonte contribui para exatamente uma janela. As características no momento t agregam dados de janelas que terminam em ou antes de t (exclusivo). O Windows começa na época Unix.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

A tabela seguinte lista os parâmetros para uma janela de tumbling.

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

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

Exemplo da janela que cai

• window_duration=timedelta(days=5): Isto cria janelas de duração fixa pré-determinadas de 5 dias cada. Exemplo: A Janela #1 abrange do Dia 0 ao Dia 4, a Janela #2 abrange o Dia 5 ao Dia 9, a Janela #3 abrange o Dia 10 ao Dia 14, e assim sucessivamente. Especificamente, a Janela #1 inclui todos os eventos com carimbo temporal a partir 00:00:00.00 do Dia 0 até (mas não incluindo) quaisquer eventos com carimbo 00:00:00.00 temporal no Dia 5. Cada evento pertence exatamente a uma janela.

Janela deslizante

Para características definidas usando janelas deslizantes, as agregações são calculadas sobre uma janela de comprimento fixo pré-determinada que avança por um intervalo de deslizamento, produzindo janelas sobrepostas. Cada evento na fonte pode contribuir para a agregação de funcionalidades em várias janelas. As características no momento t agregam dados de janelas que terminam em ou antes de t (exclusivo). O Windows começa na época Unix.

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

A tabela seguinte lista os parâmetros para uma janela deslizante.

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

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

Exemplo da janela deslizante

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): Isto cria janelas sobrepostas de 5 dias que avançam 1 dia a cada vez. Exemplo: A Janela #1 abrange o Dia 0 ao Dia 4, a Janela #2 abrange o Dia 1 ao Dia 5, a Janela #3 abrange o Dia 2 ao Dia 6, e assim sucessivamente. Cada janela inclui eventos desde 00:00:00.00 o dia de início até (mas não incluindo) 00:00:00.00 o dia final. Como as janelas se sobrepõem, um único evento pode pertencer a várias janelas (neste exemplo, cada evento pertence a até 5 janelas diferentes).

Gatilhos de materialização

Desencadeia o controlo quando um pipeline de materialização é executado. O tipo de gatilho depende do tipo de característica.

CronSchedule

Utilização CronSchedule para funcionalidades de agregação (AggregationFunction). O pipeline funciona num cronograma fixo definido por uma expressão cron de quartzo.

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

TableTrigger Use para ColumnSelection funcionalidades suportadas por um DeltaTableSource. O pipeline corre sempre que a tabela Delta a montante recebe um novo commit.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

StreamingMode Use para funcionalidades suportadas por um StreamSource. O pipeline funciona como um pipeline contínuo de 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(),
)

Escolher um gatilho

Não podes materializar funcionalidades que exijam diferentes tipos de gatilho numa única materialize_features chamada. Em vez disso, emita chamadas separadas.

Migrar funcionalidades beta para a Pré-visualização Pública

O Feature Views Public Preview introduz entidades de Feature de primeira classe no Unity Catalog, regidas pelos CREATE FEATURE privilégios e READ FEATURE e requerem databricks-feature-engineering a versão 0.16.0 ou posterior. Funcionalidades criadas durante a beta (com a versão 0.15.0) são armazenadas como funções do Catálogo Unity e não suportam toda a funcionalidade de Pré-visualização Pública. Para obter suporte de Pré-visualização Pública a longo prazo, recrie as suas funcionalidades beta com a versão 0.16.0. As funcionalidades têm de ser eliminadas e recriadas, não apenas rematerializadas.

Para mais informações sobre funcionalidades, consulte Visualizações de Funcionalidades.

O que precisas de fazer

• Atualizar para a 0.16.0. Esta é a versão cliente obrigatória para funcionalidades de Pré-visualização Pública (batch e streaming).
• Recria as tuas funcionalidades. As Vistas de Funcionalidades Beta devem ser eliminadas e recriadas, não rematerializadas, porque não suportam toda a funcionalidade de Pré-visualização Pública.
• Migra antes que a janela feche. As funcionalidades beta existentes devem ser migradas antes de 22 de julho de 2026.

Identificar funcionalidades beta e Pré-visualização Pública

As funcionalidades de Pré-visualização Pública aparecem como um objeto de funcionalidades no Catálogo Unity, por exemplo, no Explorador de Catálogos. As funcionalidades beta aparecem como uma função com uma definição YAML. Qualquer funcionalidade representada como função é uma funcionalidade beta que precisas de migrar.

Funcionalidades de migração beta

Migrar uma funcionalidade beta tem três partes:

• Recrie a funcionalidade como uma funcionalidade de Pré-visualização Pública.
• Rematerializar a funcionalidade, para que as suas tabelas offline e online sejam reconstruídas sob a nova funcionalidade.
• Depois de verificares as funcionalidades migradas, apaga as funcionalidades beta e as suas materializações.

Recriar as funcionalidades

Use list_beta_feature_views para encontrar as suas funcionalidades beta, Feature.clone() criar uma cópia não registada e register_feature re-registar cada cópia como uma funcionalidade de Pré-visualização Pública. A clonagem limpa o registo, catálogo e esquema para que a funcionalidade possa ser re-registada.

Para evitar colisões de nomes, regista funcionalidades migradas com um nome diferente ou num esquema diferente das funcionalidades beta. O exemplo seguinte volta a registar cada característica no seu esquema original com um _migrated sufixo de nome.

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

Rematerializar as características migradas

Se uma funcionalidade beta fosse materializada, rematerializa-se a sua contraparte de Pré-visualização Pública para que as suas tabelas offline e online fossem reconstruídas sob a nova funcionalidade. Forneça as configurações offline e online da loja para a funcionalidade migrada e reconstrua o gatilho a partir da materialização existente da funcionalidade beta.

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

Eliminar as funcionalidades beta

Depois de verificar as funcionalidades migradas, apague as materializações de cada funcionalidade beta e depois a própria funcionalidade beta.

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)