view de Recursos

info

Visualização

Este recurso está em Prévia Pública. Administradores do Workspace podem controlar o acesso a este recurso na página **Pré-visualizações**. Consulte Gerenciar prévias do Databricks.

Feature Views permitem definir e calcular recursos de fontes de dados. Os recursos podem ser definidos usando uma variedade de fontes (tabela Delta, transmissão Kafka e dados de tempo de solicitação) e computações (agregações com janelas de tempo, seleções simples de colunas e muito mais). Este guia abrange os seguintes fluxos de trabalho:

• **Fluxo de trabalho de desenvolvimento de recursos**

  • Use create_feature para definir objetos de recurso do Unity Catalog que podem ser usados em fluxos de trabalho de treinamento e disponibilização de modelos.
  • Alternativamente, construa objetos Feature localmente e use register_feature para persistí-los no Unity Catalog mais tarde. Os recursos construídos localmente podem ser usados com create_training_set antes do registro.

• **Fluxo de trabalho de treinamento de modelo**

  • Use create_training_set para calcular recursos agregados pontuais para Machine Learning. Para documentação detalhada sobre treinamento com Recursos de visualização, consulte Treinar modelos com Recursos de visualização.

• **Fluxo de trabalho de materialização e disponibilização de recursos**

  • Depois de definir um recurso com create_feature ou recuperá-lo usando get_feature, você pode usar materialize_features para materializar o recurso ou conjunto de recursos em um armazenamento offline para reutilização eficiente, ou em um armazenamento online para servir online.
  • Use create_training_set com a view materializada para preparar um dataset de treinamento em lotes offline.

Para detalhes da API, consulte a referência da API de recurso views.

Requisitos

• Compute Serverless ou um cluster compute clássico executando Databricks Runtime 17.0 ML ou acima.

• É preciso instalar o pacote Python personalizado. Execute as seguintes linhas de código sempre que executar um notebook:

  Python

  %pip install databricks-feature-engineering>=0.16.0
  dbutils.library.restartPython()

Exemplo de Início rápido

Para um notebook de início rápido executável, consulte Exemplo de notebook.

Python

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    CronSchedule, DeltaTableSource, Feature, AggregationFunction,
    Sum, Avg, ColumnSelection, TableTrigger,
    TumblingWindow, SlidingWindow,
    OfflineStoreConfig, OnlineStoreConfig,
)
from datetime import timedelta

CATALOG_NAME = "main"
SCHEMA_NAME = "feature_store"
TABLE_NAME = "transactions"

# 1. Create data source
source = DeltaTableSource(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name=TABLE_NAME,
)

# 2. Define features locally (no catalog/schema needed yet)
avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), TumblingWindow(window_duration=timedelta(days=30))),
    name="avg_transaction_30d",
)

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), SlidingWindow(window_duration=timedelta(days=7), slide_duration=timedelta(days=1))),
    # name auto-generated: "amount_sum_sliding_7d_1d"
)

fe = FeatureEngineeringClient()

# 3. Explore features with compute_features
feature_df = fe.compute_features(features=[avg_feature, sum_feature])
feature_df.display()

# 4. Create training set using local features
# `labeled_df` should have columns "user_id", "transaction_time", and "target".
training_set = fe.create_training_set(
    df=labeled_df,
    features=[avg_feature, sum_feature],
    label="target",
)
training_set.load_df().display()

# 5. Register features in Unity Catalog
avg_feature = fe.register_feature(
    feature=avg_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)
sum_feature = fe.register_feature(
    feature=sum_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)

# 6. Or use create_feature for a one-step define-and-register workflow
latest_amount = fe.create_feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    name="latest_amount",
)

# 7. Train model
with mlflow.start_run():
    training_df = training_set.load_df()

    # training code

    fe.log_model(
        model=model,
        artifact_path="recommendation_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name=f"{CATALOG_NAME}.{SCHEMA_NAME}.recommendation_model",
    )

