Referência da API de Exibições de Recursos

Controle de acesso

Os recursos são objetos governáveis do Catálogo do Unity. O acesso a um recurso é controlado pelos privilégios do CREATE FEATUREREAD FEATURECatálogo do Unity e MANAGE do Unity. Para obter descrições completas, consulte a referência de privilégios do Catálogo do Unity.

• CREATE FEATURE — Necessário para criar um recurso em um esquema. create_feature e register_feature exija CREATE FEATURE no esquema pai. Seguindo o princípio de privilégio mínimo, conceda CREATE FEATURE no nível do esquema; você também pode concedê-lo em um catálogo para permitir a criação de recursos em qualquer esquema nesse catálogo.
• READ FEATURE — Necessário para ler um recurso e seus dados. get_feature, create_training_sete a leitura de dados de recursos materializados para treinamento ou serviço exigem READ FEATURE no recurso. READ FEATURE concedido em um esquema ou catálogo se aplica a todos os recursos atuais e futuros que ele contém.
• MANAGE — Necessário para gerenciar o ciclo de vida e as concessões de um recurso. Excluir um recurso com delete_feature, e materializar um recurso com materialize_features ou delete_materialized_feature, exigir MANAGE no recurso.

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

API do Modo de Exibição de Recurso

Feature construtor e register_feature()

A abordagem recomendada é construir um Feature objeto localmente e usá-lo register_feature para persistê-lo no Catálogo do Unity. Esse fluxo de trabalho de duas etapas permite que você experimente recursos (incluindo create_training_set) antes de registrá-los.

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() registra uma construção Feature local no Catálogo do Unity.

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 registra imediatamente um recurso no Catálogo do Unity em uma única etapa. Use isso quando você não precisar experimentar o recurso 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 na computação de recursos (DeltaTableSourceouRequestSourceStreamSource).
• function: um AggregationFunction que agrupa o operador (por exemplo, Sum(input="amount")), a coluna de entrada e a janela de tempo. Ou ColumnSelection("column_name") para recursos de passagem.
• catalog_name: o nome do catálogo do Catálogo do Unity para o recurso.
• schema_name: o nome do esquema do Catálogo do Unity para o recurso.
• entity: lista de nomes de coluna que definem o nível de agregação (chaves primárias). Necessário para recursos de agregação. Por exemplo, ["user_id"] agregações por usuário.
• timeseries_column: a coluna de carimbo de data/hora usada para agregação de janela de tempo. Necessário para recursos de agregação.
• name: nome do recurso opcional. Se omitido, gerado automaticamente da coluna de entrada, função e janela (por exemplo, amount_avg_rolling_7d).
• description: descrição opcional do recurso.

Retorna: Uma instância de funcionalidade validada

Gera: ValueError se alguma validação falhar

delete_feature()

Exclui um recurso do Catálogo do Unity por 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 excluir um recurso, remova ou atualize quaisquer modelos ou especificações de recursos que façam referência a ele. Se o recurso tiver sido materializado, exclua primeiro o recurso materializado. Veja como excluir um recurso materializado.

Nomes gerados automaticamente

Quando name é omitido, um nome é gerado automaticamente. 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 do carimbo de data/hora do evento)

Funções com suporte

Funções de agregação

ColumnSelection (passagem)

ColumnSelection seleciona uma única coluna de uma origem sem aplicar nenhuma agregação. Ele é encapsulado diretamente no function parâmetro (não dentro AggregationFunction). O tipo de retorno é inferido do esquema de origem.

ColumnSelection pode ser usado com qualquer fonte de dados:

• DeltaTableSource: retorna o valor mais recente por chave de entidade por meio de uma junção pontual (sem agregação de janela de lookback).
• StreamSource: retorna o valor mais recente por chave de entidade do fluxo (sem agregação de janela de lookback).
• RequestSource: passa pelo valor fornecido em tempo de inferência (ou extraído do DataFrame rotulado no tempo de treinamento).

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: recursos de agregação e seleção de coluna

O exemplo a seguir mostra os recursos definidos na 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",
)

Funções com condições de filtro

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

