Verify & fix tutorials against Hopsworks 5.0

batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb

@@ -503,11 +503,11 @@
     "previous_applications_fg = fs.get_or_create_feature_group(\n",
     "    name='previous_applications',\n",
     "    version=1,\n",
-    "    primary_key=['sk_id_prev','sk_id_curr'],\n",
+    "    primary_key=['sk_id_curr'],\n",
     "    online_enabled=False,\n",
     "    event_time='datetime',\n",
     ")\n",
-    "previous_applications_fg.insert(previous_applications_df)"
+    "previous_applications_fg.insert(collapse_to_customer(previous_applications_df))"
    ]
   },
   {
@@ -528,11 +528,11 @@
     "pos_cash_balances_fg = fs.get_or_create_feature_group(\n",
     "    name='pos_cash_balances',\n",
     "    version=1,\n",
-    "    primary_key=['sk_id_prev','sk_id_curr'],\n",
+    "    primary_key=['sk_id_curr'],\n",
     "    online_enabled=False,\n",
     ")\n",
     "pos_cash_balances_fg.insert(\n",
-    "    pos_cash_balances_df,\n",
+    "    collapse_to_customer(pos_cash_balances_df),\n",
     "    write_options={\"wait_for_job\": True},\n",
     ")"
    ]
@@ -555,11 +555,11 @@
     "installment_payments_fg = fs.get_or_create_feature_group(\n",
     "    name='installment_payments',\n",
     "    version=1,\n",
-    "    primary_key=['sk_id_prev','sk_id_curr'],\n",
+    "    primary_key=['sk_id_curr'],\n",
     "    online_enabled=False,\n",
     "    event_time='datetime',\n",
     ")\n",
