From 80d9387a58861a2512e841986ef218688c806c94 Mon Sep 17 00:00:00 2001 From: MagicLex Date: Fri, 12 Jun 2026 14:44:32 +0300 Subject: [PATCH 1/3] fix(fraud_batch): keyword args for feature_view.log() on 5.0 5.0 prepended a `logging_data` parameter to FeatureView.log(), shifting positional args. The inference notebook passed untransformed features and predictions positionally, so they landed in the wrong slots and logging raised FeatureStoreException. Pass them by keyword. Co-Authored-By: Claude Opus 4.8 (1M context) --- batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb b/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb index fc31d386..f3e6a8f9 100644 --- a/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb +++ b/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)" ] }, From 57b0416db80aa57aeb14290c30069b202949b55c Mon Sep 17 00:00:00 2001 From: MagicLex Date: Fri, 12 Jun 2026 15:39:29 +0300 Subject: [PATCH 2/3] fix(credit_scores): applicant-grain joins for 5.0 + handle null labels/example MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 5.0 enforces that a feature view join covers every column of the joined feature group's primary key (errorCode 270249). The previous_applications, pos_cash_balances, installment_payments and credit_card_balances groups are at (sk_id_prev, sk_id_curr) grain but were joined on sk_id_curr only, so FV creation was rejected. - functions.py: add collapse_to_customer() — reduce a previous-credit table to one row per sk_id_curr (most recent by datetime), dropping sk_id_prev. - nb1: apply it before the four inserts; set their primary key to [sk_id_curr]. - nb3: drop sk_id_prev from the four select_except lists so the join is on the full primary key. Two pre-existing data-hygiene issues surfaced once the FV builds: - the bureaus-rooted left join leaves a null target for bureau records whose applicant is absent from applications; drop those rows before training. - the input_example carried pd.NA (encoded null categoricals); sample a complete row. Note: running this end-to-end also requires the hsfs builtin label_encoder null-safety fix (separate SDK change). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../1_credit_scores_feature_backfill.ipynb | 16 +++++----- .../3_credit_scores_training_pipeline.ipynb | 19 ++++++++---- batch-ai-systems/credit_scores/functions.py | 29 +++++++++++++++++++ 3 files changed, 50 insertions(+), 14 deletions(-) diff --git a/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb b/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb index d6db1e10..1669b49d 100644 --- a/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb +++ b/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))" ] }, { diff --git a/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb b/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb index b1ba4d5d..8bc5f878 100644 --- a/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb +++ b/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", diff --git a/batch-ai-systems/credit_scores/functions.py b/batch-ai-systems/credit_scores/functions.py index 199679cc..f7d298bf 100644 --- a/batch-ai-systems/credit_scores/functions.py +++ b/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. From 5bc6f26c40971d17b617def3a17f1a87ee59d075 Mon Sep 17 00:00:00 2001 From: MagicLex Date: Fri, 12 Jun 2026 15:52:02 +0300 Subject: [PATCH 3/3] fix(hospital_wait_time): run on 5.0 (column renaming, NaN, cell order) This Prophet tutorial was not runnable end to end. Fixes: - 5.0 renames transformed columns to `__` and prefixes joined event-time columns with `___`. Add restore_column_names() after train_test_split / get_batch_data so the `date` column and the per-feature Prophet regressors resolve again. - Prophet rejects NaN in regressor columns; drop rows with a missing regressor or label before fit/predict (train, test and batch). - nb2: the MAE cell referenced `forecast` before the predict cell defined it; reorder predict before MAE. - nb2: matplotlib was imported as `from matplotlib import pyplot` but the plot helpers use `plt`; import as `plt`. - input_example: sample a complete row (encoded nulls are pd.NA, not JSON serializable). Running end-to-end also requires the hsfs builtin label_encoder null-safety fix (separate SDK change). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2_training_pipeline.ipynb | 60 ++++++++++++++----- .../3_inference_pipeline.ipynb | 19 ++++++ 2 files changed, 64 insertions(+), 15 deletions(-) diff --git a/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb b/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb index 9c52242f..2058e0bf 100644 --- a/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb +++ b/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 \"__\" and\n", + "# prefixes joined event-time columns with \"___\". 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", diff --git a/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb b/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb index 846e34d0..f4b4f67c 100644 --- a/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb +++ b/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 \"__\" and\n", + "# prefixes joined event-time columns with \"___\". 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",