Os filtros são úteis ao trabalhar com tabelas de origem grandes que incluem um superconjunto de dados necessários para a computação de recursos e minimizar a necessidade de criar exibições 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 os recursos são computados de uma tabela de origem. Ele não cria uma nova tabela. Ele especifica a configuração para ler dados e agregar recursos.

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 Catálogo do Unity.
• 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 isso para renomear colunas, tipos de conversão ou colunas derivadas de computação antes da agregação. Se omitido, todas as colunas serã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 do Spark StructType (de df.schema.json()). Necessário se transformation_sql for fornecido. Isso informa ao sistema os nomes e tipos de coluna resultantes da transformação.

Quando ambos filter_conditiontransformation_sql e sã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: Derivando transformation_sql e dataframe_schema de um DataFrame do PySpark

Você pode escrever sua transformação como uma consulta PySpark e, em seguida, 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()

Como uma conveniência, você pode criar uma DeltaTableSource consulta SQL. O método analisa a consulta para extrair automaticamente o nome transformation_sqlda tabela e filter_condition.

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

Há suporte apenas para consultas simples SELECT ... FROM ... [WHERE ...] . O SQL complexo (JOINs, subconsultas, CTEs, UNIONs) é rejeitado. Para consultas complexas, construa 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 visualizar os dados que serão usados para computação de recursos. Isso é útil para iterar e filter_conditiontransformation_sql até que eles 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()

Noções básicas sobre entidades

As colunas de entidade definem o nível de agregação para seus recursos. Eles são especificados na Feature definição, não em DeltaTableSource. As entidades determinam:

• Como os dados são agrupados: os recursos são agregados por combinação exclusiva de valores de entidade (semelhante ao GROUP BY SQL)
• A estrutura de chave primária: cada combinação de entidade exclusiva resulta em uma linha de recursos computados

Exemplo: recursos no nível do cliente

O código a seguir agrega recursos no 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: recursos no nível da Loja do Cliente

Para agregar recursos em um nível mais detalhado (uma linha por combinação de repositório de clientes), use várias 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 você precisar de recursos em diferentes níveis de agregação (por exemplo, no nível do cliente e no nível do repositório do cliente), use valores diferentes entity em suas definições de recurso. O mesmo DeltaTableSource pode ser compartilhado entre recursos com diferentes configurações de entidade.

StreamSource

