prev_model["learner"]["gradient_booster"]["model"]["iteration_indptr"].append( KeyError: 'iteration_indptr

[AvrAmit56741]

Python version (python3 -V)

Python 3.10.6

NVFlare version (python3 -m pip list | grep "nvflare")

2.4.1

NVFlare branch (if running examples, please use the branch that corresponds to the NVFlare version, git branch)

No response

Operating system

Ubuntu22

Have you successfully run any of the following examples?

• hello-numpy-sag with simulator
• hello-pt with simulator
• hello-numpy-sag with POC
• hello-pt with POC

Please describe your question

I am currently implementing a federated learning project using XGBoost within a Dockerized environment, as outlined in your Containerized Deployment with Docker guide. The setup consists of one Ubuntu server and two Docker clients, both derived from the same Dockerfile mentioned in the guide. I run the server using the start.sh script which located at server1/startup directory

both docker clients are running on windows 11 if it matters (the server is crash).
The clients successfully connect to the server, and I am able to initiate tasks without any issues. However, I encounter a server crash after completing the first round of tasks. The specific error message logged is as follows:

2024-06-25 17:33:13,092 - ScatterAndGather - ERROR - [identity=example_project, run=1d1dfb11-6be3-41c8-84f2-cee3d2d1b2f2, wf=scatter_and_gather]: Exception in ScatterAndGather control_flow: KeyError: 'iteration_indptr'
2024-06-25 17:33:13,103 - ScatterAndGather - ERROR - Traceback (most recent call last):
File "/home/amit/Desktop/NVFlare/venv/lib/python3.10/site-packages/nvflare/app_common/workflows/scatter_and_gather.py", line 275, in control_flow
self._global_weights = self.shareable_gen.shareable_to_learnable(aggr_result, fl_ctx)
File "/home/amit/Desktop/NVFlare/venv/lib/python3.10/site-packages/nvflare/app_opt/xgboost/tree_based/shareable_generator.py", line 136, in shareable_to_learnable
model = update_model(model, update)
File "/home/amit/Desktop/NVFlare/venv/lib/python3.10/site-packages/nvflare/app_opt/xgboost/tree_based/shareable_generator.py", line 57, in update_model
prev_model["learner"]["gradient_booster"]["model"]["iteration_indptr"].append(
KeyError: 'iteration_indptr'
2024-06-25 17:33:13,105 - ServerRunner - ERROR - [identity=example_project, run=1d1dfb11-6be3-41c8-84f2-cee3d2d1b2f2, wf=scatter_and_gather]: Aborting current RUN due to FATAL_SYSTEM_ERROR received: Exception in ScatterAndGather control_flow: KeyError: 'iteration_indptr'

Both clients have uninterrupted access to the necessary data, and there are no errors reported on the client side prior to the crash.

(venv) amit@amit-VMware-Virtual-Platform:/tmp/nvflare/jobs/xgboost/app/config$ cat config_fed_server.conf
{
format_version = 2
task_data_filters = []
task_result_filters = []
workflows = [
{
id = "scatter_and_gather"
name = "ScatterAndGather"
args {
min_clients = 2
num_rounds = 100
start_round = 0
wait_time_after_min_received = 0
aggregator_id = "aggregator"
persistor_id = "persistor"
shareable_generator_id = "shareable_generator"
train_task_name = "train"
train_timeout = 0
allow_empty_global_weights = true
}
}
]
components = [
{
id = "persistor"
path = "nvflare.app_opt.xgboost.tree_based.model_persistor.XGBModelPersistor"
args {
save_name = "xgboost_model.json"
}
}
{
id = "shareable_generator"
path = "nvflare.app_opt.xgboost.tree_based.shareable_generator.XGBModelShareableGenerator"
args {}
}
{
id = "aggregator"
path = "nvflare.app_opt.xgboost.tree_based.bagging_aggregator.XGBBaggingAggregator"
args {}
}
]
}

Help!
Thanks.

[YuanTingHsieh]

@AvrAmit56741 have you just tried running with our example job? or you have modified the code?
Have you tried running hello-numpy-sag first?

[AvrAmit56741]

I didn't modify any of the core code (for example did not edit shareable_generator.py).
I created the job based on the following xgboost_fl.py which I did edit and basically added more metrics and changed the evaluation method to rmse instead of roc-auc. Note that the job based on xgboost_fl.py worked in the simulator mode.
Command to creating the job: nvflare job create -j /tmp/nvflare/jobs/xgboost -force -w xgboost_tree -sd code -f config_fed_client.conf app_script="xgboost_fl.py" app_config="--data_root_dir /tmp/nvflare/dataset/output“.
xgboost_fl.py:

# Copyright (c) 2023, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import csv
import json
from typing import Dict, List, Tuple

import pandas as pd
import xgboost as xgb
import numpy as np
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_squared_log_error
from sklearn.metrics import mean_squared_error



# (1) import nvflare client API
from nvflare import client as flare
from nvflare.app_opt.xgboost.tree_based.shareable_generator import update_model


def to_dataset_tuple(data: dict):
    dataset_tuples = {}
    for dataset_name, dataset in data.items():
        dataset_tuples[dataset_name] = _to_data_tuple(dataset)
    return dataset_tuples


def _to_data_tuple(data):
    data_num = data.shape[0]
    # split to feature and label
    x = data.iloc[:, 1:]
    y = data.iloc[:, 0]
    return x.to_numpy(), y.to_numpy(), data_num


def load_features(feature_data_path: str) -> List:
    try:
        features = []
        with open(feature_data_path, "r") as file:
            # Create a CSV reader object
            csv_reader = csv.reader(file)
            line_list = next(csv_reader)
            features = line_list
        return features
    except Exception as e:
        raise Exception(f"Load header for path'{feature_data_path} failed! {e}")


def load_data(
    data_path: str, data_features: List, random_state: int, test_size: float, skip_rows=None
) -> Dict[str, pd.DataFrame]:
    try:
        df: pd.DataFrame = pd.read_csv(
            data_path, names=data_features, sep=r"\s*,\s*", engine="python", na_values="?", skiprows=skip_rows
        )

        train, test = train_test_split(df, test_size=test_size, random_state=random_state)

        return {"train": train, "test": test}

    except Exception as e:
        raise Exception(f"Load data for path '{data_path}' failed! {e}")


def transform_data(data: Dict[str, Tuple]) -> Dict[str, Tuple]:
    # Standardize features by removing the mean and scaling to unit variance
    scaler = StandardScaler()
    scaled_datasets = {}
    for dataset_name, (x_data, y_data, data_num) in data.items():
        x_scaled = scaler.fit_transform(x_data)
        scaled_datasets[dataset_name] = (x_scaled, y_data, data_num)
    return scaled_datasets


def main():
    parser = define_args_parser()
    args = parser.parse_args()
    data_root_dir = args.data_root_dir
    random_state = args.random_state
    test_size = args.test_size
    skip_rows = args.skip_rows
    num_client_bagging = args.num_client_bagging

    # (2) initializes NVFlare client API
    flare.init()

    site_name = flare.get_site_name()
    feature_data_path = f"{data_root_dir}/{site_name}_header.csv"
    features = load_features(feature_data_path)
    n_features = len(features) - 1  # remove label

    data_path = f"{data_root_dir}/{site_name}.csv"
    data = load_data(
        data_path=data_path, data_features=features, random_state=random_state, test_size=test_size, skip_rows=skip_rows
    )

    data = to_dataset_tuple(data)
    dataset = transform_data(data)
    x_train, y_train, train_size = dataset["train"]
    x_test, y_test, test_size = dataset["test"]

    # convert to xgboost data matrix
    dmat_train = xgb.DMatrix(x_train, label=y_train)
    dmat_test = xgb.DMatrix(x_test, label=y_test)

    xgb_params = {
        "eta": 0.1067 / num_client_bagging,
        "objective": "reg:squarederror",
        "max_depth": 5,
        "gamma": 1.513,
        "min_child_weight": 4.961,
        "subsample": 0.7981,
        "colsample_bytree": 0.9234,
        "eval_metric": "rmse",
        "nthread": 16,
        "n_estimators": 210,
        "learning_rate": 0.1067,
        "num_parallel_tree": 1,
        "tree_method": "hist",

    }
    rmse = 0.5
    global_model_as_dict = None
    while flare.is_running():
        # (3) receives FLModel from NVFlare
        input_model = flare.receive()
        global_params = input_model.params
        curr_round = input_model.current_round

        print(f"current_round={curr_round}")
        if curr_round == 0:
            # (4) first round, no global model
            model = xgb.train(
                xgb_params,
                dmat_train,
                num_boost_round=1,
                evals=[(dmat_train, "train"), (dmat_test, "test")],
            )
            config = model.save_config()
        else:
            # (5) update model based on global updates
            model_updates = global_params["model_data"]
            for update in model_updates:
                global_model_as_dict = update_model(global_model_as_dict, json.loads(update))
            loadable_model = bytearray(json.dumps(global_model_as_dict), "utf-8")
            # load model
            model.load_model(loadable_model)
            model.load_config(config)

            # (6) evaluate model
            rmse = evaluate_model(x_test, model, y_test)
            mape = evaluate_model_mape(x_test, model, y_test)
            msle = evaluate_model_msle(x_test, model, y_test)
            # Print the results
            print(f"{site_name}: global model mse: {rmse:.5f}")
            print(f"{site_name}: global model mape: {mape:.5f}")
            print(f"{site_name}: global model msle: {msle:.5f}")

            # train model in two steps
            # first, eval on train and test
            eval_results = model.eval_set(
                evals=[(dmat_train, "train"), (dmat_test, "test")], iteration=model.num_boosted_rounds() - 1
            )
            print(eval_results)
            # second, train for one round
            model.update(dmat_train, model.num_boosted_rounds())

        # (7) construct trained FL model
        # Extract newly added tree using xgboost_bagging slicing api
        bst_new = model[model.num_boosted_rounds() - 1 : model.num_boosted_rounds()]
        local_model_update = bst_new.save_raw("json")
        params = {"model_data": local_model_update}
        metrics = {"accuracy": rmse}
        output_model = flare.FLModel(params=params, metrics=metrics)

        # (8) send model back to NVFlare
        flare.send(output_model)


def evaluate_model(x_test, model, y_test):
    dtest = xgb.DMatrix(x_test)
    y_pred = model.predict(dtest)
    mse = mean_squared_error(y_test, y_pred)  # Using MSE instead of AUC
    rmse = np.sqrt(mse)
    return rmse


def evaluate_model_mape(x_test, model, y_test):
    dtest = xgb.DMatrix(x_test)
    y_pred = model.predict(dtest)
    mape = np.mean(np.abs((y_test - y_pred) / y_test)) * 100
    return mape

def evaluate_model_msle(x_test, model, y_test):
    dtest = xgb.DMatrix(x_test)
    y_pred = model.predict(dtest)
    msle = mean_squared_log_error(y_test, y_pred)
    return msle


#def evaluate_model(x_test, model, y_test):
#    # Make predictions on the testing set
#    dtest = xgb.DMatrix(x_test)
#    y_pred = model.predict(dtest)

#    # Evaluate the model
#    auc = roc_auc_score(y_test, y_pred)
#    return auc


def define_args_parser():
    parser = argparse.ArgumentParser(description="scikit learn linear model with SGD")
    parser.add_argument("--data_root_dir", type=str, help="root directory path to csv data file")
    parser.add_argument("--random_state", type=int, default=0, help="random state")
    parser.add_argument("--test_size", type=float, default=0.2, help="test ratio, default to 20%")
    parser.add_argument(
        "--num_client_bagging",
        type=int,
        default=3,
        help="number of clients with uniform data sizes participating in bagging",
    )
    parser.add_argument(
        "--skip_rows",
        type=str,
        default=None,
        help="""If skip_rows = N, the first N rows will be skipped, 
       if skiprows=[0, 1, 4], the rows will be skip by row indices such as row 0,1,4 will be skipped. """,
    )
    return parser


if __name__ == "__main__":
    main()

[avramit5674]

That is the command I've executed in order to run the docker clients on windows 11:
docker run --rm -it --ipc=host --ulimit memlock=-1 --ulimit stack=67108864 -v C:\build\my-workspace:/workspace/my-workspace -w /workspace/my-workspace nvflare-pt-example:latest
Note that the flag "--gpus all" is missing.

[chesterxgchen]

thanks @avramit5674 we will take a look

[nvkevlu]

@AvrAmit56741 I consulted a researcher working with xgboost here and the error causing the crash, KeyError: 'iteration_indptr', is likely due to differences in the expected xgboost lib version and model format.

The model representation of xgboost has updates across versions, and some keys will be different. We will add to the documentation to specify what versions are supported.