Referência da API de 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: Obrigatório para ler metadados de funcionalidades. get_feature, create_training_set, e list_materialized_features requerem READ FEATURE sobre a característica. Este privilégio não concede acesso a dados de funcionalidades em tabelas de origem ou de saída materializadas. Para ler esses dados para treino ou serviço, também deve estar SELECT nas tabelas aplicáveis. 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 de uma funcionalidade e os subsídios. Eliminar uma funcionalidade com delete_feature, e materializar uma característica com materialize_features, requerem MANAGE sobre a funcionalidade. Eliminar uma funcionalidade materializada com delete_materialized_feature não é regida por MANAGE: apenas o criador da funcionalidade materializada pode apagá-la.

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, RequestSource, or FeatureViewSource
    function: Union[AggregationFunction, ColumnSelection, CustomUDF], # Required for all features
    entity: Optional[List[str]] = None,                    # Required for DeltaTableSource and StreamSource
    timeseries_column: Optional[str] = None,               # Required for DeltaTableSource and StreamSource
    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, RequestSource, or FeatureViewSource
    function: Union[AggregationFunction, ColumnSelection, CustomUDF], # Required for all features
    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 DeltaTableSource and StreamSource
    timeseries_column: Optional[str] = None,               # Required for DeltaTableSource and StreamSource
    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, RequestSource, ou FeatureViewSource).
• function: E AggregationFunction que agrupa um operador e uma janela temporal, ColumnSelection("column_name") para características de passagem, ou CustomUDF para transformações por linhas. Veja Funções suportadas para tipos de fonte compatíveis.
• 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 as chaves de agregação ou consulta (chaves primárias). Necessária para DeltaTableSource e StreamSource. Por exemplo, ["user_id"] agrega ou pesquisa por utilizador. Omita e RequestSourceFeatureViewSource.
• timeseries_column: A coluna de carimbo temporal usada para agregação de janelas temporais ou seleção de valores mais recentes. Necessária para DeltaTableSource e StreamSource. Omita e RequestSourceFeatureViewSource.
• 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. Uma funcionalidade não pode ser eliminada enquanto ainda tiver funcionalidades materializadas. Apaga primeiro as funcionalidades materializadas, depois elimina a funcionalidade. 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

FirstN, LastN, , e LastDistinct usar as funcionalidades timeseries_column para ordenar as linhas de entrada e devolver um array contendo até n valoresFirstDistinct. O n parâmetro deve ser um inteiro positivo. FirstN e FirstDistinct selecionar valores do mais antigo ao mais recente. LastN e LastDistinct selecionar valores do mais recente para o mais antigo, e depois devolver os valores selecionados por ordem de carimbo temporal. FirstDistinct e LastDistinct remover valores duplicados ao selecionar valores nessa direção.

Por exemplo, se as linhas fonte de uma entidade estiverem ordenadas por event_time , ["A", "A", "B", "C", "B", "B"]as seguintes funções retornam:

FirstN, LastN, FirstDistinct, e LastDistinct requerem databricks-feature-engineering a versão 0.17.0 ou posterior.

CustomUDF

CustomUDFaplica uma função Python (UDF) registada no Unity Catalog a cada linha. Use-o para transformar entradas de pedidos ou combinar valores de funcionalidades. Não agrega linhas nem define uma janela temporal.

CustomUDF(
    function_name="main.ecommerce.log_amount_udf",
    input_bindings={"amount": "transaction_amount"},
)

input_bindings mapeia cada nome de parâmetro UDF para uma entrada. Para RequestSource, a entrada é o nome da coluna de origem. Para FeatureViewSource, é uma referência de funcionalidades a montante. Atribui todos os parâmetros da UDF, incluindo os parâmetros com os valores predefinidos. Os tipos de entrada devem corresponder exatamente aos tipos de parâmetros UDF, sem lançamentos numéricos implícitos. Use tipos de entrada e retorno escalar.

