多任务学习中的数据分布问题(二)

多任务学习中的数据分布(2)

在上一篇博文[《多任务学习中的数据分布问题(一)》](https://www.cnblogs.com/Orion-Orion/p/15621953.html)(链接：https://www . cn blogs.com/Orion-Orion/p/15621953 . html)中，我们提到论文[1]在联邦学习的背景下引入了多任务学习。使用的方法是使每个客户端/Synthetic节点的训练数据分布不同，这样每个任务节点就可以学习不同的模型。然后，按照病态非独立同分布划分数据有两种方式(其实最开始是论文[2]提出的方式，传入命令行参数` ` ` `` args.pathological _ non _ iid _ split=true ` ` ` ` ` ` `)时，按照标签划分数据(如果没有设置命令行参数` ` ` ` args . patrical)

在上一篇博文《多任务学习中的数据分布问题(一)》(链接：https://www . cn blogs.com/Orion-Orion/p/15621953 . html)中，我们提到论文[1]在联邦学习的背景下引入了多任务学习，采用的方法是让每个客户端/Synthetic节点的训练数据分布不同，这样每个任务节点就可以学习不同的模型。

本文的实验手段是用FEMNIST、CIFAR10、Shakespare、Synthetic等数据集对模型进行检验，这些数据集包括CV、NLP和一般分类/回归。但是，在用一组数据集进行测试的过程中，需要解决的任务类型和所有客户端节点上的运行模型是相同的(例如，如果使用CIFAR10数据集，则所有客户端节点都将使用MobileNet-v2网络；采用Shakespare数据集，所有任务节点均为堆叠-LSTM网络)。此时，疑惑来了。既然单个实验的数据集和网络的数据集是一样的，何必多任务呢？本文采用的方法是在单个实验中对原始数据集进行非独立同分布(non_idd)处理。

随机抽样，对不同的非任务生成不同分布的\(T\)数据集，使每个任务节点训练的模型不同。

接下来，我们将仔细研究本文的数据集划分和随机抽样算法。本文的代码已在Github上开源[2]。我们以CIFAR10数据集的生成为例，详细分析了本文的数据集采样算法。

1.数据集导入

首先，训练和测试数据集从torchvision导入，并拼接成数据集。

从torchvision.datasets导入CIFAR10

从torchvision.transforms导入撰写，到传感器，规范化

从torch.utils.data导入连接数据集

#相对于此文档的相对目录

RAW_DATA_PATH='raw_data/'

转换=撰写([

ToTensor()，

归一化((0.4914，0.4822，0.4465)，(0.2023，0.1994，0.2010))

])

#此数据集对象可以与torch.utils.data.DataLoader并行加载

数据集=\

串联数据集([

#转换到输入处理(目标转换到目标处理)

# download为真，会自动下载到参数root对应的目录。如果它已经存在，将不会被下载。

# download为false，不会自动下载。

# train=True，从训练集中创建数据

CIFAR10(根=RAW_DATA_PATH，下载=真，训练=真，变换=变换)，

# test=False，从测试集中创建数据

CIFAR10(根=RAW_DATA_PATH，下载=假，训练=假，变换=变换)

])

2. 数据集拆分到client

那么划分数据有两种方式，一种是按照病态的非独立同分布来划分数据(其实第一种方式是论文[2]中提出的划分方法，此时命令行参数args。病理_ non _ iid _ split=true)，另一种是根据标签划分数据(如果没有设置命令行参数A)

rgs.pathological_non_iid_split ，则默认按照标签划分）。

2.1 病态独立同分布划分方式((pathological non iid split))

我们先来看按照病态独立同分布来划分数据。如果选择这种划分方式，则每个client会受到\(n\)个shard（碎片）的数据集，每个碎片最多包含两个类别。此时我们可以选择传入参数args.n_shard，该参数表示每个client/task的碎片数量（默认值为2）。（当然，如果没有按照病态非独立同分布来划分数据，则不需要设置args.n_shard参数）
然后，我们将数据集划分到各client上，此时我们需要将这个功能编写成一个函数并进行调用：

clients_indices = \
    clients_indices =\
        pathological_non_iid_split(
            dataset=dataset,
            n_classes=N_CLASSES,
            n_clients=args.n_tasks,
            n_classes_per_client=args.n_shards,
            frac=args.s_frac,
            seed=args.seed
        )

接下来我们来看这个函数如何设计。先看函数原型：

def pathological_non_iid_split(dataset, n_classes, n_clients, n_classes_per_client, frac=1, seed=1234):

我们解释一下函数的参数，这里dataset是torch.utils.Dataset类型的数据集，n_classes表示数据集里样本分类数，n_client表示client节点的数量，n_client_per_client表示每个client中的类别数，frac是使用数据集的比例(默认是1，即使用全部数据)，seed是传入的随机数种子。该函数返回一个由n_client个subgroup组成的列表client_indices，每个subgroup对应某个client所需的样本索引组成的列表。

接下来我们看这个函数的内容。该函数完成的功能可以概括为：先将样本按照标签进行排序；再将样本划分为n_client * n_classes_per_client个shards（每个shard大小相等），对n_clients中的每一个client分配n_classes_per_client个shards（分配到client后，每个client中的shards要合并）。

首先，我们根据frac获取数据集的子集。

    rng_seed = (seed if (seed is not None and seed = 0) else int(time.time())) 
    rng = random.Random(rng_seed)
    np.random.seed(rng_seed)
    # get subset
    n_samples = int(len(dataset) * frac)
    selected_indices = rng.sample(list(range(len(dataset))), n_samples)

然后从被选出的数据集索引selected_indices建立一个key为类别\(\{0,1,...,n\_classes-1\}\)，value为对应样本集索引列表的字典，这在实际上这就相当于按照label对样本进行排序了。

    label2index = {k: [] for k in range(n_classes)}
    for idx in selected_indices:
        _, label = dataset[idx]
        label2index[label].append(idx)
    sorted_indices = []
    for label in label2index:
        sorted_indices += label2index[label]

然后该函数将数据分为n_clients * n_classes_per_client 个独立同分布的shards，每个shards大小相等。然后给n_clients中的每一个client分配n_classes_per_client个shards（分配到client后，每个client中的shards要合并），代码如下：

    def iid_divide(l, g):
        """
        将列表`l`分为`g`个独立同分布的group（其实就是直接划分）
        每个group都有 `int(len(l)/g)` 或者 `int(len(l)/g)+1` 个元素
        返回由不同的groups组成的列表
        """
        num_elems = len(l)
        group_size = int(len(l) / g)
        num_big_groups = num_elems - g * group_size
        num_small_groups = g - num_big_groups
        glist = []
        for i in range(num_small_groups):
            glist.append(l[group_size * i: group_size * (i + 1)])
        bi = group_size * num_small_groups
        group_size += 1
        for i in range(num_big_groups):
            glist.append(l[bi + group_size * i:bi + group_size * (i + 1)])
        return glist
    n_shards = n_clients * n_classes_per_client
    # 一共分成n_shards个独立同分布的shards
    shards = iid_divide(sorted_indices, n_shards)
    random.shuffle(shards)
    # 然后再将n_shards拆分为n_client份
    tasks_shards = iid_divide(shards, n_clients)
    clients_indices = [[] for _ in range(n_clients)]
    for client_id in range(n_clients):
        for shard in tasks_shards[client_id]:
            # 这里shard是一个shard的数据索引(一个列表)
            # += shard 实质上是在列表里并入列表
            clients_indices[client_id] += shard 

最后，返回clients_indices

    return clients_indices

2.2 按照标签划分划分方式(split dataset by labels)

现在我们来看按照标签来划分数据。如果选择这种划分方式，则不再传入参数args.n_shard进行shard的划分。我们只需要将数据集标签进行排序后直接划分到各client上，此时我们需要将这个功能编写成一个函数并进行调用：

clients_indices = \
    split_dataset_by_labels(
        dataset=dataset,
        n_classes=N_CLASSES,
        n_clients=args.n_tasks,
        n_clusters=args.n_components,
        alpha=args.alpha,
        frac=args.s_frac,
        seed=args.seed
    )

接下来我们来看这个函数如何设计。先看函数原型：

def split_dataset_by_labels(dataset, n_classes, n_clients, n_clusters, alpha, frac, seed=1234):

我们解释一下函数的参数，这里dataset是torch.utils.Dataset类型的数据集，n_classes表示数据集里样本分类数，n_clusters是簇的个数（后面会解释其含义，如果设置为-1，则就默认n_clusters=n_classes），alpha 用于控制clients之间的数据diversity（多样性），frac是使用数据集的比例(默认是1，即使用全部数据)，seed是传入的随机数种子。该函数返回一个由n_client个subgroup组成的列表client_indices，每个subgroup对应某个client所需的样本索引组成的列表。

接下来我们看这个函数的内容。这个函数的内容可以概括为：先将所有类别分组为n_clusters个簇；再对每个簇c，将样本划分给不同的clients（每个client的样本数量按照dirichlet分布来确定）。

首先，我们判断n_clusters的数量，如果为-1，则默认每一个cluster对应一个数据class：

    if n_clusters == -1:
        n_clusters = n_classes

然后得到随机数生成器（简称rng）：

    rng_seed = (seed if (seed is not None and seed = 0) else int(time.time()))
    rng = random.Random(rng_seed)
    np.random.seed(rng_seed)

然后将打乱后的标签集合\(\{0,1,...,n\_classes-1\}\)分为n_clusters个独立同分布的簇。

    all_labels = list(range(n_classes))
    rng.shuffle(all_labels)
    clusters_labels = iid_divide(all_labels, n_clusters)

然后再建立根据上面划分为簇的标签(clusters_labels)建立key为label, value为簇id(group_idx)的字典，

    label2cluster = dict()  # maps label to its cluster
    for group_idx, labels in enumerate(clusters_labels):
        for label in labels:
            label2cluster[label] = group_idx

接着获取数据集的子集

    n_samples = int(len(dataset) * frac)
    selected_indices = rng.sample(list(range(len(dataset))), n_samples)

之后，我们

    # 记录每个cluster大小的向量
    clusters_sizes = np.zeros(n_clusters, dtype=int)
    # 存储每个cluster对应的数据索引
    clusters = {k: [] for k in range(n_clusters)}
    for idx in selected_indices:
        _, label = dataset[idx]
        # 由样本数据的label先找到其cluster的id
        group_id = label2cluster[label]
        # 再将对应cluster的大小+1
        clusters_sizes[group_id] += 1
        # 将样本索引加入其cluster对应的列表中
        clusters[group_id].append(idx)
    # 将每个cluster对应的样本索引列表打乱
    for _, cluster in clusters.items():
        rng.shuffle(cluster)

接着，我们按照dirichlet分布设置每一个cluster的样本个数。

    # 记录来自每个cluster的client的样本数量
    clients_counts = np.zeros((n_clusters, n_clients), dtype=np.int64) 
    # 遍历每一个cluster
    for cluster_id in range(n_clusters):
        # 对每个cluster中的每个client赋予一个满足dirichlet分布的权重
        weights = np.random.dirichlet(alpha=alpha * np.ones(n_clients))
        # np.random.multinomial 表示投掷骰子clusters_sizes[cluster_id]次，落在各client上的权重依次是weights
        # 该函数返回落在各client上各多少次，也就对应着各client应该分得的样本数
        clients_counts[cluster_id] = np.random.multinomial(clusters_sizes[cluster_id], weights)
    # 对每一个cluster上的每一个client的计数次数进行前缀（累加）求和，
    # 相当于最终返回的是每一个cluster中按照client进行划分的样本分界点下标
    clients_counts = np.cumsum(clients_counts, axis=1)

然后，我们根据每一个cluster中的每一个client分得的样本情况（我们已经得到了每一个cluster中按照client进行划分的样本分界点下标），合并归纳得到每一个client中分得的样本情况。

    def split_list_by_indices(l, indices):
        """
        将列表`l` 划分为长度为 `len(indices)` 的子列表
        第`i`个子列表从下标 `indices[i]` 到下标`indices[i+1]`
        （从下标0到下标`indices[0]`的子列表另算）
        返回一个由多个子列表组成的列表
        """
        res = []
        current_index = 0
        for index in indices: 
            res.append(l[current_index: index])
            current_index = index
        return res
    
    clients_indices = [[] for _ in range(n_clients)]
    for cluster_id in range(n_clusters):
        # cluster_split为一个cluster中按照client划分好的样本
        cluster_split = split_list_by_indices(clusters[cluster_id], clients_counts[cluster_id])
        # 将每一个client的样本累加上去
        for client_id, indices in enumerate(cluster_split):
            clients_indices[client_id] += indices

最后，我们返回每个client对应的样本索引：

    return clients_indices

3. 总结

按照病态独立同分布划分和按照样本标签划分两种方式，其实本质上都是要使每个client的分布不同，而这也是我们进行多任务学习的前提。

参考文献

• [1] Marfoq O, Neglia G, Bellet A, et al. Federated multi-task learning under a mixture of distributions[J]. Advances in Neural Information Processing Systems, 2021, 34.
• [2] McMahan B, Moore E, Ramage D, et al. Communication-efficient learning of deep networks from decentralized data[C]//Artificial intelligence and statistics. PMLR, 2017: 1273-1282.

数学是符号的艺术，音乐是上界的语言。