Launch notebook online Binder badge or download it Download badge
[1]:
%matplotlib inline

Using scikit-learn FedAvg on IRIS dataset

This example illustrate an advanced usage of SubstraFL as it does not use the SubstraFL PyTorch interface, but showcases the general SubstraFL interface that you can use with any ML framework.

This example is based on:

  • Dataset: IRIS, tabular dataset to classify iris type

  • Model type: Logistic regression using Scikit-Learn

  • FL setup: three organizations, two data providers and one algo provider

This example does not use the deployed platform of Substra, it runs in local mode.

To run this example, you need to download and unzip the assets needed to run it in the same directory as used this example:

Please ensure to have all the libraries installed. A requirements.txt file is included in the zip file, where you can run the command pip install -r requirements.txt to install them.

Substra and SubstraFL should already be installed. If not, follow the instructions described here.

Setup

We work with three different organizations. Two organizations provide a dataset, and a third one provides the algorithm and registers the machine learning tasks.

This example runs in local mode, simulating a federated learning experiment.

In the following code cell, we define the different organizations needed for our FL experiment.

[2]:
import numpy as np

from substra import Client

SEED = 42
np.random.seed(SEED)

# Choose the subprocess mode to locally simulate the FL process
N_CLIENTS = 3
clients_list = [Client(client_name=f"org-{i+1}") for i in range(N_CLIENTS)]
clients = {client.organization_info().organization_id: client for client in clients_list}

# Store organization IDs
ORGS_ID = list(clients)
ALGO_ORG_ID = ORGS_ID[0]  # Algo provider is defined as the first organization.
DATA_PROVIDER_ORGS_ID = ORGS_ID[1:]  # Data provider orgs are the last two organizations.
/home/docs/checkouts/readthedocs.org/user_builds/owkin-substra-documentation/conda/stable/lib/python3.10/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
  from .autonotebook import tqdm as notebook_tqdm

Data and metrics

Data preparation

This section downloads (if needed) the IRIS dataset using the Scikit-Learn dataset module. It extracts the data locally create two folders: one for each organization.

Each organization will have access to half the train data, and to half the test data.

[3]:
import pathlib
from sklearn_fedavg_assets.dataset.iris_dataset import setup_iris


# Create the temporary directory for generated data
(pathlib.Path.cwd() / "tmp").mkdir(exist_ok=True)
data_path = pathlib.Path.cwd() / "tmp" / "data_iris"

setup_iris(data_path=data_path, n_client=len(DATA_PROVIDER_ORGS_ID))

Dataset registration

[4]:
from substra.sdk.schemas import DatasetSpec
from substra.sdk.schemas import Permissions
from substra.sdk.schemas import DataSampleSpec

assets_directory = pathlib.Path.cwd() / "sklearn_fedavg_assets"

permissions_dataset = Permissions(public=False, authorized_ids=[ALGO_ORG_ID])

dataset = DatasetSpec(
    name="Iris",
    type="npy",
    data_opener=assets_directory / "dataset" / "iris_opener.py",
    description=assets_directory / "dataset" / "description.md",
    permissions=permissions_dataset,
    logs_permission=permissions_dataset,
)

dataset_keys = {}
train_datasample_keys = {}
test_datasample_keys = {}

for i, org_id in enumerate(DATA_PROVIDER_ORGS_ID):
    client = clients[org_id]

    # Add the dataset to the client to provide access to the opener in each organization.
    dataset_keys[org_id] = client.add_dataset(dataset)
    assert dataset_keys[org_id], "Missing data manager key"

    client = clients[org_id]

    # Add the training data on each organization.
    data_sample = DataSampleSpec(
        data_manager_keys=[dataset_keys[org_id]],
        path=data_path / f"org_{i+1}" / "train",
    )
    train_datasample_keys[org_id] = client.add_data_sample(
        data_sample,
        local=True,
    )

    # Add the testing data on each organization.
    data_sample = DataSampleSpec(
        data_manager_keys=[dataset_keys[org_id]],
        path=data_path / f"org_{i+1}" / "test",
    )
    test_datasample_keys[org_id] = client.add_data_sample(
        data_sample,
        local=True,
    )

Metrics registration

[5]:
from sklearn.metrics import accuracy_score
import numpy as np


def accuracy(datasamples, predictions_path):
    y_true = datasamples["targets"]
    y_pred = np.load(predictions_path)

    return accuracy_score(y_true, y_pred)