StreamSource faz referência a um Stream. O Stream contém conexão, autenticação, esquema e configuração de ingestão para a fonte de streaming. Para o Kafka, as referências de coluna em definições de recurso devem ser prefixadas ou value.key. indicar qual parte da mensagem deve ser lida.

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 de 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 de coluna prefixada por ponto (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 fornecidos em tempo de inferência no conteúdo da solicitação em vez de pesquisados de uma tabela pré-materializada. Durante o treinamento, essas colunas são extraídas do DataFrame rotulado passado para create_training_set. Durante o serviço de modelo, o chamador deve incluí-los no conteúdo da solicitação HTTP.

RequestSource é usado com ColumnSelection (para passar um valor diretamente). Ele não dá suporte a funções de agregação ou janelas de tempo.

Definindo o 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 com suporte

RequestSourcedá suporte aos tipos escalares definidos emScalarDataType: INTEGER, , FLOAT, BOOLEAN, STRING, DOUBLE, , LONG, TIMESTAMP, , DATE, . SHORT Não há suporte para tipos complexos, como matrizes, mapas e structs.

Como os dados de solicitação são hidratados

Assinatura de modelo

Quando um modelo é registrado usando log_model um conjunto de treinamento que inclui RequestSource recursos, as RequestSource colunas são adicionadas à assinatura do modelo do MLflow como entradas necessárias. Isso significa que o esquema de API do ponto de extremidade de serviço reflete quais campos os chamadores devem fornecer em tempo de inferência.

API de treinamento e inferência

create_training_set()

Cria um conjunto de dados de treinamento com computação de recurso correta pontual. Para obter detalhes, consulte Treinar modelos com exibições de recursos.

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

Registra um modelo com metadados de recurso para acompanhamento de linhagem e pesquisa automática de recursos durante a inferência. Para obter detalhes, consulte Treinar modelos com exibições de recursos.

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

Executa a inferência do lote offline com pesquisa automática de recursos. Usa os metadados de recurso armazenados com o modelo para computar recursos corretos pontuais, garantindo consistência com o treinamento.

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 de entidade e de timeseries usadas durante o treinamento. Os recursos são calculados 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 de horas

Os Modos de Exibição de Recursos dão suporte a três tipos de janela diferentes para controlar o comportamento de pesquisa para agregações baseadas em janela de tempo: rolagem, cascata e deslizamento.

• Janelas sem interrupção olham para trás a partir da hora do evento. Duração e atraso são definidos explicitamente.
• Janelas deslizantes são janelas de tempo fixas e não sobrepostas. Cada ponto de dados pertence a exatamente uma janela.
• Janelas deslizantes são janelas de tempo sobrepostas e contínuas, com um intervalo de deslizamento configurável.

A ilustração a seguir mostra como eles funcionam.

Janela sem interrupção

Janelas sem interrupção são up-toagregações de data e tempo real, normalmente usadas em dados de streaming. Em pipelines de streaming, a janela sem interrupção emite uma nova linha somente quando o conteúdo da janela de comprimento fixo é alterado, como quando um evento entra ou sai. Quando um recurso de janela sem interrupção é usado em pipelines de treinamento, um cálculo preciso de recurso pontual é executado nos dados de origem usando a duração da janela de comprimento fixo imediatamente anterior ao carimbo de data/hora de um evento específico. Isso ajuda a evitar distorção ou vazamento de dados online offline. Recursos no tempo T agregam eventos de [T – duração, T).

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

A tabela a seguir lista os parâmetros de uma janela sem interrupção. Os horários de início e término da janela são baseados nesses parâmetros da seguinte maneira:

• Hora de início: evaluation_time - window_duration - delay (inclusive)
• 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 sem interrupção 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 janela sem interrupção

• window_duration=timedelta(days=7): cria-se uma janela de retrospetiva de 7 dias, que termina no momento da avaliação atual. Para um evento agendado para as 14:00 do dia 7, isso inclui todos os eventos que ocorreram 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): Isso estabelece uma janela de retrospectiva de 1 hora que termina 30 minutos antes do tempo de avaliação. Para um evento às 15h, isso inclui todos os eventos das 13h30 até (mas não incluindo) 14h30. Isso é útil para considerar os atrasos na ingestão de dados.

Janela Deslizante

Para recursos definidos usando janelas de salto, as agregações são computadas em uma janela de comprimento fixo predeterminada que avança por um intervalo de deslizamento, produzindo janelas não sobrepostas que fazem o particionamento completo do tempo. Como resultado, cada evento na origem contribui para exatamente uma janela. Os recursos no momento t agregam dados de janelas que terminam em ou antes t (exclusivo). O Windows começa na época do Unix.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

A tabela a seguir lista os parâmetros de uma janela em cascata.

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

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

Exemplo de janela em cascata

• window_duration=timedelta(days=5): isso cria janelas de comprimento fixo predeterminadas de 5 dias cada. Exemplo: a janela nº 1 abrange o dia 0 até o dia 4, a janela nº 2 abrange o dia 5 ao dia 9, a janela nº 3 abrange o dia 10 até o dia 14 e assim por diante. Especificamente, a Janela nº 1 inclui todos os eventos com carimbos de data/hora que começam no 00:00:00.00 do dia 0 até (mas não incluindo) qualquer evento com carimbo de data/hora 00:00:00.00 no dia 5. Cada evento pertence a exatamente uma janela.

Janela Deslizante

Para recursos definidos usando janelas deslizantes, as agregações são computadas em uma janela de comprimento fixo predeterminada que avança por um intervalo de slides, produzindo janelas sobrepostas. Cada evento na origem pode contribuir para a agregação de recursos para várias janelas. Os recursos no momento t agregam dados de janelas que terminam em ou antes t (exclusivo). O Windows começa na época do Unix.

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

