A reference dataset is the feature dataset associated with users across training and testing populations. Store it with the model run so production comparisons remain reproducible. section 4.1.2
In drift monitoring, the reference dataset represents historical data, while the current dataset contains features about users to predict. Comparing both lets a report measure changes in feature distributions, missing values, and model quality. section 6.2.1
Use a reference dataset to:
- Compare each feature's reference distribution with its current inference distribution. section 6.2.1
- Anchor regression-quality measurements such as mean error, mean absolute error, and mean absolute percentage error. section 6.2.1
- Retrieve the exact dataset associated with a model run and compare it with newly retrieved Feast features. section 6.2.2
- Stop an inference pipeline when drift between reference and current feature data exceeds the monitoring rule. section 6.2.2
Model-run baseline for drift detection
Save the feature, training, and test datasets in MinIO, log their external paths as MLflow inputs, and later retrieve the reference dataset from the model run. Pass it with the current inference dataset to Evidently's DataDriftPreset, then publish the report to the monitoring dashboard. section 4.1.2 section 6.2.1 section 6.2.2
Listing 6.9 KFP component for detecting data drift
def detect_drift(
model_name: str,
model_stage: str,
mlflow_host: str,
minio_host: str,
access_key: str,
secret_key: str,
reference_dataset_name: str,
feature_dataset_path: str,
evidently_workspace_url: str,
evidently_ui_project_name: str,
):
os.environ["AWS_ACCESS_KEY_ID"] = access_key
os.environ["AWS_SECRET_ACCESS_KEY"] = secret_key
if not minio_host.startswith("http"):
os.environ["AWS_ENDPOINT_URL"] = "http://" + minio_host
else:
os.environ["AWS_ENDPOINT_URL"] = minio_host
mlflow.set_tracking_uri(mlflow_host)
mlflow_client = MlflowClient(mlflow_host) #1
evidently_workspace = \
RemoteWorkspace(evidently_workspace_url) #2
model_run_id = None
for model in mlflow_client.search_model_versions(f"name='{model_name}'"):
if model.current_stage == model_stage:
model_run_id = model.run_id
break
if not model_run_id:
raise ValueError(
f"Model in stage {model_stage} "
f"not found for {model_name}."
)
run = mlflow.get_run(model_run_id)
dataset_source = None
for dataset_input in run.inputs.dataset_inputs:
for tag in dataset_input.tags:
if tag.value == reference_dataset_name:
dataset_source = json.loads(
dataset_input.dataset.source
)["uri"]
break
if not dataset_source:
raise ValueError(
f"Reference dataset {reference_dataset_name} "
"not found."
)
bucket_name = dataset_source.split("/")[0]
object_name = "/".join(dataset_source.split("/")[1:])
file_path = object_name.split("/")[-1]
download_file_from_minio(
minio_host=minio_host,
access_key=access_key,
secret_key=secret_key,
bucket_name=bucket_name,
object_name=object_name,
file_path=file_path,
) #3
reference_df = pd.read_csv(file_path)
feature_df = pd.read_parquet(feature_dataset_path) #4
feature_df.drop(columns=["user_id"], errors="ignore", inplace=True)
report = Report(
metrics=[
DatasetDriftMetric(),
DatasetMissingValuesMetric(),
ColumnDriftMetric(column_name="Education"),
ColumnSummaryMetric(column_name="Education"),
ColumnDriftMetric(column_name="Marital-Status"),
ColumnSummaryMetric(column_name="Marital-Status"),
ColumnDriftMetric(column_name="Native_country"),
ColumnSummaryMetric(column_name="Native_country"),
ColumnDriftMetric(column_name="Occupation"),
ColumnSummaryMetric(column_name="Occupation"),
ColumnDriftMetric(column_name="Race"),
ColumnSummaryMetric(column_name="Race"),
ColumnDriftMetric(column_name="Relationship"),
ColumnSummaryMetric(column_name="Relationship"),
ColumnDriftMetric(column_name="Sex"),
ColumnSummaryMetric(column_name="Sex"),
ColumnDriftMetric(column_name="Workclass"),
ColumnSummaryMetric(column_name="Workclass"),
],
) #5
report.run(
reference_data=reference_df,
current_data=feature_df
) #6
report_file_path = "drift_report.json"
with open(report_file_path, "w") as f:
f.write(report.json())
print("report written")
project = get_evidently_project(
evidently_workspace,
evidently_ui_project_name
)
print("project retreived")
evidently_workspace.add_report(project.id, report) #7
project.save()
Inspect the complete pipeline component that retrieves the model run's reference dataset, reads current feature data, generates an Evidently report, and integrates drift detection into Kubeflow. section 6.2.2