# 8. (Optional) Materialize features for serving
# Features must be registered in UC before calling materialize_features
online_config = OnlineStoreConfig(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name_prefix="customer_features_serving",
    online_store_name="customer_features_store",
)

# Aggregation features support CronSchedule or TableTrigger, and support both offline and online configs
fe.materialize_features(
    features=[avg_feature, sum_feature],
    offline_config=OfflineStoreConfig(
        catalog_name=CATALOG_NAME,
        schema_name=SCHEMA_NAME,
        table_name_prefix="customer_features",
    ),
    online_config=online_config,
    trigger=CronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # Hourly
        timezone_id="UTC",
    ),
)

# ColumnSelection features use TableTrigger and only support online config
fe.materialize_features(
    features=[latest_amount],
    online_config=online_config,
    trigger=TableTrigger(),
)

Notebook de exemplo

Notebook de início rápido de Recurso Views

Abrir notebook em uma nova abaOpen in Databricks

Recursos de transmissão

Use recursos de transmissão quando os valores dos recursos precisarem ser atualizados continuamente em vez de programar em lotes. Recursos de transmissão e de lotes usam os mesmos construtores Feature, funções de agregação e fluxos de trabalho de treinamento e serviço.

Os recursos de transmissão não são materializados em um armazenamento offline. Para treinamento e inferência em lote, o Databricks calcula os valores dos recursos a partir da fonte.

Os recursos de transmissão têm os seguintes requisitos:

• Você deve fornecer um online_config. Os recursos de transmissão não oferecem suporte a offline_config.
• Você não pode combinar recursos de transmissão e de lotes em uma única chamada materialize_features. Faça uma chamada separada para cada tipo de Trigger.
• transformation_sql não é compatível com recursos de transmissão.
• A materialização de transmissão processa apenas os registros que chegam após o pipeline começar e não realiza o preenchimento de registros históricos. Agregados de janela deslizante retornam resultados completos apenas após a chegada da primeira janela completa de dados.

Escolhendo uma fonte de recurso de transmissão

Escolha uma fonte com base em seus requisitos de atualização e na configuração de ingestão existente:

• Use um StreamSource quando a atualização abaixo de um segundo for a prioridade. Os recursos StreamSource oferecem uma latência de ponta a ponta p99 de 200 milissegundos. Primeiro configure uma transmissão e, em seguida, faça referência a ela usando um StreamSource. As fontes de transmissão oferecem suporte ao Kafka como entrada e mantêm automaticamente uma tabela Delta de ingestão como uma cópia histórica dos dados para treinamento.
• Use um DeltaTableSource quando você já tiver um caminho de ingestão de baixa latência para uma tabela Delta. Espere uma atualização na ordem de dezenas de segundos.
• Use o Zerobus para popular um DeltaTableSource quando você ainda não tiver um caminho de ingestão de baixa latência. A ingestão do Zerobus leva na ordem de dezenas de segundos, portanto, espere uma atualização de recurso inferior a um minuto.

Defina um recurso de transmissão usando um StreamSource

Uma StreamSource faz referência a uma Transmissão por seu nome de três partes (catalog.schema.stream_name). Uma Transmissão não é um objeto protegível do Unity Catalog, mas tem seu escopo definido para um esquema do Unity Catalog e o acesso é regido pela tabela de ingestão da Transmissão. Referências de coluna em definições de entidade, série temporal e função devem ser prefixadas com value. ou key. para indicar qual parte da mensagem do Kafka ler. Campos aninhados têm suporte usando a notação de ponto (por exemplo, value.user.address.city).

Python

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    StreamSource,
    Feature,
    AggregationFunction,
    Sum,
    RollingWindow,
)
from datetime import timedelta

client = FeatureEngineeringClient()

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

feature = Feature(
    name="user_purchase_sum",
    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)),
    ),
)

Defina um recurso de transmissão usando uma DeltaTableSource

Para materializar um recurso definido em um DeltaTableSource como um recurso de transmissão, passe StreamingMode como o trigger para materialize_features. A definição do recurso usa as mesmas APIs que um recurso em lote suportado por um DeltaTableSource. Fontes de tabela Delta oferecem suporte a recursos de agregação e seleção de colunas.