Specify the machine learning components

SubstraFL can be used with any machine learning framework. The framework dependent functions are written in the Algorithm object.

In this section, you will:

  • register a model and its dependencies

  • write your own Sklearn SubstraFL algorithm

  • specify the federated learning strategy

  • specify the organizations where to train and where to aggregate

  • specify the organizations where to test the models

  • actually run the computations

Model definition

The machine learning model used here is a logistic regression. The warm_start argument is essential in this example as it indicates to use the current state of the model as initialization for the future training. By default scikit-learn uses max_iter=100, which means the model trains on up to 100 epochs. When doing federated learning, we don’t want to train too much locally at every round otherwise the local training will erase what was learned from the other centers. That is why we set max_iter=3.

[6]:
import os
from sklearn import linear_model

cls = linear_model.LogisticRegression(random_state=SEED, warm_start=True, max_iter=3)

# Optional:
# Scikit-Learn raises warnings in case of non convergence, that we choose to disable here.
# As this example runs with python subprocess, the way to disable it is to use following environment
# variable:
os.environ["PYTHONWARNINGS"] = "ignore:lbfgs failed to converge (status=1):UserWarning"

SubstraFL algo definition

This section is the most important one for this example. We will define here the function that will run locally on each node to train the model.

As SubstraFL does not provide an algorithm comptatible with Sklearn, we need to define one using the provided documentation on substrafl_doc/api/algorithms:Base Class.

To define a custom algorithm, we will need to inherit from the base class Algo, and to define two properties and four methods:

  • strategies (property): the list of strategies our algorithm is compatible with.

  • model (property): a property that returns the model from the defined algo.

  • train (method): a function to describe the training process to apply to train our model in a federated way. The train method must accept as parameters datasamples and shared_state.

  • predict (method): a function to describe how to compute the predictions from the algo model. The predict method must accept as parameters datasamples, shared_state and predictions_path.

  • save (method): specify how to save the important states of our algo.

  • load (method): specify how to load the important states of our algo from a previously saved filed by the save function describe above.

[7]:
from substrafl import algorithms
from substrafl import remote
from substrafl.strategies import schemas as fl_schemas

import joblib
from typing import Optional
import shutil

# The Iris dataset proposes four attributes to predict three different classes.
INPUT_SIZE = 4
OUTPUT_SIZE = 3


class SklearnLogisticRegression(algorithms.Algo):
    def __init__(self, model, seed=None):
        super().__init__(model=model, seed=seed)

        self._model = model

        # We need all different instances of the algorithm to have the same
        # initialization.
        self._model.coef_ = np.ones((OUTPUT_SIZE, INPUT_SIZE))
        self._model.intercept_ = np.zeros(3)
        self._model.classes_ = np.array([-1])

        if seed is not None:
            np.random.seed(seed)

    @property
    def strategies(self):
        """List of compatible strategies"""
        return [fl_schemas.StrategyName.FEDERATED_AVERAGING]

    @property
    def model(self):
        return self._model

    @remote.remote_data
    def train(
        self,
        datasamples,
        shared_state: Optional[fl_schemas.FedAvgAveragedState] = None,
    ) -> fl_schemas.FedAvgSharedState:
        """The train function to be executed on organizations containing
        data we want to train our model on. The @remote_data decorator is mandatory
        to allow this function to be sent and executed on the right organization.

        Args:
            datasamples: datasamples extracted from the organizations data using
                the given opener.
            shared_state (Optional[fl_schemas.FedAvgAveragedState], optional):
                shared_state provided by the aggregator. Defaults to None.

        Returns:
            fl_schemas.FedAvgSharedState: State to be sent to the aggregator.
        """

        if shared_state is not None:
            # If we have a shared state, we update the model parameters with
            # the average parameters updates.
            self._model.coef_ += np.reshape(
                shared_state.avg_parameters_update[:-1],
                (OUTPUT_SIZE, INPUT_SIZE),
            )
            self._model.intercept_ += shared_state.avg_parameters_update[-1]

        # To be able to compute the delta between the parameters before and after training,
        # we need to save them in a temporary variable.
        old_coef = self._model.coef_
        old_intercept = self._model.intercept_

        # Model training.
        self._model.fit(datasamples["data"], datasamples["targets"])

        # We compute de delta.
        delta_coef = self._model.coef_ - old_coef
        delta_bias = self._model.intercept_ - old_intercept

        # We reset the model parameters to their state before training in order to remove
        # the local updates from it.
        self._model.coef_ = old_coef
        self._model.intercept_ = old_intercept

        # We output the length of the dataset to apply a weighted average between
        # the organizations regarding their number of samples, and the local
        # parameters updates.
        # These updates are sent to the aggregator to compute the average
        # parameters updates, that we will receive in the next round in the
        # `shared_state`.
        return fl_schemas.FedAvgSharedState(
            n_samples=len(datasamples["targets"]),
            parameters_update=[p for p in delta_coef] + [delta_bias],
        )

    @remote.remote_data
    def predict(self, datasamples, shared_state, predictions_path):
        """The predict function to be executed on organizations containing
        data we want to test our model on. The @remote_data decorator is mandatory
        to allow this function to be sent and executed on the right organization.

        Args:
            datasamples: datasamples extracted from the organizations data using
                the given opener.
            shared_state: shared_state provided by the aggregator.
            predictions_path: Path where to save the predictions.
                This path is provided by Substra and the metric will automatically
                get access to this path to load the predictions.
        """
        predictions = self._model.predict(datasamples["data"])

        if predictions_path is not None:
            np.save(predictions_path, predictions)

            # np.save() automatically adds a ".npy" to the end of the file.
            # We rename the file produced by removing the ".npy" suffix, to make sure that
            # predictions_path is the actual file name.
            shutil.move(str(predictions_path) + ".npy", predictions_path)

    def save_local_state(self, path):
        joblib.dump(
            {
                "model": self._model,
                "coef": self._model.coef_,
                "bias": self._model.intercept_,
            },
            path,
        )

    def load_local_state(self, path):
        loaded_dict = joblib.load(path)
        self._model = loaded_dict["model"]
        self._model.coef_ = loaded_dict["coef"]
        self._model.intercept_ = loaded_dict["bias"]
        return self

