Feature Views

Important

This feature is in Public Preview. Workspace admins can control access to this feature from the Previews page. See Manage Azure Databricks previews.

Feature Views enable you to define and compute features from data sources. Features can be defined using a variety of sources (Delta table, Kafka Stream, and request-time data) and computations (time-windowed aggregations, simple column selections, and more). This guide covers the following workflows:

  • Feature development workflow
    • Use create_feature to define Unity Catalog feature objects that can be used in model training and serving workflows.
    • Alternatively, construct Feature objects locally and use register_feature to persist them to Unity Catalog later. Locally constructed features can be used with create_training_set before registration.
  • Model training workflow
    • Use create_training_set to calculate point-in-time aggregated features for machine learning. For detailed documentation on training with Feature Views, see Train models with Feature Views.
  • Feature materialization and serving workflow
    • After defining a feature with create_feature or retrieving it using get_feature, you can use materialize_features to materialize the feature or set of features to an offline store for efficient reuse, or to an online store for online serving.
    • Use create_training_set with the materialized view to prepare an offline batch training dataset.

For API details, see Feature Views API reference.

Requirements

  • Serverless compute or a classic compute cluster running Databricks Runtime 17.0 ML or above.

  • You must install the custom Python package. Run the following lines of code each time you run a notebook:

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

Quickstart example

For a runnable quickstart notebook, see Example notebook.

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

Example notebook

Feature Views quickstart notebook

Get notebook

Streaming features

Use streaming features when feature values must update continuously instead of on a batch schedule. Streaming and batch features use the same Feature constructors, aggregation functions, and training and serving workflows.

Streaming features do not materialize to an offline store. For training and batch inference, Databricks calculates feature values from the source.

Streaming features have the following requirements:

  • You must provide an online_config. Streaming features do not support offline_config.
  • You cannot combine streaming and batch features in one materialize_features call. Make a separate call for each trigger type.
  • transformation_sql is not supported for streaming features.
  • Streaming materialization processes only records that arrive after the pipeline starts and does not backfill historical records. Rolling-window aggregates return complete results only after the first full window of data arrives.

Choosing a streaming feature source

Choose a source based on your freshness requirements and existing ingestion setup:

  • Use a StreamSource when sub-second freshness is the priority. StreamSource features provide a p99 end-to-end latency of 200 milliseconds. First set up a Stream, then reference it using a StreamSource. Stream sources support Kafka as input and automatically maintain an ingestion Delta table as a historical copy of the data for training.
  • Use a DeltaTableSource when you already have a low-latency ingestion path into a Delta table. Expect freshness on the order of tens of seconds.
  • Use Zerobus to populate a DeltaTableSource when you do not already have a low-latency ingestion path. Zerobus ingestion takes on the order of tens of seconds, so expect sub-minute feature freshness.

Define a streaming feature using a StreamSource

A StreamSource references a Stream by its three-part name (catalog.schema.stream_name). A Stream is not a Unity Catalog securable object, but it is scoped to a Unity Catalog schema and access is governed by the Stream's ingestion table. Column references in entity, timeseries, and function definitions must be prefixed with value. or key. to indicate which part of the Kafka message to read. Nested fields are supported using dot notation (for example, 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)),
    ),
)

Define a streaming feature using a DeltaTableSource

To materialize a feature defined on a DeltaTableSource as a streaming feature, pass StreamingMode as the trigger to materialize_features. The feature definition uses the same APIs as a batch feature backed by a DeltaTableSource. Delta table sources support aggregation and column selection features.

The source Delta table must have change data feed (CDF) enabled by setting delta.enableChangeDataFeed=true.

The following example defines and materializes an aggregation feature with a Delta table source.

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 a Delta table populated by Zerobus

A Delta table populated by Zerobus can serve as a streaming feature source. Zerobus does not set delta.enableChangeDataFeed=true automatically. You must set this property manually on the target Delta table before using it as a streaming feature source.

Filter conditions on streaming sources

Use filter_condition to filter rows before aggregation for either a StreamSource or DeltaTableSource.

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

Column selection from streaming sources

