Federated Learning with Non-IID Data

1. 引言

联邦学习的核心假设之一是参与客户端的数据独立同分布（Independent and Identically Distributed, IID）。然而在实际场景中，这一假设几乎不可能满足。不同医院、不同银行、不同物联网设备产生的数据，其分布必然存在显著差异。这种非独立同分布（Non-IID）数据是联邦学习面临的最大挑战之一。

---

2. 非IID数据的成因

2.1 用户行为差异

不同用户的生活习惯、职业背景、年龄等因素导致其行为数据分布不同：

• 输入法习惯：不同用户打字频率、常用词、输入速度差异巨大
• 健康数据：不同年龄段、性别、职业的健康指标分布不同
• 购物行为：不同收入水平、地区的消费者偏好不同

2.2 设备传感器差异

物联网场景下，设备本身的差异也会导致数据分布不同：

• 相机传感器：不同手机的摄像头参数不同
• 加速度计：不同设备的传感器精度和噪声水平不同
• 麦克风：不同设备的采样率和频率响应不同

2.3 时间与地理因素

数据采集的时间和地理位置也会影响分布：

• 季节性变化：夏季和冬季的用电习惯不同
• 地理位置：不同地区的用户偏好存在差异
• 社会事件：突发事件会改变短期数据分布

---

3. 非IID数据的形式化分类

3.1 基于数据分布的分类

根据概率分布的不同角度，非IID数据可分为以下几类：

非IID数据分类体系
│
├── 协变量偏移 (Covariate Shift)
│   ├── P(X) 变化，P(Y|X) 不变
│   └── 场景：不同设备采集的数据分布不同
│
├── 先验偏移 (Prior Shift)
│   ├── P(Y) 变化，P(X|Y) 不变
│   └── 场景：不同地区的疾病发病率不同
│
├── 概念偏移 (Concept Drift)
│   ├── P(Y|X) 变化
│   └── 场景：欺诈模式随时间演变
│
└── 全面偏移 (General Shift)
    └── P(X,Y) 都变化

3.2 基于特征的分类（Li等，2020）

类型	数学定义	直观理解
特征分布非IID	$P_k(X)\ne P(X)$	各客户端的特征分布不同
标签分布非IID	$P_k(Y)\ne P(Y)$	各客户端的标签频率不同
相同标签，不同特征	$P_k(X\Vert Y=k)\ne P(X\Vert Y=k)$	同一类别的样本特征不同
相同特征，不同标签	$P_k(Y\Vert X=x)\ne P(Y\Vert X=x)$	相同特征的标签不同
数据量偏移	$n_k\gg n_j$	各客户端数据量差异大

3.3 数量偏移的严重性

数量偏移（Quantity Skew）是非IID的一个重要子类型：

• 极端情况：某些客户端可能拥有99%的数据
• 影响：简单按数据量加权可能导致少数客户端主导全局模型

设客户端 $k$ 的数据量为 $n_k$，则其权重为：

$$p_k=\frac{n_k}{{\sum }_j{n}_j}$$

当 $n_k$ 分布极不均匀时，加权平均会偏向数据量大的客户端。

---

4. 非IID数据对FedAvg的影响

4.1 客户端漂移问题

客户端漂移（Client Drift）是非IID数据对FedAvg的主要影响。当数据非IID时，各客户端的本地梯度方向与全局梯度方向存在偏差：

$$\mathbb{E}[\nabla F_k(\mathbf{w})]\ne \nabla F(\mathbf{w})$$

设全局最优为 ${\mathbf{w}}^{\ast }$，客户端 $k$ 的局部最优为 ${\mathbf{w}}_k^{\ast }$，则：

$$\Vert {\mathbf{w}}_k^{\ast }-{\mathbf{w}}^{\ast }\Vert \ge \frac{{\rho }_k}{\mu }$$

其中 ${\rho }_k$ 是客户端 $k$ 与全局的异构程度，$\mu$ 是损失函数的光滑参数。

4.2 收敛性分析

定理：FedAvg在非IID数据下的收敛上界为：

$$\mathbb{E}[F({\mathbf{w}}_T)]-F^{\ast }\le \underset{\text{标准项}}{\underbrace{O\left(\frac{1}{T}\right)}}+\underset{\text{漂移项}}{\underbrace{O\left(\frac{\rho E}{T^2}\right)}}+\underset{\text{累积项}}{\underbrace{O\left(\frac{{\rho }^2{E}^2}{T}\right)}}$$

其中 $\rho$ 是异构性参数：

$$\rho =\underset{k}{\max }\Vert \nabla F_k({\mathbf{w}}^{\ast })-\nabla F({\mathbf{w}}^{\ast })\Vert$$

直观理解：

• 异构性 $\rho$ 越大，漂移越严重
• 本地epoch数 $E$ 越大，漂移累积越严重
• 当 $E$ 很大时，累积项 $O({\rho }^2{E}^2/T)$ 可能主导，导致收敛困难

4.3 实验观察

在CIFAR-10数据集上的实验显示：

