Pular para o conteúdo principal
Última atualização em

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

  • **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 use CronSchedule 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​

Além dos recursos em lote de tabelas Delta, é possível definir recursos de fontes de transmissão para casos de uso em tempo real. Recursos de transmissão usam a mesma classe Feature que os recursos em lote — mesmos construtores Feature, mesmas funções de agregação, mesmos fluxos de trabalho de treinamento e serviço — então a atualização de lote para tempo real requer mudanças mínimas no código. Uma vez materializados, os recursos de transmissão entregam frescor de ponta a ponta em subssegundos (latência p99 de 200ms) diretamente para seus endpoints de servindo modelo.

Para usar recursos de transmissão, primeiro configure uma transmissão, depois a referencie usando um StreamSource. As fontes de transmissão oferecem suporte ao Kafka como entrada e mantêm automaticamente uma tabela de ingestão (Delta) como uma cópia histórica dos dados para treinamento.

Definir um recurso de transmissão​

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

Condições de filtro em StreamSource​

Use filter_condition para filtrar linhas da transmissão antes da agregação, assim como em DeltaTableSource.

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

Seleção de coluna de transmissões​

Recursos ColumnSelection funcionam com fontes de transmissão. A coluna selecionada representa o valor mais recente da Transmissão para cada entidade, respeitando a precisão pontual.

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

Acessar campos aninhados​

Você pode acessar campos JSON aninhados usando notção de ponto (por exemplo, value.nested_field.amount). No momento da exibição, a carga útil da solicitação e a resposta usam nomes de nó folha (por exemplo, amount em vez de value.amount). Os nomes de nós folha devem ser únicos em todas as colunas de saída de entidade, séries temporais e recursos dentro de um modelo ou Especificação de Recurso, porque o Endpoint de disponibilização usa os nomes dos nós 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.

Nesta página