A tabela Delta de origem deve ter o change data feed (CDF) habilitado definindo delta.enableChangeDataFeed=true.

O exemplo a seguir define e materializa um recurso de agregação com uma fonte de tabela Delta.

Python

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    DeltaTableSource,
    AggregationFunction,
    OnlineStoreConfig,
    Sum,
    RollingWindow,
    StreamingMode,
)
from datetime import timedelta

client = FeatureEngineeringClient()

source = DeltaTableSource(
    catalog_name="my_catalog",
    schema_name="my_schema",
    table_name="transactions",
)

feature = client.create_feature(
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(
        operator=Sum(input="amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
)

online_config = OnlineStoreConfig(
    catalog_name="my_catalog",
    schema_name="my_schema",
    table_name_prefix="streaming_features",
    online_store_name="my_online_store",
)

client.materialize_features(
    features=[feature],
    online_config=online_config,
    trigger=StreamingMode(),
)

Use uma tabela Delta populada pelo Zerobus

Uma tabela Delta populada pelo Zerobus pode servir como uma fonte de recurso de transmissão. O Zerobus não define delta.enableChangeDataFeed=true automaticamente. Você deve definir esta propriedade manualmente na tabela Delta de destino antes de usá-la como uma fonte de recurso de transmissão.

Condições de filtro em fontes de transmissão

Use filter_condition para filtrar linhas antes da agregação para um StreamSource ou DeltaTableSource.

Python

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

Seleção de coluna de fontes de transmissão

Os ColumnSelection recursos transmitem o valor mais recente da fonte para cada entity key sem agregação. No momento do treinamento, os valores dos recursos respeitam a precisão pontual.

Os recursos de seleção de coluna não possuem um TTL. Para remover um valor selecionado do armazenamento online, a fonte deve emitir um valor nulo para a coluna selecionada.

Python

from databricks.feature_engineering.entities import ColumnSelection

passenger_count = Feature(
    name="passenger_count",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=ColumnSelection(column="value.passenger_count"),
)

Acesse campos aninhados de um StreamSource

Para um StreamSource, você pode acessar campos JSON aninhados usando notação de ponto (por exemplo, value.nested_field.amount). No momento da disponibilização, a carga útil da solicitação e a resposta usam nomes de nós folha (por exemplo, amount em vez de value.amount). Os nomes dos nós folha devem ser exclusivos em todos os nomes de entidade, série temporal e recurso dentro de um modelo ou Especificação de Recurso, porque o endpoint de disponibilização usa nomes folha para rotear valores.

Janelas de tempo para recursos de transmissão

Recursos de transmissão suportam apenas RollingWindow para agregações. Janelas deslizantes são continuamente recalculadas sobre os dados mais recentes, o que se alinha com a natureza em tempo real das fontes de transmissão. TumblingWindow e SlidingWindow são projetados para computação em lotes em intervalos históricos fixos.

Exemplo de notebook de recursos de transmissão

Notebook de início rápido de Transmissão recurso views

Abrir notebook em uma nova abaOpen in Databricks

Treinamento e inferência de modelo

Para treinar modelos e executar inferência em lote com Feature Views, incluindo log_model(), score_batch() e create_training_set(), consulte Treinar modelos com Feature Views.

Materialização de recursos

Depois de definir os recursos, você pode materializá-los em armazenamentos offline ou online para reutilização eficiente em fluxos de trabalho de treinamento e disponibilização. Depois de materializar recursos, você pode servir modelos usando servindo modelo de CPU. Para detalhes, consulte Materializar recursos views.

Práticas recomendadas

Nomenclatura de recursos

• Use nomes descritivos para recursos críticos para os negócios.
• Adote convenções de nomenclatura consistentes entre as equipes.
• Use nomes gerados automaticamente ao começar a desenvolver recursos.

Janelas de tempo

• Alinhe os limites da janela com os ciclos de negócios (diários, semanais).
• Janelas mais curtas capturam tendências recentes, mas podem ser ruidosas. Janelas mais longas produzem distribuições de recursos mais estáveis, mas podem perder mudanças comportamentais recentes. Escolha com base na rapidez com que o sinal subjacente muda para seu caso de uso. Por exemplo, uma janela de 7 dias suaviza as flutuações diárias e produz entradas de modelo consistentes, enquanto uma janela de 1 hora reage rapidamente às mudanças comportamentais, mas pode introduzir variação que degrada o desempenho do modelo. Se a precisão do seu modelo se degradar quando a distribuição mudar, use uma janela mais longa para estabilizar as entradas.
• Janelas em cascata e deslizantes são mais escaláveis do que janelas contínuas (rolling). Comece com janelas deslizantes para a maioria dos casos de uso.

desempenho

• Materialize recursos da mesma fonte de dados em uma única chamada materialize_features para minimizar as varreduras de dados.
• Use a mesma granularidade (por exemplo, todas as durações de slides de 1 hora ou todas as de 1 dia) para os recursos na mesma fonte de dados, para permitir um melhor agrupamento durante a materialização.

Colunas de entidade vs. condições de filtro

Use este guia de decisão ao trabalhar com recursos da mesma tabela de origem:

Use entity (em create_feature) quando forem necessários diferentes níveis de agregação:

• **Recursos em nível de cliente** (uma linha por cliente): entity=["customer_id"]
• Recursos de cliente-comerciante (várias linhas por cliente): entity=["customer_id", "merchant_id"]
• Diferentes níveis de agregação podem compartilhar o mesmo DeltaTableSource : especifique diferentes valores entity em cada definição de recurso

Use filter_condition (em DeltaTableSource) quando precisar filtrar linhas no mesmo nível de agregação:

• **Somente transações de alto valor**: filter_condition="amount > 100" (ainda agregado por cliente)
• Somente pedidos concluídos : filter_condition="status = 'completed'" (ainda agregados por cliente)

Regra geral: Se sua alteração resultar em um número diferente de linhas por valor de entidade, use valores entity diferentes em suas definições de recursos. Se você estiver apenas filtrando quais linhas contribuem para a mesma agregação, use filter_condition na fonte.

Padrões comuns

Analítica do cliente

Python

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow

fe = FeatureEngineeringClient()
features = [
    # Recency: Number of transactions in the last day
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=1)))),

    # Frequency: transaction count over the last 90 days
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=90)))),

    # Monetary: total spend in the last month
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30)))),
]