数据划分	FedAvg准确率	收敛所需轮数
IID	85.2%	200轮
2-class per client	71.3%	500轮（不收敛）
10-class per client (shuffled)	78.6%	350轮

---

5. 非IID数据的量化度量

5.1 Earth Mover’s Distance (EMD)

EMD用于度量两个分布之间的距离：

$$\text{EMD}(P,Q)=\underset{\gamma \in \Gamma (P,Q)}{\inf }\int \Vert x-y\Vert d\gamma (x,y)$$

其中 $\Gamma (P,Q)$ 是所有从 $P$ 到 $Q$ 的运输计划的集合。

5.2 分布散度

散度	公式	特点
KL散度	$D_{KL}(P\Vert Q)=\sum P\log \frac{P}{Q}$	非对称，非对称度量
JS散度	$D_{JS}(P\Vert Q)=\frac{1}{2}{D}_{KL}(P\Vert M)+\frac{1}{2}{D}_{KL}(Q\Vert M)$	对称，有界
Wasserstein距离	$W(P,Q)$	满足三角不等式

5.3 异构性参数估计

在实际中，可通过以下方式估计客户端异构性：

def estimate_heterogeneity(clients_data, model):
    """
    估计客户端间的异构性
    """
    gradients = []
    for client_data in clients_data:
        # 计算各客户端的梯度
        grad = compute_gradient(model, client_data)
        gradients.append(grad)
    
    # 计算梯度均值
    mean_grad = torch.stack(gradients).mean(dim=0)
    
    # 计算异构性：梯度与均值的偏差
    heterogeneity = torch.stack([
        torch.norm(g - mean_grad) for g in gradients
    ]).mean()
    
    return heterogeneity.item()

---

6. 解决非IID问题的策略

6.1 策略分类

非IID问题解决方案
│
├── 客户端层面
│   ├── 客户端选择策略
│   ├── 本地训练调整
│   └── 数据增强
│
├── 服务器层面
│   ├── 聚合算法改进
│   ├── 全局正则化
│   └── 服务器端优化
│
└── 架构层面
    ├── 个性化模型
    ├── 知识蒸馏
    └── 混合架构

6.2 客户端选择策略

6.2.1 基于重要性的选择

选择对全局模型贡献最大的客户端：

$$\text{Score}(k)=\frac{n_k}{n}\cdot \Vert \Delta {\mathbf{w}}_k\Vert$$

def importance_based_selection(scores, num_select, threshold):
    """
    基于重要性选择客户端
    """
    selected = []
    for i, score in enumerate(scores):
        if score >= threshold:
            selected.append(i)
    
    # 如果选中数量不足，随机补充
    if len(selected) < num_select:
        remaining = set(range(len(scores))) - set(selected)
        additional = random.sample(remaining, num_select - len(selected))
        selected.extend(additional)
    
    return selected

6.2.2 基于聚类的选择

将相似客户端分组，选择各组的代表：

from sklearn.cluster import KMeans
 
def cluster_based_selection(client_features, num_clusters, samples_per_cluster):
    """
    基于聚类的客户端选择
    """
    # 聚类客户端
    kmeans = KMeans(n_clusters=num_clusters)
    labels = kmeans.fit_predict(client_features)
    
    # 从每个簇中选择样本
    selected = []
    for cluster_id in range(num_clusters):
        cluster_clients = [i for i, l in enumerate(labels) if l == cluster_id]
        # 随机选择或基于数据量选择
        selected.extend(random.sample(
            cluster_clients, 
            min(samples_per_cluster, len(cluster_clients))
        ))
    
    return selected

6.3 本地训练调整

6.3.1 学习率调整

根据客户端的本地数据分布调整学习率：

$${\eta }_k=\eta \cdot \sqrt{\frac{n_k}{n_{\max }}}$$

6.3.2 梯度修正

在本地梯度中加入全局梯度信息：

$${\mathbf{g}}_k^{修正}={\mathbf{g}}_k+\lambda ({\mathbf{g}}_{global}-{\mathbf{g}}_k)$$

6.4 服务器端正则化

6.4.1 FedNova的正则化

FedNova通过归一化梯度来解决异构性问题：

$$\Delta {\mathbf{w}}_k^{norm}=\frac{{\tau }_k}{{\sum }_j{\tau }_j}\cdot \frac{\Delta {\mathbf{w}}_k}{\Vert \Delta {\mathbf{w}}_k\Vert }$$

其中 ${\tau }_k$ 是客户端 $k$ 的有效更新步数。

6.5 数据增强

6.5.1 客户端数据增强

在客户端本地进行数据增强，减小数据分布差异：

