Skip to main content
Last updated on

Use a Lakeflow pipeline as a feature source

SdpSource defines Feature Views on the output table of a triggered Lakeflow pipeline, and Feature Store runs the pipeline for you before each materialization and backfill. Use it to serve and train on features that need joins, multi-step transforms, or custom windowing, without scheduling or backfilling the pipeline yourself.

When to use SdpSource​

Use SdpSource when your features need logic that a DeltaTableSource with transformation_sql can't express. You write that logic in a Lakeflow pipeline, and then point Feature Views at one of the pipeline's output tables.

With SdpSource, Feature Store orchestrates the pipeline and manages syncs into online and offline stores:

  • Each materialization run updates the pipeline before it reads the output table.
  • Materialized features sync to their online and offline stores each time materialization runs, on the trigger or schedule that you set.
  • A backfill fully refreshes the output table, and then backfills the requested features over the time ranges that you pass.
  • You can register many Feature Views on one pipeline, using ColumnSelection or aggregation functions.

If you schedule the pipeline yourself, or if the pipeline is continuous, use a DeltaTableSource on the pipeline's output table instead. Feature Store then reads the table but doesn't run the pipeline.

How Feature Store runs the pipeline​

Each materialization run for SdpSource features does the following:

  1. The run starts an update of the pipeline. If an update is already running, the run waits for it to finish first. If that update completed successfully, is a full update, and started no more than 10 minutes before the run started, the run reuses it. Otherwise, the run starts its own update. Backfill updates and partial or validate-only updates aren't reused. A pipeline runs one update at a time, so runs that share a pipeline take turns.
  2. After the update completes, Feature Store reads the output table and materializes the features the same way it does for a DeltaTableSource.

If the pipeline update fails, is canceled, or doesn't finish within 4 hours, the materialization run fails and no feature values are written for that run.

A backfill doesn't follow these steps. It starts its own full refresh of the output table and doesn't reuse another update. See Backfill pipeline source features.

A reused update might have read the input tables before the commit that started the run. Rows from that commit then appear in the feature values after the next run. Runs on the same pipeline that start at the same time don't always share one update, and can each run their own update, one after another.

A pipeline's output table doesn't exist until the pipeline's first update creates it, so you can create a Feature View before the pipeline has ever run. Feature Store tracks this with the source's pipeline_state:

  • PENDING: The pipeline hasn't run yet, so the output table doesn't exist. Feature Store validates the feature's columns after the pipeline first runs.
  • RESOLVED: The output table exists, and the feature's columns are validated against it.
  • ERROR: The pipeline run or the output table validation failed. Fix the pipeline or the feature definition, and then run the pipeline again.

Feature Store resolves the state each time it uses the feature: when a materialization runs, and when you call create_training_set, compute_features, or refresh_sdp_source.

Requirements​

Create features from a pipeline​

To define and materialize Feature Views on a pipeline's output table, follow these steps.

Find the pipeline ID​

The pipeline ID appears in the pipeline details pane of the pipelines UI.

Define an SdpSource​

Name the pipeline and the output table to read. To use a TableTrigger, also list the tables the pipeline reads from in input_tables:

Python
from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import SdpSource

fe = FeatureEngineeringClient()

source = SdpSource(
    pipeline_id="1234abcd-5678-90ef-1234-567890abcdef",
    output_table="main.features.tutor_review_stats",
    input_tables=["main.raw.reviews", "main.raw.tutors"],
)

Create Feature Views on the source​

You can create many Feature Views on one source. The following example creates a ColumnSelection feature for a value that the pipeline computes and an aggregation feature that Feature Store computes over the output table:

Python
from datetime import timedelta
from databricks.feature_engineering.entities import (
    AggregationFunction,
    Avg,
    ColumnSelection,
    TumblingWindow,
)

avg_score_7d = fe.create_feature(
    source=source,
    entity=["tutor_id"],
    timeseries_column="window_end",
    function=ColumnSelection("avg_score_7d"),
    catalog_name="main",
    schema_name="features",
    name="tutor_avg_score_7d",
)

