SageMaker Feature Store: Point-in-Time Correct Training Sets and Low-Latency Online Serving

Most production ML incidents we get called into are not model problems. They are feature problems: the training set was built with a JOIN that quietly leaked future data, or the online service computes a feature slightly differently than the notebook did, and the model that scored 0.91 offline scores 0.62 in production. SageMaker Feature Store exists to kill both of those bugs. This tutorial builds a working feature group, ingests records, generates a point-in-time correct training set, and reads features back at single-digit millisecond latency from an endpoint.

Everything here uses the SageMaker Python SDK v2 (pip install "sagemaker>=2.200") and boto3, run from a Studio space or any machine with an execution role.

The two stores, and why you need both

A feature group has an online store (a low-latency key/value store, keyed by record identifier, holding the latest value per record) and an offline store (Parquet in S3, append-only, queryable through Athena or Glue). You write once; both stores get the record.

• Training reads the offline store, filtering on event time so a row only sees feature values that existed at that moment.
• Inference reads the online store with GetRecord / BatchGetRecord.

Because both are populated by the same ingestion path, the feature definition lives in exactly one place. That is the entire point.

Step 1: Define a feature group

Two columns are mandatory: a record identifier and an event time. Event time must be a string (ISO-8601) or a fractional-seconds numeric type.

import time, pandas as pd, sagemaker
from sagemaker.feature_store.feature_group import FeatureGroup

sess = sagemaker.Session()
role = sagemaker.get_execution_role()
bucket = sess.default_bucket()

df = pd.DataFrame({
    "customer_id": ["c-1001", "c-1002", "c-1003"],
    "orders_30d": [3, 0, 11],
    "avg_basket_usd": [42.10, 0.0, 88.75],
    "days_since_signup": [412, 9, 1180],
    "event_time": [time.time()] * 3,       # float64 seconds
})
df["customer_id"] = df["customer_id"].astype("string")

fg = FeatureGroup(name="customer-behaviour-v1", sagemaker_session=sess)
fg.load_feature_definitions(data_frame=df)

fg.create(
    s3_uri=f"s3://{bucket}/feature-store/",
    record_identifier_name="customer_id",
    event_time_feature_name="event_time",
    role_arn=role,
    enable_online_store=True,
    online_store_storage_type="Standard",   # or "InMemory" for very high TPS
    table_format="Iceberg",
)

Two choices worth pausing on:

• table_format="Iceberg" is the one to pick for new feature groups. Iceberg-backed offline stores compact well and can be maintained with standard table operations; the legacy Glue format leaves you with millions of small files after a year of streaming ingestion.
• online_store_storage_type: Standard is fine for the vast majority of workloads. InMemory buys you tighter tail latency at higher cost — justify it with a measured p99 requirement, not a hunch.

Creation is asynchronous. Poll fg.describe()["FeatureGroupStatus"] until it reads Created before you ingest.

Step 2: Ingest

For batch backfills, the SDK's threaded ingest is enough:

fg.ingest(data_frame=df, max_workers=8, max_processes=2, wait=True)

For streaming, call PutRecord directly from whatever already handles the event — a Lambda on a Kinesis stream, a Flink job, your API service:

import boto3, time
fsr = boto3.client("sagemaker-featurestore-runtime")

fsr.put_record(
    FeatureGroupName="customer-behaviour-v1",
    Record=[
        {"FeatureName": "customer_id", "ValueAsString": "c-1002"},
        {"FeatureName": "orders_30d", "ValueAsString": "1"},
        {"FeatureName": "avg_basket_usd", "ValueAsString": "23.40"},
        {"FeatureName": "days_since_signup", "ValueAsString": "10"},
        {"FeatureName": "event_time", "ValueAsString": str(time.time())},
    ],
    TargetStores=["OnlineStore", "OfflineStore"],
)

TargetStores is the underrated parameter. Ephemeral features that only matter for live scoring can skip the offline store; heavy historical backfills can skip the online store so you do not blow out write costs replaying two years of events.

Offline records appear in S3 within a few minutes — it is a buffered write, not a synchronous one. Do not write tests that assert an Athena query sees a record one second after PutRecord.

Step 3: Build a point-in-time correct training set

This is the step people skip, and it is the reason offline and online metrics diverge. You have a label table: customer, timestamp of the event you want to predict, outcome. For each label row you need the feature values as of that timestamp — never later.

q = fg.athena_query()
table = q.table_name

