Vistas de Destaque

As Vistas de Características permitem-lhe definir e calcular características a partir de fontes de dados. As características podem ser definidas utilizando uma variedade de fontes (tabela Delta, fluxo do Kafka e dados no momento da solicitação) e cálculos (agregações com janelas temporais, seleções simples de colunas, entre outras). Este guia cobre os seguintes fluxos de trabalho:

• Fluxo de trabalho de desenvolvimento de funcionalidades
  • Use create_feature para definir objetos de funcionalidades do Unity Catalog que podem ser utilizados no treino e na disponibilização de modelos em fluxos de trabalho.
  • Alternativamente, constrói Feature objetos localmente e usa-os register_feature para os persistir no Unity Catalog mais tarde. Elementos construídos localmente podem ser usados com create_training_set antes do registo.
• Fluxo de trabalho de treino de modelos
  • Use create_training_set para calcular características agregadas em ponto no tempo para aprendizagem automática. Para obter documentação detalhada sobre como treinar modelos com o Feature Views, consulte Treinar modelos com o Feature Views.
• Materialização e disponibilização de funcionalidades no fluxo de trabalho
  • Depois de definir uma funcionalidade com create_feature ou recuperá-la usando get_feature, pode-se usar materialize_features para materializar a funcionalidade ou conjunto de funcionalidades numa loja offline para reutilização eficiente, ou numa loja online para serviço online.
  • Use create_training_set com a visualização materializada para preparar um conjunto de dados de treino em lote offline.

Para detalhes da API, consulte a referência da API Feature Views.

Requerimentos

• Computação sem servidor ou um cluster de computação clássico com o Databricks Runtime 17.0 ML ou superior.

• Tens de instalar o pacote Python personalizado. Execute as seguintes linhas de código cada vez que executar um caderno:

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

Exemplo de início rápido

Para um caderno de início rápido executável, veja o Exemplo de caderno.

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

Exemplo de bloco de notas

Caderno de início rápido de Visualizações de Funcionalidades

Obter caderno

Funcionalidades de streaming

Para além das funcionalidades em lote das tabelas Delta, pode definir funcionalidades a partir de fontes de streaming para casos de uso em tempo real. As funcionalidades de streaming usam a mesma classe Feature que as funcionalidades em lote — mesmos Feature construtores, mesmas funções de agregação, mesmos fluxos de trabalho de treino e de serviço — pelo que a atualização de batch para tempo real requer alterações mínimas no código. Uma vez materializadas, as funcionalidades de streaming entregam uma frescura de ponta a ponta abaixo do segundo (latência p99 de 200ms) diretamente para o seu modelo que serve os endpoints.

Para usar funcionalidades de transmissão, primeiro configurar um Stream e depois fazer referência ao mesmo utilizando um StreamSource. As origens de fluxo suportam o Kafka como entrada e mantêm automaticamente uma tabela de ingestão (Delta) como cópia histórica dos dados para fins de treino.

Defina uma funcionalidade de streaming

A StreamSource refere-se a um Stream pelo seu nome em três partes (catalog.schema.stream_name). Um Stream não é um objeto protegível do Unity Catalog, mas está associado a um esquema do Unity Catalog, e o acesso é controlado pela tabela de ingestão do Stream. As referências às colunas nas definições de entidade, série temporal e função devem ser precedidas por value. ou key. para indicar qual parte da mensagem de Kafka ler. Os campos aninhados são suportados através da notação por pontos (por exemplo, value.user.address.city).

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 do filtro no StreamSource

Use filter_condition para filtrar linhas do fluxo antes da agregação, tal como em DeltaTableSource.

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

Seleção de colunas a partir de fluxos

ColumnSelection funcionalidades funcionam com fontes de streaming. A coluna selecionada representa o valor mais recente da Stream para cada entidade, mantendo a exatidão temporal.

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

Aceder a campos aninhados

Pode aceder a campos JSON aninhados usando notação de pontos (por exemplo, value.nested_field.amount). No momento da execução, a carga útil do pedido e a resposta utilizam nomes de nós de folha (por exemplo, amount em vez de value.amount). Os nomes dos nós folha devem ser únicos entre todas as colunas de saída de entidades, séries temporais e atributos num modelo ou numa Especificação de Atributos, porque o ponto de extremidade de disponibilização usa os nomes dos nós folha para encaminhar os valores.

Janelas temporais para funcionalidades de streaming

As funcionalidades de streaming suportam apenas RollingWindow para agregações. As janelas deslizantes recalculam continuamente com base nos dados mais recentes, o que está em consonância com a natureza em tempo real das fontes de dados em fluxo. TumblingWindow e SlidingWindow são concebidos para cálculo em lotes em intervalos históricos fixos.

Exemplo de caderno de funcionalidades de streaming

Notebook de introdução rápida do Streaming Feature Views

Obter caderno

Treino de modelos e inferência

Para treinar modelos e executar inferência em lote com Feature Views, incluindo log_model(), score_batch(), e create_training_set(), veja Train models with Feature Views.

Materialização de funcionalidades

Depois de definir funcionalidades, pode materializá-las em lojas offline ou online para uma reutilização eficiente em workflows de formação e serviço. Depois de materializar as funcionalidades, pode servir modelos usando o serviço de modelos da CPU. Para mais detalhes, veja Materialize Feature Views.

Melhores práticas

Nomenclatura das funcionalidades

• Use nomes descritivos para características críticas para o negócio.
• Siga convenções de nomes consistentes entre as equipas.
• Use nomes gerados automaticamente ao começar a desenvolver funcionalidades.

Janelas temporais

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

Performance

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

Colunas de entidade vs. condições de filtro

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

Use entity (em create_feature) quando precisar de diferentes níveis de agregação:

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

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

• Apenas transações de alto valor: filter_condition="amount > 100" (ainda agregadas por cliente)
• Apenas encomendas concluídas: filter_condition="status = 'completed'" (continuam a ser agregadas por cliente)

Regra geral: Se a sua alteração resultar num número diferente de linhas por valor de entidade, use valores diferentes entity nas definições de características. Se estás apenas a filtrar quais as linhas que contribuem para a mesma agregação, usa filter_condition na fonte.

Padrões comuns

Análise de clientes

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

# 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

# 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 entidades e séries temporais devem estar de acordo entre o conjunto de dados de treino (rotulado) e as definições de funcionalidades quando usados na create_training_set API.
• O nome da coluna usado como label coluna no conjunto de dados de treino não deve existir nas tabelas de origem usadas para definir Features.
• Uma lista limitada de funções (UDAFs) é suportada na create_feature API. Consulte Funções suportadas.
• Colunas de entidade não podem ser do tipo DATE ou TIMESTAMP.
• RequestSourcesuporta apenas tipos de dados escalares definidos em ScalarDataType (INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMPDATE). SHORT Tipos complexos como arrays, mapas e structs não são suportados.
• RequestSource não suporta funções de agregação nem janelas temporais. Só 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 funcionalidades de pedido deve ser único globalmente em todas as fontes dentro de um conjunto de treino ou endpoint de servição.

• score_batch pode não ter sucesso em computação serverless. Contorne este problema usando um cluster de computação clássico com Databricks Runtime 17.0 ML ou superior.

Para limitações específicas da materialização, veja Limitações.