As funcionalidades apoiadas CustomUDF por Delta não podem ser materializadas ou servidas online. Para transformar valores de funcionalidades apoiadas em tabelas para treino e serviço, defina uma funcionalidade de agregação apoiada em Delta ou seleção de colunas e referencia-a através de FeatureViewSource.

CustomUDF não é suportado por StreamSource. Para transformar a saída de uma funcionalidade de streaming, faça referência a essa funcionalidade através FeatureViewSourcede .

CustomUDF com RequestSource requer databricks-feature-engineering a versão 0.17.0 ou posterior.

Para usar um CustomUDF, precisa do EXECUTE privilégio no UDF, do USE CATALOG privilégio no seu catálogo pai e do USE SCHEMA privilégio no seu esquema pai.

O exemplo seguinte utiliza o NumPy para calcular log(1 + amount), reduzindo a escala de grandes quantidades de transações. Executa-o em computação serverless com dependências UDF personalizadas ativadas. O main.ecommerce esquema tem de existir.

spark.sql("""
CREATE OR REPLACE FUNCTION main.ecommerce.log_amount_udf(amount DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
ENVIRONMENT (
  dependencies = '["numpy==1.26.4"]',
  environment_version = '5'
)
AS $$
import numpy as np

if amount is None or not np.isfinite(amount) or amount < 0:
    return None
return float(np.log1p(amount))
$$
""")

Registar uma funcionalidade que vincule a coluna transaction_amount de pedido ao parâmetro amountUDF :

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    CustomUDF, FieldDefinition, RequestSource, ScalarDataType,
)

fe = FeatureEngineeringClient()

log_transaction_amount = fe.create_feature(
    source=RequestSource(
        schema=[
            FieldDefinition(name="transaction_amount", data_type=ScalarDataType.DOUBLE),
        ]
    ),
    function=CustomUDF(
        function_name="main.ecommerce.log_amount_udf",
        input_bindings={"amount": "transaction_amount"},
    ),
    catalog_name="main",
    schema_name="ecommerce",
    name="log_transaction_amount",
)

O ENVIRONMENT UDF configura dependências para computação offline. Para o serviço online, declare também as encomendas em create_feature_spec(extra_pip_requirements=...) ou log_model(extra_pip_requirements=...). Eles não são copiados automaticamente do UDF. Veja Dependências de Serviço de Funcionalidades e Dependências de modelos.

CustomUDF funcionalidades não podem ser concretizadas. Os UDFs apoiados por pedidos e funcionalidades funcionam sob demanda durante a formação e o serviço. Cada UDF numa cadeia de dependências adiciona computação, por isso mantém funções e cadeias pequenas. Os UDFs devem tratar de entradas em falta, que podem ser None offline ou NaN online.

Para orientações sobre como lidar com valores em falta, veja Como lidar com valores de funcionalidades em falta.

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 Suporta as seguintes fontes 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 Stream (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).

Para um DeltaTableSource, ColumnSelection as funcionalidades suportam filter_condition e transformation_sql, aplicado antes da seleção de valor mais recente, o mesmo que as funcionalidades de agregação.

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-lhe filtrar linhas da tabela de origem antes de calcular agregações ou selecionar o valor da coluna mais recente. 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
    lateness: Optional[SourceLateness] = None,       # Optional: Expected source settling time
)

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 ou seleção de colunas. 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 ou seleção de colunas. 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.
• lateness: Um SourceLateness objeto que descreve quanto tempo a fonte normalmente demora a ficar completa em tempo de evento. Se for omitido, a fonte é considerada completa imediatamente.

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

SourceLateness.settling_delay é a forma recomendada de simular, durante o treino, um atraso ETL consistente que afeta a materialização online. O Azure Databricks recua o tempo de avaliação de treino elegível por este período, de modo a que um exemplo de treino não utilize dados que ainda estariam em trânsito online. Durante a materialização, o Azure Databricks espera o mesmo período antes de publicar uma janela concluída e serve a última janela concluída durante o período intermédio.