query = f"""
WITH labels AS (
  SELECT customer_id, label_event_time, churned
  FROM   "labels_db"."churn_labels"
),
joined AS (
  SELECT l.customer_id,
         l.label_event_time,
         l.churned,
         f.orders_30d,
         f.avg_basket_usd,
         f.days_since_signup,
         ROW_NUMBER() OVER (
           PARTITION BY l.customer_id, l.label_event_time
           ORDER BY f.event_time DESC
         ) AS rn
  FROM   labels l
  JOIN   "{table}" f
    ON   f.customer_id = l.customer_id
   AND   f.event_time <= l.label_event_time
  WHERE  f.is_deleted = false
)
SELECT customer_id, label_event_time, churned,
       orders_30d, avg_basket_usd, days_since_signup
FROM   joined
WHERE  rn = 1
"""

q.run(query_string=query, output_location=f"s3://{bucket}/fs-query-results/")
q.wait()
train_df = q.as_dataframe()

The three clauses that do the work: f.event_time <= l.label_event_time prevents leakage, ROW_NUMBER() picks the most recent qualifying version of each feature, and is_deleted = false drops soft-deleted records. Miss the first one and your model will look brilliant right up until it goes live.

If your labels span a long window, add a lower bound (f.event_time >= l.label_event_time - interval '90' day) so Athena is not scanning the entire history for every row.

Step 4: Read features at inference time

The pattern that removes training/serving skew: the caller sends identifiers, not features. The serving layer fetches the feature vector from the online store and assembles the payload in the same column order the model was trained on.

import boto3, json

fsr = boto3.client("sagemaker-featurestore-runtime")
smr = boto3.client("sagemaker-runtime")

COLUMNS = ["orders_30d", "avg_basket_usd", "days_since_signup"]

def score(customer_ids):
    resp = fsr.batch_get_record(
        Identifiers=[{
            "FeatureGroupName": "customer-behaviour-v1",
            "RecordIdentifiersValueAsString": customer_ids,
            "FeatureNames": COLUMNS,
        }]
    )
    rows = []
    for rec in resp["Records"]:
        vals = {f["FeatureName"]: f["ValueAsString"] for f in rec["Record"]}
        rows.append(",".join(vals[c] for c in COLUMNS))

    out = smr.invoke_endpoint(
        EndpointName="churn-xgb-prod",
        ContentType="text/csv",
        Body="\n".join(rows).encode(),
    )
    return json.loads(out["Body"].read())

Handle resp["Errors"] and resp["UnprocessedIdentifiers"] explicitly. A cold customer with no record in the online store must hit a defined default path, not a KeyError in production. BatchGetRecord handles up to 100 identifiers per call; batch your fan-out rather than issuing one GetRecord per user.

Keep COLUMNS in one module that both the training job and the serving code import. If the order can drift, it will.

Step 5: Wire it into the pipeline and keep it clean

In a SageMaker Pipeline, the Athena query above becomes a ProcessingStep that writes train/validation splits to S3, feeding the existing training step. Nothing else about the pipeline changes.

Operational habits worth adopting on day one:

• Version in the name, not in place. customer-behaviour-v1 → customer-behaviour-v2 when semantics change. Mutating the meaning of a feature under a stable name is how you silently corrupt every model that consumes it.
• TTL the online store. Set TtlDuration on the online store config so dormant records expire instead of accruing storage cost forever.
• Compact the offline store. With Iceberg, schedule table maintenance; without it, budget for a periodic compaction job.
• Watch the write path. Alarm on PutRecord throttles and errors in CloudWatch. Silent ingestion failure looks exactly like stale features, which looks exactly like model drift — and you will spend a week debugging the wrong layer.
• Delete costs money to get wrong. DeleteRecord soft-deletes by default (a tombstone with is_deleted = true); use DeletionMode="HardDelete" when a GDPR erasure request means the row must actually leave the offline store.

Is it worth it?

Not always. One model, one batch scoring job, features computed in the same script that trains — a feature store is overhead. It pays off when two or more models share features, when the same feature must be computed for both batch training and live serving, or when you need to defend a training set's provenance to an auditor. That is when the online/offline split stops being architecture astronomy and starts being the cheapest bug prevention you can buy.

Building or untangling a feature pipeline on SageMaker AI? Our senior SageMaker consultants do this work every week — get in touch with the shape of your data and we will tell you honestly whether you need a feature store or a better JOIN.