Point-in-Time Feature Lookups and Automated Feature Pipelines

1.Why a naive join leaks the future

Take a room-occupancy model. Sensor tables record temperature, humidity, light and CO2, and a ground-truth table records whether someone was in the room. Every table is keyed by room and by a timestamp ts. If you build the training set by joining on the room alone, a label observed at 8:50 can end up paired with a CO2 reading of 630 taken at 8:52. The model then trains on a value it could never have seen at prediction time. This kind of leakage is hard to spot, and it hurts the model's performance once deployed.

Point-in-time correctness fixes this. For each label row, the join takes the latest feature value recorded before that row's timestamp. Databricks calls this an AS OF join. If no value had been recorded yet, the feature is null. Use this pattern whenever feature values change over time: time series data, event data, or time-aggregated data.

Sources1

2.Declaring a time series feature table

In Unity Catalog, any table with a TIMESERIES primary key is a time series feature table. From Databricks Runtime 13.3 LTS onward, any Delta table in Unity Catalog that has primary keys and timestamp keys can be used this way. With FeatureEngineeringClient you declare the time column using the timeseries_columns argument. The legacy Workspace Feature Store uses timestamp_keys instead. In SQL you write PRIMARY KEY (user_id, event_timestamp TIMESERIES).

fe = FeatureEngineeringClient()

# Create a time series table for point-in-time joins
fe.create_table(
    name="catalog.schema.user_behavior_features",
    primary_keys=["user_id", "event_timestamp"],
    timeseries_columns="event_timestamp",  # Enables point-in-time logic
    df=features_df # DataFrame must contain primary keys and time series columns
)

If you really need a date as a plain exact-match key, with no as-of behaviour, the docs say to change the column type to STRING.

Sources1

3.Performing the as-of lookup

Declaring the table is half the job. The lookup has to say which column in your label DataFrame holds the event time. You do this by setting timestamp_lookup_key on the FeatureLookup. The store then matches the non-timestamp primary keys against lookup_key and returns the latest feature values recorded before each row's timestamp, or null if there are none. Any FeatureLookup against a time series table must be a point-in-time lookup. In the example below, user features are joined as-of the ad impression time, while ad features use a plain key lookup.

feature_lookups = [
  FeatureLookup(
    table_name="ml.ads_team.user_features",
    feature_names=["purchases_30d", "is_free_trial_active"],
    lookup_key="u_id",
     ? ="ad_impression_ts"
  ),
  FeatureLookup(
    table_name="ml.ads_team.ad_features",
    feature_names=["sports_relevance", "food_relevance"],
    lookup_key="ad_id",
  )
]

Two details come up in exam questions. First, point-in-time lookup does not skip null values stored in the table. If the most recent row before the timestamp holds a null, you get that null back. Second, passing use_spark_native_join=True to create_training_set speeds up lookups when Photon is enabled. This needs databricks-feature-engineering 0.6.0 or above.

Sources1

4.Building automated feature computation pipelines with the FeatureEngineeringClient

A feature table only stays useful if its values stay current. Databricks recommends a scheduled job that runs a notebook to recompute features and write them back with mode='merge'. You can turn an existing unscheduled job into a scheduled one (see Lakeflow Jobs). In a merge, rows whose primary key appears in the DataFrame are updated, and all other rows stay as they are.

from databricks.feature_engineering import FeatureEngineeringClient
fe = FeatureEngineeringClient()

customer_features_df = compute_customer_features(data)

fe.write_table(
  df=customer_features_df,
  name='ml.recommender_system.customer_features',
  mode='merge'
)

The same call can also add new features. You can extend the existing computation function, or write a new function whose output includes the table's primary keys. Either way, write_table updates the schema and merges the new values by primary key. Some metadata must never change: the primary key, the partition key, and the name or data type of an existing feature. Changing any of these breaks downstream training and serving pipelines.