Por exemplo, suponha que um trabalho diário de ETL é concluído 8 horas depois da meia-noite num fuso horário local onde a meia-noite corresponde às 07:00 UTC. Use um atraso de decantação de 8 horas e um desfasamento de janela de 7 horas:

from datetime import timedelta
from databricks.feature_engineering.entities import (
    DeltaTableSource,
    SourceLateness,
    TumblingWindow,
)

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    lateness=SourceLateness(settling_delay=timedelta(hours=8)),
)

window = TumblingWindow(
    window_duration=timedelta(days=1),
    offset=timedelta(hours=7),
)

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

Expressões suportadas transformation_sql

As mesmas regras aplicam-se a transformation_sql em DeltaTableSource e StreamSource.

transformation_sql suporta quaisquer expressões linha a linha; operações avaliadas independentemente para cada linha. Não alteram o número de linhas nem a correspondência direta com a fonte. Expressões por linhas incluem renomeações de colunas, casts, operações aritméticas e mais.

Operações que alteram a forma ou a contagem de linhas não são suportadas, como agregações como SUM() ou COUNT(). Use AggregationFunction na definição de funcionalidades em vez disso.

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] = None,      # Optional: SQL WHERE clause applied before aggregation
    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
    lateness: Optional[SourceLateness] = None,    # Optional: Expected source settling time
)

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'").
• transformation_sql(opcional): Uma expressão SQL SELECT aplicada antes da agregação ou seleção de colunas, usando referências com prefixo de pontos às key estruturas e.value Suporta as mesmas expressões por linhas que DeltaTableSource. Se omitido, a fonte usa todas as colunas (*).
• dataframe_schema: O esquema JSON Spark StructType da saída projetada. Obrigatório se definires transformation_sql.
• lateness: Um SourceLateness objeto que descreve quanto tempo o fluxo normalmente demora a ficar completo em tempo de evento. Consulte SourceLateness.settling_delay.

from databricks.feature_engineering.entities import StreamSource

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

Derive-se dataframe_schema ao executar a projeção contra a tabela de ingestão do Stream, que expõe as key estruturas e value .

transformation_sql = (
    "value.amount * value.conversion_rate AS converted_amount, "
    "struct(value.user_id AS user_id, value.event_time AS time) AS event"
)

ingestion_table = "my_catalog.my_schema.events_ingestion"
dataframe_schema = spark.sql(
    f"SELECT {transformation_sql} FROM {ingestion_table} LIMIT 0"
).schema.json()

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

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 pode ser usado com as funções Custom UDF ou ColumnSelection Feature View. 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.

FeatureViewSource

FeatureViewSource usa as saídas de outras Feature Views como entradas para um CustomUDF. A encadeamento de características cria um grafo direcionado acíclico (DAG). Por exemplo, uma funcionalidade de margem pode combinar agregados de receita e custos, e outra funcionalidade pode transformar a margem.

Use databricks-feature-engineering a versão 0.18.0 ou posterior para FeatureViewSource.

Passa uma lista de Feature objetos para features, não cadeias de nomes de características. Recuperar funcionalidades registadas com get_feature. Em input_bindings, use a funcionalidade full_namede cada registada . Para uma funcionalidade local, não registada, use o seu name em vez disso.

O exemplo seguinte assume duas funcionalidades registadas, revenue_sum_7d e cost_sum_7d, que retornam DOUBLE valores por customer_id e usam event_time para computação em ponto no tempo:

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import CustomUDF, FeatureViewSource

fe = FeatureEngineeringClient()
revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d")
cost = fe.get_feature(full_name="main.ecommerce.cost_sum_7d")

spark.sql("""
CREATE OR REPLACE FUNCTION main.ecommerce.margin_udf(revenue DOUBLE, cost DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
AS $$
import math

if revenue is None or cost is None:
    return None
if not math.isfinite(revenue) or not math.isfinite(cost) or revenue <= 0:
    return None
return (revenue - cost) / revenue
$$
""")