-    "installment_payments_fg.insert(installment_payments_df)"
+    "installment_payments_fg.insert(collapse_to_customer(installment_payments_df))"
    ]
   },
   {
@@ -580,10 +580,10 @@
     "credit_card_balances_fg = fs.get_or_create_feature_group(\n",
     "    name='credit_card_balances',\n",
     "    version=1,\n",
-    "    primary_key=['sk_id_prev','sk_id_curr'],\n",
+    "    primary_key=['sk_id_curr'],\n",
     "    online_enabled=False,\n",
     ")\n",
-    "credit_card_balances_fg.insert(credit_card_balances_df)"
+    "credit_card_balances_fg.insert(collapse_to_customer(credit_card_balances_df))"
    ]
   },
   {

batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb

@@ -225,12 +225,12 @@
     "                                                 'amt_credit', 'weekday_appr_process_start',\n",
     "                                                 'hour_appr_process_start']))\\\n",
     "            .join(bureau_balances_fg.select_except(['sk_id_bureau','months_balance']))\\\n",
-    "            .join(previous_applications_fg.select_except(['sk_id_prev', 'sk_id_curr','datetime',\n",
+    "            .join(previous_applications_fg.select_except(['sk_id_curr','datetime',\n",
     "                                                          'name_contract_type', 'name_contract_status']))\\\n",
-    "            .join(pos_cash_balances_fg.select_except(['sk_id_prev','sk_id_curr', 'months_balance',\n",
+    "            .join(pos_cash_balances_fg.select_except(['sk_id_curr', 'months_balance',\n",
     "                                                      'name_contract_status', 'sk_dpd', 'sk_dpd_def']))\\\n",
-    "            .join(installment_payments_fg.select_except(['sk_id_prev', 'sk_id_curr', 'datetime']))\\\n",
-    "            .join(credit_card_balances_fg.select_except(['sk_id_prev', 'sk_id_curr']))\\\n",
+    "            .join(installment_payments_fg.select_except(['sk_id_curr', 'datetime']))\\\n",
+    "            .join(credit_card_balances_fg.select_except(['sk_id_curr']))\\\n",
     "            .join(previous_loan_counts_fg.select_except('sk_id_curr'))\n",
     "\n",
     "selected_features_show5 = selected_features.show(5)\n",
@@ -387,7 +387,14 @@
    "source": [
     "X_train, X_test, y_train, y_test = feature_view.train_test_split(\n",
     "    test_size=0.2,\n",
-    ")"
+    ")",
+    "\n",
+    "# Drop rows with a missing label: bureau records for applicants not present in\n",
+    "# the labeled `applications` feature group yield a null `target` after the join.\n",
+    "train_valid = y_train[\"target\"].notna().to_numpy()\n",
+    "X_train, y_train = X_train[train_valid], y_train[train_valid]\n",
+    "test_valid = y_test[\"target\"].notna().to_numpy()\n",
+    "X_test, y_test = X_test[test_valid], y_test[test_valid]"
    ]
   },
   {
@@ -593,7 +600,7 @@
     "    name=\"credit_scores_model\",\n",
     "    metrics={\"f1_score\": score}, \n",
     "    description=\"XGB for Credit Scores Project\",\n",
-    "    input_example=X_train.sample(),\n",
+    "    input_example=X_train.dropna().sample(),\n",
     "    feature_view=feature_view,\n",
     ")\n",
     "\n",

batch-ai-systems/credit_scores/functions.py

@@ -2,6 +2,35 @@
 import numpy as np
 import matplotlib.axes._axes as axes
 
+def collapse_to_customer(df: pd.DataFrame) -> pd.DataFrame:
+    '''
+    Collapse a feature group that has one row per previous credit
+    (grain sk_id_prev, sk_id_curr) down to one row per applicant (sk_id_curr),
+    keeping the most recent record per applicant.
+
+    Hopsworks 5.0 requires a feature view join to cover every column of the
+    joined feature group's primary key. The previous_* tables are at
+    (sk_id_prev, sk_id_curr) grain while the training set is at sk_id_curr
+    grain, so they are reduced to sk_id_curr grain here before joining.
+
+    Args:
+    -----
+    df: pd.DataFrame
+        DataFrame at (sk_id_prev, sk_id_curr) grain.
+
+    Returns:
+    --------
+    pd.DataFrame
+        One row per sk_id_curr (most recent by datetime), without sk_id_prev.
+    '''
+    if 'datetime' in df.columns:
+        df = df.sort_values('datetime')
+    df = df.groupby('sk_id_curr', as_index=False).last()
+    if 'sk_id_prev' in df.columns:
+        df = df.drop(columns=['sk_id_prev'])
+    return df
+
+
 def remove_nans(df: pd.DataFrame) -> pd.DataFrame:
     '''
     Function which removes missing values.

batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb

@@ -168,7 +168,7 @@
    "outputs": [],
    "source": [
     "# log both transformed and untransformed features\n",
-    "feature_view.log(untransformed_batch_data.head(1000), predictions[:1000], training_dataset_version=1, model=retrieved_model)\n",
+    "feature_view.log(untransformed_features=untransformed_batch_data.head(1000), predictions=predictions[:1000], training_dataset_version=1, model=retrieved_model)\n",
     "feature_view.log(transformed_features=transformed_batch_data.head(1000), predictions=predictions[:1000], training_dataset_version=1, model=retrieved_model)"
    ]
   },

batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb

@@ -20,7 +20,7 @@
     "import datetime\n",
     "import pandas as pd\n",
     "import numpy as np\n",
-    "from matplotlib import pyplot\n",
+    "import matplotlib.pyplot as plt\n",
     "\n",
     "from sklearn.metrics import mean_absolute_error\n",
     "from prophet import Prophet\n",
@@ -269,6 +269,36 @@
     "    test_end=split_dict['test_end'],  \n",
     "    event_time=True,\n",
     ")\n",
+    "\n",
+    "# Hopsworks 5.0 renames transformed columns to \"<transform>_<feature>_\" and\n",
+    "# prefixes joined event-time columns with \"<project>_<fg>_<version>_\". Restore the\n",
+    "# plain names this notebook references (the `date` column and the per-feature\n",
+    "# Prophet regressors).\n",
+    "import re\n",
+    "def restore_column_names(df):\n",
+    "    renamed = {}\n",
+    "    for col in df.columns:\n",
+    "        match = re.match(r\"(?:label_encoder|standard_scaler|min_max_scaler|robust_scaler)_(.+)_$\", col)\n",
+    "        if match:\n",
+    "            renamed[col] = match.group(1)\n",
+    "        elif col.endswith(\"patient_info_1_date\"):\n",
+    "            renamed[col] = \"date\"\n",
+    "    df = df.rename(columns=renamed)\n",
+    "    # drop the duplicate event-time columns coming from the joined feature groups\n",
+    "    return df.drop(columns=[c for c in df.columns if c.endswith(\"_date\")], errors=\"ignore\")\n",
+    "\n",
+    "X_train = restore_column_names(X_train)\n",
+    "X_test = restore_column_names(X_test)\n",
+    "\n",
+    "\n",
+    "# Prophet rejects NaN in regressor columns; drop rows with a missing regressor or label.\n",
+    "_regressors = ['age_at_list_registration', 'cpra', 'hla_a1', 'hla_a2', 'hla_b1', 'hla_b2', 'hla_dr1', 'hla_dr2']\n",
+    "def _drop_nan_rows(X, y):\n",
+    "    keep = X[_regressors].notna().all(axis=1) & y['duration'].notna()\n",
+    "    return X[keep], y[keep]\n",
+    "X_train, y_train = _drop_nan_rows(X_train, y_train)\n",
+    "X_test, y_test = _drop_nan_rows(X_test, y_test)\n",
+    "\n",
     "X_train.head(3)"
    ]
   },
@@ -372,32 +402,32 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "b580f954-98be-4199-888a-e6bf3d194216",
+   "id": "ce527621",
    "metadata": {},
    "outputs": [],
    "source": [
-    "# Calculate MAE between expected and predicted values for december\n",
-    "y_pred = forecast['yhat']\n",
-    "mae = mean_absolute_error(y_test, y_pred)\n",
-    "print('⛳️ MAE: %.3f' % mae)\n",
+    "forecast = model.predict(X_test)\n",
     "\n",
-    "metrics = {\n",
-    "    \"mae\": round(mae,2)\n",
-    "}\n",
-    "metrics"
+    "# Summarize the forecast\n",
+    "print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].head())"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "ce527621",
+   "id": "b580f954-98be-4199-888a-e6bf3d194216",
    "metadata": {},
    "outputs": [],
    "source": [
-    "forecast = model.predict(X_test)\n",
+    "# Calculate MAE between expected and predicted values for december\n",
+    "y_pred = forecast['yhat']\n",
+    "mae = mean_absolute_error(y_test, y_pred)\n",
+    "print('⛳️ MAE: %.3f' % mae)\n",
     "\n",
-    "# Summarize the forecast\n",
-    "print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].head())"
+    "metrics = {\n",
+    "    \"mae\": round(mae,2)\n",
+    "}\n",
+    "metrics"
    ]
   },
   {
@@ -524,7 +554,7 @@
     "    name=\"waiting_time_forecast_model\",   # Name for the model\n",
     "    description=\"Waiting time for a deceased donor kidney transplant forecasting model\",  # Description of the model\n",
     "    metrics=metrics,                      # Metrics used for evaluation\n",
-    "    input_example=X_test.sample(),        # Example input data for reference\n",
+    "    input_example=X_test.dropna().sample(),        # Example input data for reference\n",
     "    feature_view=feature_view,\n",
     "    \n",
     ")\n",

batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb

@@ -127,6 +127,25 @@
     "    event_time=True,\n",
     ")\n",
     "\n",
+    "# Hopsworks 5.0 renames transformed columns to \"<transform>_<feature>_\" and\n",
+    "# prefixes joined event-time columns with \"<project>_<fg>_<version>_\". Restore the\n",
+    "# plain names the model expects and drop rows with a missing regressor.\n",
+    "import re\n",
+    "def restore_column_names(df):\n",
+    "    renamed = {}\n",
+    "    for col in df.columns:\n",
+    "        match = re.match(r\"(?:label_encoder|standard_scaler|min_max_scaler|robust_scaler)_(.+)_$\", col)\n",
+    "        if match:\n",
+    "            renamed[col] = match.group(1)\n",
+    "        elif col.endswith(\"patient_info_1_date\"):\n",
+    "            renamed[col] = \"date\"\n",
+    "    df = df.rename(columns=renamed)\n",
+    "    return df.drop(columns=[c for c in df.columns if c.endswith(\"_date\")], errors=\"ignore\")\n",
+    "\n",
+    "batch_data = restore_column_names(batch_data)\n",
+    "_regressors = ['age_at_list_registration', 'cpra', 'hla_a1', 'hla_a2', 'hla_b1', 'hla_b2', 'hla_dr1', 'hla_dr2']\n",
+    "batch_data = batch_data[batch_data[_regressors].notna().all(axis=1)]\n",
+    "\n",
     "batch_data['ds'] = batch_data.date\n",
     "batch_data['ds'] = pd.to_datetime(batch_data.ds)\n",
     "batch_data['ds'] = batch_data.ds.map(lambda x: x.replace(tzinfo=None))\n",