A tabela a seguir lista os parâmetros de 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 de janela deslizante

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): isso cria janelas de cinco dias sobrepostas que avançam 1 dia cada vez. Exemplo: a janela nº 1 abrange o dia 0 ao dia 4, a janela nº 2 abrange o dia 1 ao 5º dia, a janela nº 3 abrange o dia 2 até o dia 6 e assim por diante. Cada janela inclui eventos do 00:00:00.00 dia inicial até (mas sem incluir) 00:00:00.00 no dia final. Como as janelas se sobrepõem, um único evento pode pertencer a várias janelas (neste exemplo, cada evento pertence a até cinco janelas diferentes).

Gatilhos de materialização

Dispara o controle quando um pipeline de materialização é executado. O tipo de gatilho depende do tipo de recurso.

CronSchedule

Use CronSchedule para recursos de agregação (AggregationFunction). O pipeline é executado em um agendamento fixo definido por uma expressão cron de Quartz.

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Uso TableTrigger para ColumnSelection recursos apoiados por um DeltaTableSource. O pipeline é executado sempre que a tabela Delta upstream recebe uma nova confirmação.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

Uso StreamingMode para recursos apoiados por um StreamSource. O pipeline é executado como um pipeline de streaming contínuo.

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

Escolhendo um gatilho

Não é possível materializar recursos que exigem tipos de gatilho diferentes em uma única materialize_features chamada. Em vez disso, emita chamadas separadas.

Migrar recursos beta para a Visualização Pública

A Visualização Pública de Exibições de Recursos apresenta entidades de recursos de primeira classe no Catálogo do Unity, regidas por privilégios e READ FEATURE que exigem databricks-feature-engineering a CREATE FEATURE versão 0.16.0 ou posterior. Os recursos criados durante a versão beta (com a versão 0.15.0) são armazenados como funções do Catálogo do Unity e não dão suporte a todas as funcionalidades de Visualização Pública. Para obter suporte de visualização pública de longo prazo, recrie seus recursos beta com a versão 0.16.0. Os recursos devem ser excluídos e recriados, não apenas materializados novamente.

Para obter mais informações sobre recursos, consulte Exibições de Recursos.

O que é necessário fazer

• Atualize para 0.16.0. Esta é a versão do cliente necessária para recursos de Visualização Pública (lote e streaming).
• Recrie seus recursos. As Exibições de Recursos Beta devem ser excluídas e recriadas, não materializadas novamente, pois não dão suporte a todas as funcionalidades de Visualização Pública.
• Migre antes do fechamento da janela. Os recursos beta existentes devem ser migrados antes de 22 de julho de 2026.

Identificar recursos beta e de visualização pública

Os recursos de Visualização Pública aparecem como um objeto de recurso no Catálogo do Unity, por exemplo, no Catalog Explorer. Os recursos beta aparecem como uma função com uma definição yaml. Qualquer recurso representado como uma função é um recurso beta que você precisa migrar.

Migrar recursos beta

A migração de um recurso beta tem três partes:

• Recrie o recurso como um recurso de Visualização Pública.
• Materialize novamente o recurso, de modo que suas tabelas offline e online sejam recriadas sob o novo recurso.
• Depois de verificar os recursos migrados, exclua os recursos beta e suas materializações.

Recriar os recursos

Use list_beta_feature_views para localizar seus recursos beta, Feature.clone() criar uma cópia não registrada e register_feature registrar novamente cada cópia como um recurso de Visualização Pública. A clonagem limpa o registro, o catálogo e o esquema para que o recurso possa ser registrado novamente.

Para evitar colisões de nome, registre recursos migrados com um nome diferente ou em um esquema diferente dos recursos beta. O exemplo a seguir registra novamente cada recurso em 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))

Materializar novamente os recursos migrados

Se um recurso beta foi materializado, materialize novamente seu equivalente de Visualização Pública para que suas tabelas offline e online sejam recriadas sob o novo recurso. Forneça as configurações de repositório offline e online para o recurso migrado e reconstrua o gatilho da materialização existente do recurso 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,
    )

Excluir os recursos beta

Depois de verificar os recursos migrados, exclua as materializações de cada recurso beta e, em seguida, o próprio recurso 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)