margin = fe.create_feature(
    source=FeatureViewSource(features=[revenue, cost]),
    function=CustomUDF(
        function_name="main.ecommerce.margin_udf",
        input_bindings={"revenue": revenue.full_name, "cost": cost.full_name},
    ),
    catalog_name="main",
    schema_name="ecommerce",
    name="margin",
)

Aplicam-se as seguintes restrições:

• Apenas CustomUDF é suportado como função. As funcionalidades a montante podem ser agregações, seleções de colunas ou outras CustomUDF funcionalidades.
• Omita entity e timeseries_column na característica derivada. Cada funcionalidade a montante mantém a sua própria entidade, carimbo temporal e definição de janela.
• Uma funcionalidade tem uma única fonte. Para combinar um valor de pedido com uma funcionalidade com base em tabelas, defina uma RequestSource característica e referencia-a ambos através de FeatureViewSource.
• Cada funcionalidade a montante declarada deve ser usada em input_bindings. Bicicletas não são permitidas.
• Registar as funcionalidades a montante antes de registar a funcionalidade derivada. Grafos locais, não registados, podem ser usados com create_training_set para experimentação.
• Para treino ou serviço, é necessário o privilégioREAD FEATURE ou MANAGE sobre a característica derivada e as suas características transitivas a montante. Use nomes de funcionalidades distintos ao longo do gráfico para registo e serviço, mesmo entre catálogos ou esquemas.
• Uma funcionalidade pode referenciar até 20 funcionalidades diretas a montante. Grafos registados suportam uma profundidade máxima de cinco características ao longo de um caminho de dependência, incluindo a característica base.
• FeatureViewSource As características não podem ser materializadas ou avaliadas com compute_features. Use create_training_set para os avaliar offline. Para o serviço online, materialize as funcionalidades suportadas a montante suportadas por mesas.

Para avaliação de dependências e seleção de saída, veja Treinar com funcionalidades FeatureViewSource. Para implementação, veja funcionalidades derivadas do Serve.

API de treino e inferência

create_training_set e score_batch calcular valores de características corretas em momento em tempo a pedido a partir dos dados de origem. Para funcionalidades que suportam materialização offline, como agregações de janelas deslizantes em fontes de tabelas delta, materializar funcionalidades primeiro para uma loja offline melhora o desempenho de ambas as operações. Quando funcionalidades offline materializadas estão disponíveis, as operações leem dados offline pré-computados em vez de recalcular os valores das características a partir da fonte. Consulte Materialize Feature Views para materializar funcionalidades numa loja offline.

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 quatro tipos de janelas para controlar o comportamento de lookback em agregações baseadas em janelas temporais. Os tipos de janelas disponíveis dependem da fonte da funcionalidade: as funcionalidades de origem em streaming podem usar janelas rolantes e dentes de serra, e as funcionalidades de origem em lote podem usar janelas rolantes, de tumbling e deslizantes.

• 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.
• As janelas Sawtooth mantêm uma janela longa de retrospetiva fresca sobre uma fonte de streaming usando um caminho híbrido por lote e streaming. Ver janela Sawtooth.

A ilustração seguinte mostra os tipos de janelas com rolamento, deslizamento, rolamento e dente de serra.

Temporização da janela temporal

Use delay para avaliar uma janela num ponto analítico anterior no tempo. Por exemplo, uma janela de 30 dias com um atraso de 7 dias calcula um valor de 30 dias a partir de uma semana antes do momento da avaliação. delay é independente do tempo de chegada da fonte. Para modelar o tempo que os dados de origem demoram a chegar, configure SourceLateness.settling_delay em vez disso.

Quando ambos os contextos estão presentes, compõem. O Azure Databricks trata a janela como completa após o atraso de estabilização da fonte e avalia-a usando o atraso analítico.

Use offset para alterar o alinhamento dos limites fixos das janelas. Por defeito, as janelas deslizantes e as janelas deslizantes estão alinhadas à meia-noite UTC. Por exemplo, um desfasamento de 22 horas alinha um limite diário às 22:00 UTC. Para aproximar limites num fuso horário local, configure um deslocamento estático em relação ao UTC. O deslocamento não ajusta para o horário de verão, não desloca o tempo de avaliação nem modela dados de chegada tardia.