With Feature Views, the FeatureEngineeringClient can run the automated feature computation pipeline itself, so you don't have to write a scheduled notebook. You declare a Feature on a DeltaTableSource, for example an AggregationFunction over a TumblingWindow or SlidingWindow. You register it in Unity Catalog with register_feature, or define and register it in one step with create_feature. Then you call fe.materialize_features with a trigger. Features must be registered in UC before you call materialize_features. Aggregation features accept a CronSchedule (a quartz cron expression plus a timezone) or a TableTrigger. They can write to an offline store (OfflineStoreConfig), an online store (OnlineStoreConfig), or both. ColumnSelection features use TableTrigger and support only the online config.

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

If you keep a hand-written write_table pipeline, automate its path to serving as well. publish_table defaults to the TRIGGERED publish mode, which incrementally pushes changes from the offline table to the online store. To run it periodically, create a scheduled Lakeflow Job that runs a notebook calling publish_table, or schedule updates to the pipeline returned by publish_table. TRIGGERED mode requires Change Data Feed on the offline table. You can unit-test the pipeline code outside Databricks by patching FeatureEngineeringClient.write_table with a mock and asserting it was called with the expected table name, DataFrame, and mode='merge'.

Sources234

5.Streaming feature computation

For fresher features, switch the batch source for a stream. Read events with spark.readStream, apply the same kind of feature transformation, and pass the resulting streaming DataFrame to write_table. It is the same method with the same merge mode. The difference is the return value: a StreamingQuery object that you can monitor or stop. Streaming writes to time series feature tables are also supported. For those tables, each write still has to supply values for every feature in the table.

customer_transactions = spark.readStream.table("prod.events.customer_transactions")
stream_df = compute_additional_customer_features(customer_transactions)

fe.write_table(
  df=stream_df,
  name='ml.recommender_system.customer_features',
  mode='merge'
)

Sources21

6.Designing scalable solutions for ingesting and processing streaming data to generate features in real time

Use streaming features when feature values must update continuously rather than on a batch schedule. Feature Views use the same Feature constructors, aggregation functions, and training and serving workflows for streaming and batch, so designing a real-time solution mostly comes down to choosing how you ingest the streaming data. That choice sets your feature freshness.

A StreamSource references a Stream by its three-part name. Column references in entity, timeseries, and function definitions must be prefixed with value. or key. to say which part of the Kafka message to read. filter_condition drops rows before aggregation, and transformation_sql applies a row-wise Spark SQL projection. transformation_sql works only with a StreamSource.

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

Processing rules for streaming materialization: you must provide an online_config, and offline_config is not supported. Streaming features do not materialize to an offline store. For training and batch inference, Databricks calculates their values from the source. create_training_set reads the Stream's ingestion table and performs point-in-time joins against your labeled DataFrame. You cannot mix streaming and batch features in one materialize_features call, so make a separate call for each trigger type. A single training set and model can still combine both: at serving time, batch features are looked up from offline or online stores, and streaming features from online stores. Streaming materialization processes only records that arrive after the pipeline starts. It does not backfill history, so rolling-window aggregates are complete only after the first full window. For classic feature tables, publish_table with publish_mode="CONTINUOUS" sets up a streaming pipeline that updates the online store immediately as new data lands in the offline table.

client.materialize_features(
    features=[feature],
    online_config=online_config,
    trigger= ? (),
)

Sources354

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Point-in-time feature lookup relies on Delta Lake time travel to read older table versions.Why is that wrong?

   The two are unrelated. Point-in-time lookup is an as-of join on primary key plus timestamp key.

   Covered in Why a naive join leaks the future

2. 2.Putting the timestamp column in primary_keys is enough to get point-in-time joins.Why is that wrong?

   Without timeseries_columns the column is matched exactly. As-of logic needs the timeseries declaration.

   Covered in Declaring a time series feature table

3. 3.Starting a streaming feature pipeline backfills the historical records in the source, so rolling-window features are complete immediately.Why is that wrong?

   Streaming materialization handles only records that arrive after it starts. Rolling windows are complete only once the first full window has arrived.

   Covered in Designing scalable solutions for ingesting and processing streaming data to generate features in real time