Federated Learning strategies

[8]:
from substrafl.strategies import FedAvg

strategy = FedAvg(algo=SklearnLogisticRegression(model=cls, seed=SEED))

Where to train where to aggregate

[9]:
from substrafl.nodes import TrainDataNode
from substrafl.nodes import AggregationNode


aggregation_node = AggregationNode(ALGO_ORG_ID)

# Create the Train Data Nodes (or training tasks) and save them in a list
train_data_nodes = [
    TrainDataNode(
        organization_id=org_id,
        data_manager_key=dataset_keys[org_id],
        data_sample_keys=[train_datasample_keys[org_id]],
    )
    for org_id in DATA_PROVIDER_ORGS_ID
]

Where and when to test

[10]:
from substrafl.nodes import TestDataNode
from substrafl.evaluation_strategy import EvaluationStrategy

# Create the Test Data Nodes (or testing tasks) and save them in a list
test_data_nodes = [
    TestDataNode(
        organization_id=org_id,
        data_manager_key=dataset_keys[org_id],
        test_data_sample_keys=[test_datasample_keys[org_id]],
        metric_functions=accuracy,
    )
    for org_id in DATA_PROVIDER_ORGS_ID
]

my_eval_strategy = EvaluationStrategy(test_data_nodes=test_data_nodes, eval_frequency=1)

Running the experiment

[11]:
from substrafl.experiment import execute_experiment
from substrafl.dependency import Dependency

# Number of times to apply the compute plan.
NUM_ROUNDS = 6

dependencies = Dependency(pypi_dependencies=["numpy==1.24.3", "scikit-learn==1.3.1"])

compute_plan = execute_experiment(
    client=clients[ALGO_ORG_ID],
    strategy=strategy,
    train_data_nodes=train_data_nodes,
    evaluation_strategy=my_eval_strategy,
    aggregation_node=aggregation_node,
    num_rounds=NUM_ROUNDS,
    experiment_folder=str(pathlib.Path.cwd() / "tmp" / "experiment_summaries"),
    dependencies=dependencies,
    name="IRIS documentation example",
)
2023-10-20 10:01:01,695 - INFO - Building the compute plan.
2023-10-20 10:01:01,700 - INFO - Registering the functions to Substra.
2023-10-20 10:01:01,742 - INFO - Registering the compute plan to Substra.
2023-10-20 10:01:01,743 - INFO - Experiment summary saved to /home/docs/checkouts/readthedocs.org/user_builds/owkin-substra-documentation/checkouts/stable/docs/source/examples/substrafl/go_further/tmp/experiment_summaries/2023_10_20_10_01_01_ccaa8265-2d89-47cd-b382-f9d417e2db39.json
Compute plan progress:   0%|          | 0/50 [00:00<?, ?it/s]/home/docs/checkouts/readthedocs.org/user_builds/owkin-substra-documentation/checkouts/stable/docs/src/substra/substra/sdk/backends/local/backend.py:613: UserWarning: `transient=True` is ignored in local mode
  warnings.warn("`transient=True` is ignored in local mode", stacklevel=1)