A tabela seguinte resume o suporte para estes campos:

*Ponto: Para uma janela de rotação, o deslocamento deve ser mais curto que window_duration. Para uma janela deslizante, deve ser mais curta do que slide_duration.

Hora de início

Use start_time para definir a fronteira de tempo de evento mais antiga no UTC em que uma funcionalidade pode emitir uma saída. A fronteira é inclusiva. start_time Saídas de portas. Não restringe as linhas de origem históricas que uma janela pode ler, nem altera o alinhamento da janela. Se start_time cair entre duas fronteiras alinhadas, a primeira saída elegível de janela fixa é a próxima fronteira.

Com start_time, janelas de duração fixa podem emitir antes de uma duração completa da janela ter passado na fonte. Estes primeiros resultados utilizam qualquer história de origem disponível. Por exemplo, considere uma janela deslizante com um window_duration ano e um slide_durationdia , sobre uma fonte cujos dados começam a 1 de janeiro de 2024:

• Sem start_time, a funcionalidade é lançada pela primeira vez a 1 de janeiro de 2025, uma vez que se possa formar uma janela completa de um ano.
• Com start_time o filme previsto para 21 de agosto de 2024, o filme estreia a 21 de agosto de 2024. Esse conteúdo cobre apenas o histórico de fontes disponível até agora, desde 1 de janeiro de 2024. A janela atinge o seu período completo de um ano a 1 de janeiro de 2025 e produz resultados completos a partir daí.

Como start_time não altera o alinhamento das janelas, um valor entre dois limites alinhados não cria um novo limite. Para uma janela oscilante com limites diários à meia-noite UTC, a start_time das 06:00 UTC emite primeiro na próxima meia-noite. A start_time que aterra exatamente numa fronteira emite nessa fronteira, porque a fronteira é inclusiva.

Se start_time estiver desajustado, janelas deslizantes e janelas deslizantes de duração fixa emitem primeiro numa fronteira alinhada após a formação de uma janela completa. Janelas deslizantes e janelas móveis ao longo da vida emitem assim que existem dados de fonte elegíveis.

Por exemplo:

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

window = SlidingWindow(
    window_duration=timedelta(days=365),
    slide_duration=timedelta(days=1),
    start_time=datetime(2024, 8, 21),
)

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
    start_time: Optional[datetime.datetime] = 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.

# Compute a 7-day value as of one day before the evaluation time
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(days=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.

Use Last para limitar a frescura de um valor recente

Combina Last com RollingWindow quando um valor mais recente é válido apenas por um tempo limitado. Num momento de avaliação, a característica devolve o valor da linha com o carimbo temporal mais recente neste intervalo:

[evaluation_time - delay - window_duration, evaluation_time - delay)

Se a última linha do intervalo contiver um valor nulo, a característica devolve nula. Se quiseres excluir valores de entrada nulos, define a filter_condition na fonte.

Esta combinação difere de ColumnSelection. ColumnSelection devolve o valor não nulo mais recente observado sem o expirar com base na idade.

Para funcionalidades em lote, esta combinação tem um modo especial de materialização apenas online. Suporta apenas DeltaTableSource, Last, RollingWindow, e TableTrigger. Veja Materialize valores mais recentes limitados à frescura.

Janela deslizante

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
    delay: Optional[datetime.timedelta] = None
    offset: Optional[datetime.timedelta] = None
    start_time: Optional[datetime.datetime] = None

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=1),
    delay=timedelta(hours=2),
    offset=timedelta(hours=22),
)

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 que avança por um intervalo de deslizamento. Uma janela deslizante pode ter uma duração fixa ou uma duração vitalícia. Janelas de duração fixa sobrepõem-se, pelo que cada evento de origem pode contribuir para a agregação de funcionalidades em várias janelas. Uma janela de vida inclui todos os eventos de origem antes do fim da janela. As características no momento t agregam dados de janelas que terminam em ou antes de t (exclusivo). O Windows está alinhado com a época Unix.

