Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions docs/how-to-guides/feature-monitoring.md
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,7 @@ GET /monitoring/metrics/features?project=my_project&feature_view_name=driver_sta
"feature_name": "conv_rate",
"feature_type": "numeric",
"metric_date": "2025-03-26",
"max_event_timestamp": "2025-03-27T14:30:00+00:00",
"granularity": "daily",
"data_source_type": "batch",
"row_count": 15000,
Expand All @@ -242,6 +243,8 @@ GET /monitoring/metrics/features?project=my_project&feature_view_name=driver_sta
]
```

The UI **Freshness** column uses `max_event_timestamp` — `MAX(event_timestamp)` from the source — not `metric_date` (the DQM window start). Age is `now − max_event_timestamp`.

### Per-feature-view aggregates

```
Expand Down
10 changes: 9 additions & 1 deletion sdk/python/feast/infra/offline_stores/bigquery.py
Original file line number Diff line number Diff line change
Expand Up @@ -693,7 +693,7 @@ def _bq_scalar_param_type(column: str) -> str:
return "BOOL"
if column == "metric_date":
return "DATE"
if column == "computed_at":
if column in ("computed_at", "max_event_timestamp"):
return "TIMESTAMP"
if column in {
"row_count",
Expand Down Expand Up @@ -889,6 +889,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
granularity STRING NOT NULL,
data_source_type STRING NOT NULL,
computed_at TIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP,
is_baseline BOOL NOT NULL,
feature_type STRING NOT NULL,
row_count INT64,
Expand All @@ -915,6 +916,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
granularity STRING NOT NULL,
data_source_type STRING NOT NULL,
computed_at TIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP,
is_baseline BOOL NOT NULL,
total_row_count INT64,
total_features INT64,
Expand All @@ -932,6 +934,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
granularity STRING NOT NULL,
data_source_type STRING NOT NULL,
computed_at TIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP,
is_baseline BOOL NOT NULL,
total_feature_views INT64,
total_features INT64,
Expand All @@ -958,6 +961,11 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
"""
for ddl in (feature_ddl, view_ddl, service_ddl, job_ddl):
client.query(ddl).result()
for tbl in (MON_TABLE_FEATURE, MON_TABLE_FEATURE_VIEW, MON_TABLE_FEATURE_SERVICE):
client.query(
f"ALTER TABLE `{proj}.{ds}.{tbl}` "
"ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMP"
).result()


def _bq_get_monitoring_max_timestamp(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -715,6 +715,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL,
data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL,
computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP WITH TIME ZONE,
is_baseline NUMBER(1) DEFAULT 0 NOT NULL,
feature_type VARCHAR2(50) NOT NULL,
row_count NUMBER,
Expand Down Expand Up @@ -746,6 +747,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL,
data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL,
computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP WITH TIME ZONE,
is_baseline NUMBER(1) DEFAULT 0 NOT NULL,
total_row_count NUMBER,
total_features NUMBER,
Expand All @@ -768,6 +770,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL,
data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL,
computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP WITH TIME ZONE,
is_baseline NUMBER(1) DEFAULT 0 NOT NULL,
total_feature_views NUMBER,
total_features NUMBER,
Expand All @@ -779,6 +782,19 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
""",
)

_oracle_try_execute_ddl(
con,
f"ALTER TABLE {MON_TABLE_FEATURE} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)",
)
_oracle_try_execute_ddl(
con,
f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)",
)
_oracle_try_execute_ddl(
con,
f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)",
)

_oracle_try_execute_ddl(
con,
f"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -429,6 +429,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
max_event_timestamp TIMESTAMPTZ,
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
feature_type VARCHAR(50) NOT NULL,
row_count BIGINT,
Expand Down Expand Up @@ -468,6 +469,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
max_event_timestamp TIMESTAMPTZ,
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
total_row_count BIGINT,
total_features INTEGER,
Expand All @@ -487,6 +489,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
max_event_timestamp TIMESTAMPTZ,
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
total_feature_views INTEGER,
total_features INTEGER,
Expand All @@ -497,6 +500,15 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
);
""")

cur.execute(f"""
ALTER TABLE {MON_TABLE_FEATURE}
ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ;
ALTER TABLE {MON_TABLE_FEATURE_VIEW}
ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ;
ALTER TABLE {MON_TABLE_FEATURE_SERVICE}
ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ;
""")

cur.execute(f"""
CREATE TABLE IF NOT EXISTS {MON_TABLE_JOB} (
job_id VARCHAR(36) PRIMARY KEY,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -574,6 +574,18 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
)
for stmt in _SPARK_MONITORING_DDL_STATEMENTS:
spark_session.sql(stmt)
from pyspark.sql.utils import AnalysisException

for stmt in (
f"ALTER TABLE {MON_TABLE_FEATURE} ADD COLUMNS (max_event_timestamp TIMESTAMP)",
f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD COLUMNS (max_event_timestamp TIMESTAMP)",
f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD COLUMNS (max_event_timestamp TIMESTAMP)",
):
try:
spark_session.sql(stmt)
except AnalysisException:
# Column already exists on newly created tables.
pass