daily_review_count = fe.create_feature(
    source=source,
    entity=["tutor_id"],
    timeseries_column="window_end",
    function=AggregationFunction(
        Avg(input="review_count"),
        time_window=TumblingWindow(window_duration=timedelta(days=1)),
    ),
    catalog_name="main",
    schema_name="features",
    name="tutor_daily_review_count",
)

Run the pipeline​

If the pipeline hasn't run yet, run it one time so that the output table exists and the features resolve. refresh_sdp_source starts one update for each pipeline behind the given features, waits for the updates to finish, and returns a mapping from pipeline ID to update ID:

Python
updates = fe.refresh_sdp_source(features=[avg_score_7d, daily_review_count], wait=True)

Materialize the features​

All features in one materialize_features call must use the same pipeline. To materialize the ColumnSelection feature to an online store, run the following:

Python
from databricks.feature_engineering.entities import OnlineStoreConfig, TableTrigger

fe.materialize_features(
    features=[avg_score_7d],
    online_config=OnlineStoreConfig(
        catalog_name="main",
        schema_name="features_online",
        table_name_prefix="tutor_",
        online_store_name="my_online_store",
    ),
    trigger=TableTrigger(),
)

Check the features in Catalog Explorer​

The feature's Definition section shows the pipeline and the output table, and Pipeline state shows Resolved after the pipeline's first successful run. The materialized features section shows the result of each materialization run. See Explore Feature Views in Unity Catalog.

For all SdpSource parameters, see SdpSource.

Develop with pipeline source features​

Training, batch scoring, and compute_features don't run the pipeline. To work with the latest pipeline output, run the pipeline with refresh_sdp_source first:

Python
fe.refresh_sdp_source(features=[avg_score_7d, daily_review_count], wait=True)

For more about training with Feature Views, see Train with pipeline source features.

Use feature tables that a pipeline writes​

If a Lakeflow pipeline already writes a feature table, you can define Feature Views over it without changing the pipeline. Use the feature table as the SdpSource output table, and create one ColumnSelection feature for each column you want to serve. A ColumnSelection feature returns the latest value of a column for each entity, so the feature passes the table's values through unchanged.

Map the feature table to the feature definition as follows:

  • output_table: The three-part name of the feature table. The pipeline must write this table, so for a feature table that something else writes, use a DeltaTableSource instead.
  • entity: The columns of the table's primary key that identify the entity, such as user_id.
  • timeseries_column: A timestamp column in the table. Every feature on an SdpSource requires one, so if the feature table has no timestamp column, add one in the pipeline.
  • input_tables: The tables that the pipeline reads. A ColumnSelection feature materializes on a table trigger, so it requires them.

The following example serves the country column of a feature table that a pipeline writes:

Python
from databricks.feature_engineering.entities import ColumnSelection, SdpSource

source = SdpSource(
    pipeline_id="1234abcd-5678-90ef-1234-567890abcdef",
    output_table="main.features.user_profile",
    input_tables=["main.raw.users"],
)

user_country = fe.create_feature(
    source=source,
    function=ColumnSelection("country"),
    entity=["user_id"],
    timeseries_column="updated_at",
    catalog_name="main",
    schema_name="features",
    name="user_country",
)

To serve the feature online, materialize it as described in Materialize with a table trigger. Feature Store runs the pipeline before each materialization, so the online values follow the feature table.

Materialize with a table trigger​

A TableTrigger on an SdpSource feature watches the source's input_tables, not the output table. Features on the same output table that you materialize in one materialize_features call share a job, and that job watches the input_tables of all of them. An update to any of those tables starts a run for every feature in the job, and TABLE_TRIGGER_CONDITION_ALL_UPDATED waits until every one of those tables has updated. The job can watch up to 25 tables in total, and more than 10 only when file events are enabled on every watched table. See How tables count toward the trigger limit. When an input table updates, the materialization job runs the pipeline and then materializes the features. ColumnSelection features materialize on a table trigger, so an SdpSource used by a ColumnSelection feature must list input_tables. Aggregation features can use a CronSchedule instead.