class SlidingWindow(TimeWindow):
    window_duration: Optional[datetime.timedelta]
    slide_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None
    offset: Optional[datetime.timedelta] = None
    start_time: Optional[datetime.datetime] = None

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),
    delay=timedelta(hours=2),
    offset=timedelta(hours=22),
)

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

Janela de vida

Definir window_duration=None para criar uma janela vitalícia. Em cada limite de deslizamento, a característica agrega todos os eventos de origem da entidade com carimbos temporais anteriores a esse limite. Por exemplo, um slide de um dia produz um valor acumulado uma vez por dia.

As janelas de vida são suportadas apenas por SlidingWindow. RollingWindow e TumblingWindow requerem um finito window_duration.

from datetime import timedelta
from databricks.feature_engineering.entities import (
    AggregationFunction,
    DeltaTableSource,
    Feature,
    SlidingWindow,
    Sum,
)

lifetime_spend = Feature(
    source=DeltaTableSource(
        catalog_name="main",
        schema_name="store",
        table_name="transactions",
    ),
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(
        Sum(input="amount"),
        SlidingWindow(
            window_duration=None,
            slide_duration=timedelta(days=1),
        ),
    ),
    name="lifetime_spend",
)

Janela dente de serra

Uma janela dente de serra é uma agregação que suporta atualizações altamente recentes para eventos recentes, juntamente com a compactação diária de dados históricos. A sua aresta de fuga (mais antiga) avança em passos fixos e diários, enquanto a borda dianteira (recente) mantém-se atualizada com os eventos mais recentes, por isso o comprimento efetivo da janela "serra" ao longo de cada dia. A maior parte da janela é servida a partir de dados na tabela de ingestão do Stream, e apenas os dois dias mais recentes provêm da transmissão em direto. Este é um compromisso que calcula de forma eficiente janelas de longa duração (escalando para anos) mantendo-se responsivo a atualizações recentes.

As janelas em dente de serra materializam-se num percurso híbrido de lote e fluxo. Um pipeline batch mantém a maior parte da janela, enquanto um pipeline de streaming mantém os dados mais recentes frescos em tempo real. Os dois são fundidos em leitura, por isso, para o modelo ou consumidor que serve, é uma única funcionalidade.

Como a parte histórica da janela é calculada pelo oleoduto de lote, uma estrutura de dente de serra está pronta para servir pouco depois do início da materialização, mesmo quando a janela abrange meses ou anos. Uma janela rolante só está concluída após a duração total da janela. O mínimo window_duration deve ser superior a dois dias (o limite inferior imposto). A Databricks recomenda uma janela em dente de serra para durações superiores a 7 dias. Para janelas com mais de dois dias e até sete dias, escolha entre a precisão de comprimento fixo de uma janela rolante e a prontidão de produção mais rápida de uma janela de serra.

Para saber se uma funcionalidade dente de serra está pronta, abra a Vista de Funcionalidades no Explorador de Catálogos. Na secção de funcionalidades materializadas, o preenchimento por lote está concluído assim que o último tempo de materialização da funcionalidade avança e o seu estado mostra sucesso. A parte de fluxo é materializada por um oleoduto declarativo Lakeflow. Depois de a Feature View passar a validação, a funcionalidade materializada liga-se a esse pipeline, onde pode monitorizar o seu estado de execução.

As janelas dentes de serra requerem um StreamSource e são materializadas com StreamingMode.

class SawtoothWindow(TimeWindow):
    window_duration: datetime.timedelta

As bordas de uma janela em dente de serra movem-se de forma diferente das de uma janela móvel: a borda de ataque acompanha o evento mais recente, enquanto a borda de fuga avança uma vez por dia em vez de continuamente. Todos os dias, a um limite fixo das 18:00 UTC, a borda de fuga avança até à fronteira UTC-meia-noite desse dia. Como resultado, a janela efetiva é ligeiramente mais longa e window_duration cresce ao longo do dia antes de regressar um dia no corte seguinte. O treino e o serviço usam o mesmo limite das 18:00 UTC, pelo que o treino offline e o serviço online mantêm-se consistentes.