Análise de tendências

Python

# Compare recent vs. historical behavior
fe = FeatureEngineeringClient()
recent_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

historical_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7), delay=timedelta(days=7))),
)

Padrões sazonais

Python

# Same day of week, 4 weeks ago
fe = FeatureEngineeringClient()
weekly_pattern = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=1), delay=timedelta(weeks=4))),
)

Limitações

• Os nomes das colunas de entidade e séries temporais devem corresponder entre o dataset de treinamento (rotulado) e as definições de recursos quando usados na API create_training_set.
• O nome da coluna usada como coluna label no dataset de treinamento não deve existir nas tabelas de origem usadas para definir Features.
• Uma lista limitada de funções (UDAFs) é compatível na API create_feature. Consulte funções compatíveis.
• Colunas de entidade não podem ser do tipo DATE ou TIMESTAMP.
• RequestSource suporta apenas tipos de dados escalares definidos em ScalarDataType (INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, DATE, SHORT). Tipos complexos, como arrays, mapas e structs, não são compatíveis.
• RequestSource não oferece suporte a funções de agregação ou janelas de tempo. Apenas as funções ColumnSelection podem ser usadas.
• O conjunto de nomes de colunas de entidade, nomes de colunas de séries temporais e nomes de colunas de recurso de solicitação deve ser globalmente único em todas as origens em um conjunto de treinamento ou endpoint de disponibilização.

Para limitações específicas de materialização, consulte Limitações.