在上一篇博文《多任务学习中的数据分布问题(一)》(链接:https://www.cnblogs.com/orion-orion/p/15621953.html)中我们提到论文[1]在联邦学习的情景下引入了多任务学习,其采用的手段是使每个client/Synthetic节点的训练数据分布不同,从而使各任务节点学习到不同的模型。
该论文的实验手段是先后用FEMNIST、CIFAR10、Shakespare、Synthetic等数据集对模型进行测试,这些数据集包括CV、NLP、普通分类/回归这三种不同的任务。但是,该论文在给定用一组数据集进行测试的过程中,所有client节点上要解决的任务种类和运行的模型是一样的(比如采用CIFAR10数据集,则所有client节点都采用MobileNet-v2网络;采用Shakespare数据集,所有任务节点采用Stacked-LSTM网络)。那么此时疑惑就来了,既然单次实验的数据集一样,网络也一样,那么谈何多任务呢?文章采用的手段是,在单次实验时,对原始数据集进行非独立同分布(non_idd)
的随机采样,为(T)个不同非任务生成(T)个不同分布的数据集,以做到每个任务节点训练出的模型不同。
接下来,我们就来仔细研究论文的数据集划分与随机采样算法。该论文的代码已经开源在Github上[2]。以下我们以CIFAR10数据集的生成为例,来详细地对该论文的数据集采样算法进行分析。
1.数据集导入
首先,从torchvision中导入训练和测试数据集,并统一拼接成一个dataset。
from torchvision.datasets import CIFAR10
from torchvision.transforms import Compose, ToTensor, Normalize
from torch.utils.data import ConcatDataset
# 相对于本文件的相对目录
RAW_DATA_PATH = "raw_data/"
transform = Compose([
ToTensor(),
Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010))
])
# 这个dataset对象可以用torch.utils.data.DataLoader并行加载
dataset =
ConcatDataset([
# transform对input处理(targettransform对target处理)
# download为true,会自动下载到参数root对应的目录,如果已经有了,就不会下载
# download为false,不会自动下载。
# train = True,从训练集create数据
CIFAR10(root=RAW_DATA_PATH, download=True, train=True, transform=transform),
# test = False,从测试集create数据
CIFAR10(root=RAW_DATA_PATH, download=False, train=False, transform=transform)
])
2. 数据集拆分到client
然后有两种划分方式,一种是按照病态非独立同分布来划分数据(其实最开始是论文[2]提出的划分方式,此时传入的命令行参数args.pathological_non_iid_split=True
),一种按照标签对数据进行划分(若没有设置命令行参数args.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.
文章来源: 博客园
- 还没有人评论,欢迎说说您的想法!