As janelas dentes de serra suportam as Sumfunções de agregação , Avg, Count, Min, StddevSampVarPopMaxLastVarSampFirstStddevPope agregação.

Exemplo da janela dente-de-serra

O exemplo seguinte mostra uma contagem de 7 dias das transações de um utilizador. A borda de ataque acompanha o evento atual enquanto a borda de fuga avança um dia de cada vez. Para os eventos de 10 de março, a janela remonta a cerca de 3 de março. À medida que o dia 10 de março avança, a borda de ataque continua a avançar enquanto a borda de fuga se mantém, pelo que o vão coberto aumenta. Depois, no início de 11 de março, a borda traseira avança para cerca de 4 de março. A janela efetiva é sempre um pouco maior do que sete dias. Os dois dias mais recentes são servidos a partir da transmissão em direto, e os primeiros dias são servidos pela tabela de ingestão da Stream.

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

# 7-day window kept continuously fresh with streaming data
window = SawtoothWindow(window_duration=timedelta(days=7))

Limitações das janelas dentes de serra

• O delay parâmetro não é suportado.
• SourceLateness.settling_delay não é suportado.
• Funções de agregação que não sejam Sum, Avg, Count, Min, MaxFirst, Last, StddevPopVarSampVarPope StddevSamp não são suportadas (por exemplo, ApproxCountDistinct, ApproxPercentile, FirstN, LastN, FirstDistinct, , e ).LastDistinct
• As janelas de dente de serra requerem um StreamSource. A DeltaTableSource não é suportado.

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

Uso CronSchedule para funcionalidades de agregação em lote. Por defeito, o Azure Databricks deriva um horário a partir da janela de agregação. Um cronograma derivado tem em conta o período da janela, a janela delay e offset, e a fonte settling_delay , de modo que uma execução não publica uma janela antes de se esperar que os seus dados de origem estejam completos. Os horários derivados suportam janelas de tumbling e deslizamento.

Para solicitar um horário derivado, omita a expressão cron. CronSchedule() e a forma explícita CronSchedule(mode=CronScheduleMode.DERIVED) são equivalentes:

from databricks.feature_engineering.entities import (
    CronSchedule,
    CronScheduleMode,
)

trigger = CronSchedule(mode=CronScheduleMode.DERIVED)

Não defina quartz_cron_expression com CronScheduleMode.DERIVED. Quando recupera a funcionalidade materializada, o schedule devolvido pode conter a expressão cron que o Azure Databricks calculou.

Para controlar diretamente o cronograma, forneça uma expressão cron de quartzo. CronScheduleMode.MANUAL é inferido quando fornece uma expressão:

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

Use TableTrigger para ColumnSelection funcionalidades ou funcionalidades de agregação (AggregationFunction) com suporte de um DeltaTableSource. O pipeline corre sempre que a tabela Delta a montante recebe um novo commit.

Para funcionalidades de agregação, o pipeline é limitado para não correr em todos os commits. O pipeline executa-se no máximo uma vez por metade da janela da funcionalidade, mas nunca mais frequentemente do que a cada 5 minutos. Por exemplo, uma funcionalidade com uma janela de desaceleração de 1 hora é executada no máximo uma vez a cada 30 minutos, ou uma funcionalidade com uma janela de 8 horas corre no máximo uma vez a cada 4 horas. O piso de 5 minutos aplica-se quando metade da janela é menor do que isso, por isso janelas com 10 minutos ou menos funcionam no máximo uma vez a cada 5 minutos. Funcionalidades de agregação cuja janela inferior a 5 minutos não podem usar TableTrigger, utilizam um gatilho de streaming em vez disso.

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

Cada funcionalidade usa um gatilho; As opções por tipo de funcionalidade são:

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