Feature Store doesn't check that the pipeline reads from the tables in input_tables. If you leave out a table that the pipeline reads, updates to that table don't start a run.

When a pipeline reads several tables that update at different times, use the TableTrigger settings to control when a run starts:

Python
from datetime import timedelta
from databricks.feature_engineering.entities import TableTrigger, TableTriggerCondition

trigger = TableTrigger(
    condition=TableTriggerCondition.TABLE_TRIGGER_CONDITION_ALL_UPDATED,
    wait_after_last_change=timedelta(minutes=10),
    min_time_between_triggers=timedelta(hours=1),
)
  • condition: Start a run when any input table updates (TABLE_TRIGGER_CONDITION_ANY_UPDATED, the default) or only after every input table has updated (TABLE_TRIGGER_CONDITION_ALL_UPDATED).
  • wait_after_last_change: Wait until the input tables have had no updates for this long, so that a series of updates lands before the run starts.
  • min_time_between_triggers: Start runs at most this often.

For the allowed values, see TableTrigger.

Backfill pipeline source features​

A pipeline update picks up rows that arrive late in the pipeline's sources, so the next materialization run publishes them without a backfill. Backfill an SdpSource feature when a normal pipeline update can't correct the output, for example, when you change a streaming table's logic, correct rows in its source, or need to recover rows dropped past a watermark. A backfill fully refreshes the output table, and then backfills the requested features over the time ranges you pass. See Backfill features from a pipeline source.

Change a pipeline after you register features​

Don't change a pipeline's input or output tables after you register Feature Views on it. Feature Store records input_tables when you create a feature and doesn't update them when the pipeline changes. If an output table is removed or its schema no longer fits the feature, the next materialization run fails. Training data can also become inconsistent with what was materialized earlier.

To change a pipeline's inputs or outputs, create a new pipeline with the change, and register new Feature Views on it.

Deleting a Feature View doesn't delete its pipeline. If you delete the pipeline, materialization runs for its features fail. Create new features from an existing pipeline.

Troubleshooting​

CAN_RUN permission is required on the pipeline​

The pipeline doesn't exist, or you don't have CAN RUN, CAN MANAGE, or IS OWNER permission on it. You need this permission to create a feature from the pipeline and to materialize its features. Check the pipeline ID in the error, or ask the pipeline owner to grant the permission.

Pipeline '<id>' is continuous​

SdpSource supports only triggered pipelines. Switch the pipeline to triggered mode, or use a DeltaTableSource on the pipeline's output table.

output_table '<name>' is not an output of pipeline '<id>'​

The output table exists, but the pipeline doesn't write it. Check the table name against the outputs listed in the error.

A materialization run fails on the pipeline update​

The run's error names the pipeline update. Open the update in the pipeline's run history to see why it failed. If the update was still running after 4 hours, the run fails but the update keeps running.

A table trigger starts an SDP-sourced feature's materialization when its input_tables update​

The features use a TableTrigger, or are ColumnSelection features, but their SdpSource has no input_tables. Use a CronSchedule for aggregation features, or recreate the features with input_tables.

Limitations​

  • SdpSource features are batch only. Streaming materialization isn't supported.
  • Only triggered pipelines are supported. For continuous pipelines or pipelines that you schedule yourself, use a DeltaTableSource on the output table.
  • CustomUDF features don't support SdpSource.
  • input_tables can list up to 25 tables, with no duplicates. Each table must exist when you create the feature. A TableTrigger watches the input_tables of every feature in the job, up to 25 tables in total. To watch more than 10 tables, enable file events on every watched table. With more tables than the trigger can watch, materialize_features is rejected.
  • A TableTrigger can wait for any input table or for all input tables to update. It can't wait for a specific subset.
  • One materialize_features call can't mix SdpSource features with DeltaTableSource features or with features from a different pipeline.
  • A backfill fully refreshes the whole output table, even if you pass a time range. The time range applies only to aggregation features.
  • Deploying SdpSource features with Declarative Automation Bundles isn't supported.
On this page