class FederatedDataAugmentation:
    def __init__(self, augmentation_ratio=0.3):
        self.aug_ratio = augmentation_ratio
    
    def augment(self, data, labels):
        """
        联邦学习中的数据增强
        """
        n_aug = int(len(data) * self.aug_ratio)
        indices = random.sample(range(len(data)), n_aug)
        
        augmented_data = []
        augmented_labels = []
        
        for idx in indices:
            # Mixup增强
            if random.random() < 0.5:
                # 与另一个随机样本混合
                idx2 = random.choice(range(len(data)))
                alpha = random.random()
                x_aug = alpha * data[idx] + (1-alpha) * data[idx2]
                y_aug = labels[idx]  # 或使用加权标签
            else:
                # 其他增强方法
                x_aug = self.standard_augmentation(data[idx])
                y_aug = labels[idx]
            
            augmented_data.append(x_aug)
            augmented_labels.append(y_aug)
        
        return augmented_data, augmented_labels

6.5.2 全局数据共享

部分工作允许客户端共享少量数据或生成数据来缓解异构性：

$${\mathcal{D}}_{shared}=\text{ShareFewSamples}({\mathcal{D}}_1,{\mathcal{D}}_2,\dots ,{\mathcal{D}}_K)$$

---

7. 评估非IID方法的标准

7.1 模拟非IID数据

在实验中，通过以下方式模拟非IID数据：

def partition_cifar10_non_iid(num_clients=100, alpha=0.5):
    """
    使用Dirichlet分布模拟非IID划分
    
    Args:
        num_clients: 客户端数量
        alpha: Dirichlet分布参数，越小越非IID
    """
    import numpy as np
    from torchvision import datasets
    
    # 加载CIFAR-10
    dataset = datasets.CIFAR10(root='./data', train=True, download=True)
    labels = np.array(dataset.targets)
    n_classes = 10
    
    # Dirichlet分布采样
    concentrations = np.random.dirichlet([alpha] * num_clients, n_classes)
    
    # 为每个客户端分配数据
    client_indices = [[] for _ in range(num_clients)]
    for label in range(n_classes):
        indices = np.where(labels == label)[0]
        np.random.shuffle(indices)
        
        # 按Dirichlet比例分配
        splits = np.split(indices, np.cumsum(
            concentrations[label] * len(indices)
        ).astype(int)[:-1])
        
        for client_id, split in enumerate(splits):
            client_indices[client_id].extend(split)
    
    return client_indices

7.2 评估指标

指标	定义	衡量内容
最终准确率	测试集上的准确率	模型性能
收敛速度	达到某准确率所需的轮数	训练效率
公平性	各客户端准确率的方差	服务质量均衡
通信效率	达到目标性能的总通信量	资源消耗

---

8. 实践指南

8.1 非IID程度诊断

在实际部署中，首先诊断数据的非IID程度：

def diagnose_non_iid(clients_data, test_data):
    """
    诊断非IID程度并给出建议
    """
    # 1. 计算各客户端的准确率差异
    client_accs = []
    for client_data in clients_data:
        acc = evaluate_local_model(client_data, test_data)
        client_accs.append(acc)
    
    acc_variance = np.var(client_accs)
    acc_range = np.max(client_accs) - np.min(client_accs)
    
    # 2. 评估异构性
    if acc_range > 0.2:  # 20%的准确率差异
        print("WARNING: High heterogeneity detected!")
        print(f"  Accuracy range: {acc_range:.2%}")
        print(f"  Recommendation: Use FedProx or SCAFFOLD")
    elif acc_range > 0.1:
        print("MODERATE: Some heterogeneity present")
        print(f"  Consider reducing local epochs E")
    else:
        print("LOW: Data distribution is relatively uniform")
    
    # 3. 检查数据量分布
    sizes = [len(d) for d in clients_data]
    size_variance = np.var(sizes) / np.mean(sizes)
    if size_variance > 1.0:
        print("WARNING: Significant size skew detected!")
        print(f"  Size CV: {np.sqrt(size_variance):.2f}")
        print(f"  Recommendation: Use importance-weighted aggregation")

8.2 超参数调整建议

根据非IID程度调整超参数：

非IID程度	学习率	本地epoch E	建议算法
轻微 (α > 0.5)	标准	5-10	FedAvg
中等 (0.1 < α < 0.5)	略减	2-5	FedProx
严重 (α < 0.1)	大幅减少	1-2	SCAFFOLD

8.3 监控与调试

class FLMonitor:
    def __init__(self, thresholds):
        self.thresholds = thresholds
        self.history = {
            'client_drifts': [],
            'model_updates': [],
            'client_accs': []
        }
    
    def log_round(self, client_updates, global_model):
        """
        记录每轮状态
        """
        # 计算客户端漂移
        drifts = [torch.norm(u) for u in client_updates]
        avg_drift = np.mean(drifts)
        self.history['client_drifts'].append(avg_drift)
        
        # 检测异常
        if avg_drift > self.thresholds['drift']:
            self.alert("High client drift detected!")
        
        # 更新模型更新历史
        self.history['model_updates'].append(
            torch.norm(sum(client_updates))
        )

---

9. 参考文献

---

10. 相关主题

• federated-learning-fundamentals — 联邦学习基础
• fedavg-fedprox-algorithms — FedAvg与FedProx算法
• scaffold-fednova-algorithms — SCAFFOLD与FedNova方差缩减算法
• personalized-federated-learning — 个性化联邦学习方法