@staticmethod
def save_monitoring_metrics(
Expand Down Expand Up @@ -678,6 +690,7 @@ def clear_monitoring_baseline(
granularity STRING NOT NULL,
data_source_type STRING NOT NULL,
computed_at TIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP,
is_baseline BOOLEAN NOT NULL,
feature_type STRING NOT NULL,
row_count BIGINT,
Expand All @@ -703,6 +716,7 @@ def clear_monitoring_baseline(
granularity STRING NOT NULL,
data_source_type STRING NOT NULL,
computed_at TIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP,
is_baseline BOOLEAN NOT NULL,
total_row_count BIGINT,
total_features INT,
Expand All @@ -719,6 +733,7 @@ def clear_monitoring_baseline(
granularity STRING NOT NULL,
data_source_type STRING NOT NULL,
computed_at TIMESTAMP NOT NULL,
max_event_timestamp TIMESTAMP,
is_baseline BOOLEAN NOT NULL,
total_feature_views INT,
total_features INT,
Expand Down
3 changes: 2 additions & 1 deletion sdk/python/feast/infra/offline_stores/dask.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
from feast.monitoring.monitoring_utils import (
MONITORING_DIR,
MONITORING_PARQUET_FILES,
MONITORING_TIMESTAMP_FIELDS,
monitoring_parquet_meta,
normalize_monitoring_row,
opt_float,
Expand Down Expand Up @@ -968,7 +969,7 @@ def _dask_parquet_query(
for _, row in df.iterrows():
record = {c: row.get(c) for c in columns}
normalize_monitoring_row(record)
for key in ("metric_date", "computed_at"):
for key in MONITORING_TIMESTAMP_FIELDS:
val = record.get(key)
if (
val is not None
Expand Down
3 changes: 2 additions & 1 deletion sdk/python/feast/infra/offline_stores/duckdb.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
from feast.monitoring.monitoring_utils import (
MONITORING_DIR,
MONITORING_PARQUET_FILES,
MONITORING_TIMESTAMP_FIELDS,
empty_categorical_metric,
empty_numeric_metric,
monitoring_parquet_meta,
Expand Down Expand Up @@ -487,7 +488,7 @@ def _duckdb_parquet_query(
for _, row in df.iterrows():
record = {c: row.get(c) for c in columns}
normalize_monitoring_row(record)
for key in ("metric_date", "computed_at"):
for key in MONITORING_TIMESTAMP_FIELDS:
val = record.get(key)
if (
val is not None
Expand Down
6 changes: 6 additions & 0 deletions sdk/python/feast/infra/offline_stores/redshift.py
Original file line number Diff line number Diff line change
Expand Up @@ -555,6 +555,7 @@ def clear_monitoring_baseline(
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
max_event_timestamp TIMESTAMPTZ,
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
feature_type VARCHAR(50) NOT NULL,
row_count BIGINT,
Expand Down Expand Up @@ -582,6 +583,7 @@ def clear_monitoring_baseline(
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
max_event_timestamp TIMESTAMPTZ,
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
total_row_count BIGINT,
total_features INTEGER,
Expand All @@ -600,6 +602,7 @@ def clear_monitoring_baseline(
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
max_event_timestamp TIMESTAMPTZ,
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
total_feature_views INTEGER,
total_features INTEGER,
Expand All @@ -625,6 +628,9 @@ def clear_monitoring_baseline(
PRIMARY KEY (job_id)
);
""",
f"ALTER TABLE {MON_TABLE_FEATURE} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ",
f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ",
f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ",
]


Expand Down
8 changes: 8 additions & 0 deletions sdk/python/feast/infra/offline_stores/snowflake.py
Original file line number Diff line number Diff line change
Expand Up @@ -564,6 +564,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
"granularity" VARCHAR(20) NOT NULL DEFAULT 'daily',
"data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch',
"computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
"max_event_timestamp" TIMESTAMP_TZ,
"is_baseline" BOOLEAN NOT NULL DEFAULT FALSE,
"feature_type" VARCHAR(50) NOT NULL,
"row_count" BIGINT,
Expand Down Expand Up @@ -591,6 +592,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
"granularity" VARCHAR(20) NOT NULL DEFAULT 'daily',
"data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch',
"computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
"max_event_timestamp" TIMESTAMP_TZ,
"is_baseline" BOOLEAN NOT NULL DEFAULT FALSE,
"total_row_count" BIGINT,
"total_features" INTEGER,
Expand All @@ -609,6 +611,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
"granularity" VARCHAR(20) NOT NULL DEFAULT 'daily',
"data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch',
"computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
"max_event_timestamp" TIMESTAMP_TZ,
"is_baseline" BOOLEAN NOT NULL DEFAULT FALSE,
"total_feature_views" INTEGER,
"total_features" INTEGER,
Expand Down Expand Up @@ -642,6 +645,11 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
execute_snowflake_statement(conn, ddl_view)
execute_snowflake_statement(conn, ddl_service)
execute_snowflake_statement(conn, ddl_job)
for fq in (fq_feature, fq_view, fq_service):
execute_snowflake_statement(
conn,
f'ALTER TABLE {fq} ADD COLUMN IF NOT EXISTS "max_event_timestamp" TIMESTAMP_TZ',
)

@staticmethod
def save_monitoring_metrics(
Expand Down
Loading
Loading