ColumnSelection features work with streaming sources. The selected column represents the latest value from the source for each entity while respecting point-in-time accuracy.

Column selection features do not have a TTL. To remove a selected value from the online store, the source must emit a null value for the selected column.

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

Access nested fields from a StreamSource

For a StreamSource, you can access nested JSON fields using dot notation (for example, value.nested_field.amount). At serving time, the request payload and response use leaf node names (for example, amount instead of value.amount). Leaf node names must be unique across all entity, timeseries, and feature names within a model or Feature Spec, because the serving endpoint uses leaf names to route values.

Time windows for streaming features

Streaming features support only RollingWindow for aggregations. Rolling windows continuously recompute over the most recent data, which aligns with the real-time nature of streaming sources. TumblingWindow and SlidingWindow are designed for batch computation over fixed historical intervals.

Streaming features example notebook

Streaming Feature Views quickstart notebook

Get notebook

Model training and inference

To train models and run batch inference with Feature Views, including log_model(), score_batch(), and create_training_set(), see Train models with Feature Views.

Feature materialization

After you define features, you can materialize them to offline or online stores for efficient reuse in training and serving workflows. After materializing features, you can serve models using CPU model serving. For details, see Materialize Feature Views.

Best practices

Feature naming

  • Use descriptive names for business-critical features.
  • Follow consistent naming conventions across teams.
  • Use auto-generated names as you begin developing features.

Time windows

  • Align window boundaries with business cycles (daily, weekly).
  • Shorter windows capture recent trends but can be noisy. Longer windows produce more stable feature distributions but might miss recent behavioral shifts. Choose based on how quickly the underlying signal changes for your use case. For example, a 7-day window smooths out daily fluctuations and produces consistent model inputs, while a 1-hour window reacts quickly to behavioral changes but might introduce variance that degrades model performance. If your model's accuracy degrades when the distribution shifts, use a longer window to stabilize inputs.
  • Tumbling and sliding windows are more scalable than rolling (continuous) windows. Start with sliding windows for most use cases.

Performance

  • Materialize features from the same data source in a single materialize_features call to minimize data scans.
  • Use the same granularity (for example, all 1-hour or all 1-day slide durations) for features on the same data source to enable better grouping during materialization.

Entity columns vs. filter conditions

Use this decision guide when working with features from the same source table:

Use entity (on create_feature) when you need different aggregation levels:

  • Customer-level features (one row per customer): entity=["customer_id"]
  • Customer-merchant features (multiple rows per customer): entity=["customer_id", "merchant_id"]
  • Different aggregation levels can share the same DeltaTableSource: specify different entity values on each feature definition

Use filter_condition (on DeltaTableSource) when you need to filter rows at the same aggregation level:

  • High-value transactions only: filter_condition="amount > 100" (still aggregated per customer)
  • Completed orders only: filter_condition="status = 'completed'" (still aggregated per customer)

Rule of thumb: If your change would result in a different number of rows per entity value, use different entity values on your feature definitions. If you're just filtering which rows contribute to the same aggregation, use filter_condition on the source.

Common patterns

Customer analytics

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

Trend analysis

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

Seasonal patterns

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

Limitations

  • Names of entity and timeseries columns must match between the training (labeled) dataset and the feature definitions when used in the create_training_set API.
  • The column name used as the label column in the training dataset should not exist in the source tables used for defining Features.
  • A limited list of functions (UDAFs) is supported in the create_feature API. See Supported functions.
  • Entity columns cannot be of type DATE or TIMESTAMP.
  • RequestSource supports only scalar data types defined in ScalarDataType (INTEGER, FLOAT, BOOLEAN, STRING, DOUBLE, LONG, TIMESTAMP, DATE, SHORT). Complex types such as arrays, maps, and structs are not supported.
  • RequestSource does not support aggregation functions or time windows. Only ColumnSelection functions can be used.
  • The set of entity column names, timeseries column names, and request feature column names must be globally unique across all sources in a training set or serving endpoint.
  • score_batch might not succeed on serverless compute. Work around this by using a classic compute cluster running Databricks Runtime 17.0 ML or above.

For materialization-specific limitations, see Limitations.

  • Last updated on