Compute plan progress: 100%|██████████| 50/50 [00:54<00:00,  1.09s/it]
2023-10-20 10:01:56,046 - INFO - The compute plan has been registered to Substra, its key is ccaa8265-2d89-47cd-b382-f9d417e2db39.

Explore the results

[12]:
# The results will be available once the compute plan is completed
clients[ALGO_ORG_ID].wait_compute_plan(compute_plan.key)
[12]:
{
    "key": "ccaa8265-2d89-47cd-b382-f9d417e2db39",
    "tag": "",
    "name": "IRIS documentation example",
    "owner": "MyOrg1MSP",
    "metadata": {
        "substrafl_version": "0.42.0",
        "substra_version": "0.49.0",
        "substratools_version": "0.21.0",
        "python_version": "3.10.13"
    },
    "task_count": 50,
    "waiting_count": 0,
    "todo_count": 0,
    "doing_count": 0,
    "canceled_count": 0,
    "failed_count": 0,
    "done_count": 50,
    "failed_task_key": null,
    "status": "PLAN_STATUS_DONE",
    "creation_date": "2023-10-20T10:01:01.746050",
    "start_date": "2023-10-20T10:01:01.746052",
    "end_date": "2023-10-20T10:01:56.039821",
    "estimated_end_date": "2023-10-20T10:01:56.039821",
    "duration": 54,
    "creator": null
}

Listing results

[13]:
import pandas as pd

performances_df = pd.DataFrame(client.get_performances(compute_plan.key).dict())
print("\nPerformance Table: \n")
print(performances_df[["worker", "round_idx", "performance"]])

Performance Table:

       worker  round_idx  performance
0   MyOrg2MSP          0     0.000000
1   MyOrg3MSP          0     0.000000
2   MyOrg2MSP          1     0.933333
3   MyOrg3MSP          1     1.000000
4   MyOrg2MSP          2     0.933333
5   MyOrg3MSP          2     1.000000
6   MyOrg2MSP          3     0.933333
7   MyOrg3MSP          3     1.000000
8   MyOrg2MSP          4     0.933333
9   MyOrg3MSP          4     1.000000
10  MyOrg2MSP          5     1.000000
11  MyOrg3MSP          5     1.000000
12  MyOrg2MSP          6     1.000000
13  MyOrg3MSP          6     1.000000
/tmp/ipykernel_3631/1974767673.py:3: PydanticDeprecatedSince20: The `dict` method is deprecated; use `model_dump` instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.4/migration/
  performances_df = pd.DataFrame(client.get_performances(compute_plan.key).dict())

Plot results

[14]:
import matplotlib.pyplot as plt

plt.title("Test dataset results")
plt.xlabel("Rounds")
plt.ylabel("Accuracy")

for org_id in DATA_PROVIDER_ORGS_ID:
    df = performances_df[performances_df["worker"] == org_id]
    plt.plot(df["round_idx"], df["performance"], label=org_id)

plt.legend(loc="lower right")
plt.show()
../../../_images/examples_substrafl_go_further_run_iris_sklearn_28_0.png

Download a model

[15]:
from substrafl.model_loading import download_algo_state

client_to_download_from = DATA_PROVIDER_ORGS_ID[0]
round_idx = None

algo = download_algo_state(
    client=clients[client_to_download_from],
    compute_plan_key=compute_plan.key,
    round_idx=round_idx,
)

cls = algo.model

print("Coefs: ", cls.coef_)
print("Intercepts: ", cls.intercept_)
Coefs:  [[ 1.16237637  1.80062789 -0.59844895  0.16076327]
 [ 1.02009926  0.51773141  1.04883079  0.61084198]
 [ 0.26141703  0.12553336  1.99351081  1.67228741]]
Intercepts:  [ 0.21601049  0.1066958  -0.32270629]
Launch notebook online Binder badge or download it Download badge