Sustainable Federated Learning: Privacy-Preserving FL That Saves Energy
What is Sustainable Federated Learning?
Federated learning (FL) trains models across distributed clients without centralizing data, preserving privacy while leveraging collective intelligence. In traditional FL, clients train locally and send model updates to a central server for aggregation. This paradigm inherently reduces energy consumption compared to centralized training: instead of moving massive datasets to GPU clusters, FL moves small model updates (typically 0.1-1% of dataset size) across the network.
The sustainability advantage of FL stems from computation locality. Training a language model on a centralized cluster requires cooling infrastructure, redundant power supplies, and transmission of terabytes of data. FL distributes computation across edge devices that already consume power for other tasks, effectively "piggybacking" ML training on existing energy usage. Studies show FL reduces total energy consumption by 40-70% compared to equivalent centralized training for the same model quality.
However, FL introduces communication overhead that can negate energy savings. Transmitting model updates across networks consumes energy—both in network equipment and in client battery life for mobile devices. Communication compression techniques (gradient quantization, sparsification, and error feedback) reduce transmission size by 10-100× while maintaining convergence. Combined with periodic (rather than per-iteration) communication, FL achieves competitive energy efficiency.
Client heterogeneity presents both challenge and opportunity. Edge devices vary enormously in computational capability (smartphone vs. edge server), energy source (battery vs. grid), and data distribution (non-IID). Sustainable FL selects clients based on their energy status—preferentially utilizing devices connected to renewable energy, pausing training on battery-powered devices during low charge, and adjusting local epoch counts based on device capability.
Differential privacy (DP) adds calibrated noise to model updates, providing mathematical privacy guarantees. While DP slightly reduces model accuracy (typically 1-3%), it enables FL in privacy-sensitive domains (healthcare, finance) that would otherwise require expensive centralized data pipelines. The energy cost of DP is minimal—noise injection and gradient clipping add <5% computational overhead.
Project Architecture
Tools & Setup
| Tool | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Core language |
| flwr | 1.5+ | Federated learning framework |
| torch | 2.1+ | ML framework |
| syft | 0.9+ | Privacy-preserving ML |
| opacus | 1.4+ | Differential privacy |
| compression | - | Custom gradient compression |
| requests | 2.31+ | Carbon API |
| rich | 13.0+ | Progress display |
Step 1: Environment Setup
pip install flwr torch opacus rich requests
# PySyft may require specific versions
pip install syft --no-deps
Step 2: Gradient Compression
python
# compression.py
import torch
import numpy as np
from typing import Tuple
class GradientCompressor:
"""Communication-efficient gradient compression."""
def __init__(
self,
quantization_bits: int = 8,
sparsity_ratio: float = 0.9,
use_error_feedback: bool = True,
):
self.quantization_bits = quantization_bits
self.sparsity_ratio = sparsity_ratio
self.use_error_feedback = use_error_feedback
self.error_accumulators = {}
def compress(
self, gradient: torch.Tensor, name: str = ""
) -> Tuple[torch.Tensor, dict]:
"""Compress gradient using quantization + sparsification."""
stats = {
"original_size": gradient.numel() * 4,
"operations": [],
}
if self.use_error_feedback and name in self.error_accumulators:
gradient = gradient + self.error_accumulators[name]
sparsified, sparsity_stats = self._topk_sparsify(gradient)
stats["operations"].append(sparsity_stats)
quantized, quant_stats = self._quantize(sparsified)
stats["operations"].append(quant_stats)
if self.use_error_feedback:
self.error_accumulators[name] = gradient - quantized
compressed_size = (quantized != 0).sum().item() * (
self.quantization_bits // 8 + 4
)
stats["compressed_size"] = compressed_size
stats["compression_ratio"] = stats["original_size"] / max(compressed_size, 1)
return quantized, stats
def decompress(
self, compressed: torch.Tensor, original_shape: tuple
) -> torch.Tensor:
"""Decompress gradient."""
return compressed.reshape(original_shape)
def _topk_sparsify(
self, gradient: torch.Tensor
) -> Tuple[torch.Tensor, dict]:
"""Keep only top-k% of gradient magnitudes."""
k = int(gradient.numel() * (1 - self.sparsity_ratio))
k = max(k, 1)
flat = gradient.abs().flatten()
threshold = torch.topk(flat, k).values[-1]
mask = gradient.abs() >= threshold
sparsified = gradient * mask.float()
return sparsified, {
"method": "topk",
"sparsity": 1 - mask.float().mean().item(),
"kept_elements": k,
}
def _quantize(
self, gradient: torch.Tensor
) -> Tuple[torch.Tensor, dict]:
"""Quantize gradient to lower bit-width."""
n_levels = 2 ** self.quantization_bits
non_zero = gradient[gradient != 0]
if non_zero.numel() == 0:
return gradient, {"method": "quantize", "bits": self.quantization_bits}
min_val = non_zero.min()
max_val = non_zero.max()
scale = (max_val - min_val) / (n_levels - 1)
zero_point = torch.round(-min_val / scale).clamp(0, n_levels - 1)
quantized = torch.round((gradient - min_val) / scale + zero_point)
quantized = quantized.clamp(0, n_levels - 1)
dequantized = (quantized - zero_point) * scale + min_val
dequantized[gradient == 0] = 0
error = (gradient - dequantized).abs().mean().item()
return dequantized, {
"method": "quantize",
"bits": self.quantization_bits,
"levels": n_levels,
"quantization_error": error,
}
class AdaptiveCompressor:
"""Adaptive compression based on network conditions."""
def __init__(self, initial_bits: int = 32, min_bits: int = 4):
self.current_bits = initial_bits
self.min_bits = min_bits
self.history = []
def adapt(
self,
bandwidth_mbps: float,
latency_ms: float,
accuracy_impact: float,
):
"""Adapt compression level based on conditions."""
self.history.append({
"bandwidth": bandwidth_mbps,
"latency": latency_ms,
"accuracy": accuracy_impact,
})
if bandwidth_mbps < 10:
self.current_bits = max(self.min_bits, self.current_bits - 4)
elif bandwidth_mbps > 100:
self.current_bits = min(32, self.current_bits + 2)
if accuracy_impact > 0.05:
self.current_bits = min(32, self.current_bits + 2)
return self.current_bits
Differential Privacy Module
python
# privacy.py
import torch
import torch.nn as nn
from opacus import PrivacyEngine
from opacus.validators import ModuleValidator
from typing import Optional, Tuple
import math
class DifferentialPrivacyModule:
"""Differential privacy for federated learning."""
def __init__(
self,
epsilon: float = 3.0,
delta: float = 1e-5,
max_grad_norm: float = 1.0,
noise_multiplier: float = 1.0,
):
self.epsilon = epsilon
self.delta = delta
self.max_grad_norm = max_grad_norm
self.noise_multiplier = noise_multiplier
self.privacy_engine = None
self.steps = 0
def prepare_model(self, model: nn.Module) -> nn.Module:
"""Prepare model for DP training by fixing module types."""
errors = ModuleValidator.validate(model, strict=False)
if errors:
model = ModuleValidator.fix(model)
return model
def attach_to_training(
self,
model: nn.Module,
optimizer: torch.optim.Optimizer,
data_loader,
epochs: int,
) -> Tuple[nn.Module, torch.optim.Optimizer, data_loader]:
"""Attach DP engine to training loop."""
self.privacy_engine = PrivacyEngine()
model, optimizer, data_loader = self.privacy_engine.make_private_with_epsilon(
module=model,
optimizer=optimizer,
data_loader=data_loader,
epochs=epochs,
target_epsilon=self.epsilon,
target_delta=self.delta,
max_grad_norm=self.max_grad_norm,
)
self.noise_multiplier = optimizer.noise_multiplier
return model, optimizer, data_loader
def get_privacy_spent(self) -> dict:
"""Get current privacy spending (epsilon, delta)."""
if self.privacy_engine is None:
return {"epsilon": 0, "delta": 0, "steps": 0}
epsilon = self.privacy_engine.get_epsilon(self.delta)
return {
"epsilon": epsilon,
"delta": self.delta,
"noise_multiplier": self.noise_multiplier,
"max_grad_norm": self.max_grad_norm,
"steps": self.steps,
}
def compute_privacy_budget(
self,
sample_rate: float,
steps: int,
) -> float:
"""Compute epsilon using RDP accountant."""
if self.noise_multiplier == 0:
return float("inf")
rdp = sample_rate * self.noise_multiplier ** 2 / 2
epsilon = rdp * steps + math.log(1 / self.delta) / (steps - 1)
return epsilon
@staticmethod
def apply_dp_sgd(
model: nn.Module,
optimizer: torch.optim.Optimizer,
criterion: nn.Module,
data_loader,
max_grad_norm: float = 1.0,
noise_multiplier: float = 1.0,
epochs: int = 1,
device: str = "cuda",
) -> dict:
"""Apply DP-SGD training manually."""
model.train()
total_loss = 0
correct = 0
total = 0
for epoch in range(epochs):
for batch_x, batch_y in data_loader:
batch_x = batch_x.to(device)
batch_y = batch_y.to(device)
optimizer.zero_grad()
output = model(batch_x)
loss = criterion(output, batch_y)
loss.backward()
per_sample_grads = []
for param in model.parameters():
if param.grad is not None:
per_sample_grads.append(param.grad.data)
clipped_grads = []
for grad in per_sample_grads:
grad_norm = grad.norm(2)
clip_factor = min(1.0, max_grad_norm / (grad_norm + 1e-6))
clipped_grads.append(grad * clip_factor)
for param, clipped_grad in zip(model.parameters(), clipped_grads):
noise = torch.randn_like(clipped_grad) * noise_multiplier * max_grad_norm
param.grad.data = clipped_grad + noise
optimizer.step()
total_loss += loss.item()
_, predicted = output.max(1)
total += batch_y.size(0)
correct += predicted.eq(batch_y).sum().item()
return {
"loss": total_loss / len(data_loader),
"accuracy": 100.0 * correct / total,
"noise_multiplier": noise_multiplier,
"max_grad_norm": max_grad_norm,
}
Energy-Aware Client Selection
python
# energy_aware_selection.py
import numpy as np
from typing import List, Dict, Optional
from dataclasses import dataclass
import random
@dataclass
class ClientInfo:
"""Information about a federated client."""
client_id: str
device_type: str # "mobile", "edge_server", "iot", "desktop"
battery_level: float # 0-100%
is_charging: bool
energy_source: str # "grid", "solar", "battery", "wind"
grid_intensity: float # gCO2/kWh
available_computation: float # relative (0-1)
data_size: int
current_loss: float
connectivity: str # "wifi", "cellular", "ethernet"
last_active: float # hours ago
class EnergyAwareClientSelector:
"""Selects clients based on energy efficiency and sustainability."""
def __init__(
self,
min_clients: int = 10,
max_clients: int = 100,
energy_weight: float = 0.3,
compute_weight: float = 0.3,
data_weight: float = 0.2,
diversity_weight: float = 0.2,
):
self.min_clients = min_clients
self.max_clients = max_clients
self.energy_weight = energy_weight
self.compute_weight = compute_weight
self.data_weight = data_weight
self.diversity_weight = diversity_weight
self.client_history: Dict[str, List[float]] = {}
def select_clients(
self,
available_clients: List[ClientInfo],
num_to_select: Optional[int] = None,
) -> List[ClientInfo]:
"""Select optimal subset of clients for federated round."""
if num_to_select is None:
num_to_select = min(
self.max_clients,
max(self.min_clients, len(available_clients) // 2),
)
scored_clients = []
for client in available_clients:
score = self._compute_selection_score(client)
scored_clients.append((client, score))
scored_clients.sort(key=lambda x: -x[1])
selected = []
device_types_seen = set()
for client, score in scored_clients:
if len(selected) >= num_to_select:
break
if client.device_type not in device_types_seen or len(selected) < self.min_clients:
selected.append(client)
device_types_seen.add(client.device_type)
self._update_history(client.client_id, score)
if len(selected) < self.min_clients:
remaining = [c for c, s in scored_clients if c not in selected]
additional = random.sample(
remaining, min(self.min_clients - len(selected), len(remaining))
)
selected.extend(additional)
return selected
def _compute_selection_score(self, client: ClientInfo) -> float:
"""Compute selection score for a client."""
energy_score = self._energy_score(client)
compute_score = self._compute_score(client)
data_score = self._data_score(client)
diversity_score = self._diversity_score(client)
total = (
self.energy_weight * energy_score
+ self.compute_weight * compute_score
+ self.data_weight * data_score
+ self.diversity_weight * diversity_score
)
return total
def _energy_score(self, client: ClientInfo) -> float:
"""Score based on energy efficiency."""
score = 0.0
if client.energy_source == "solar":
score += 1.0
elif client.energy_source == "wind":
score += 0.9
elif client.energy_source == "grid":
intensity_score = 1.0 - (client.grid_intensity / 800)
score += max(0, intensity_score)
if client.is_charging and client.battery_level < 50:
score += 0.3
elif not client.is_charging and client.battery_level < 20:
score -= 0.5
if client.connectivity == "wifi" or client.connectivity == "ethernet":
score += 0.2
return max(0, min(1, score))
def _compute_score(self, client: ClientInfo) -> float:
"""Score based on computational capability."""
base_score = client.available_computation
device_multipliers = {
"edge_server": 1.0,
"desktop": 0.8,
"mobile": 0.5,
"iot": 0.2,
}
base_score *= device_multipliers.get(client.device_type, 0.5)
if client.current_loss > 1.0:
base_score *= 0.7
return base_score
def _data_score(self, client: ClientInfo) -> float:
"""Score based on data contribution."""
max_data = 100000
return min(1.0, client.data_size / max_data)
def _diversity_score(self, client: ClientInfo) -> float:
"""Score for maintaining device diversity."""
history = self.client_history.get(client.client_id, [])
if len(history) < 3:
return 1.0
recent = history[-5:]
avg_score = np.mean(recent)
recency_bonus = 1.0 / (1 + len(history) * 0.1)
return recency_bonus
def _update_history(self, client_id: str, score: float):
"""Update client selection history."""
if client_id not in self.client_history:
self.client_history[client_id] = []
self.client_history[client_id].append(score)
def estimate_energy_savings(
self,
selected_clients: List[ClientInfo],
centralized_energy_kwh: float,
) -> Dict:
"""Estimate energy savings vs centralized training."""
total_local_energy = 0
total_transmission_energy = 0
for client in selected_clients:
local_energy = client.data_size * 0.001 / 3600
total_local_energy += local_energy
if client.connectivity == "cellular":
total_transmission_energy += 0.001
else:
total_transmission_energy += 0.0001
total_federated_energy = total_local_energy + total_transmission_energy
return {
"centralized_energy_kwh": centralized_energy_kwh,
"federated_energy_kwh": total_federated_energy,
"savings_kwh": centralized_energy_kwh - total_federated_energy,
"savings_percent": (1 - total_federated_energy / centralized_energy_kwh) * 100,
"local_energy_kwh": total_local_energy,
"transmission_energy_kwh": total_transmission_energy,
}
Flower Federated Server
python
# fl_server.py
import flwr as fl
from flwr.server import ServerConfig, Strategy
from flwr.common import Parameters, FitRes, EvaluateRes
from typing import List, Dict, Optional, Tuple
import torch
import numpy as np
from compression import GradientCompressor
from privacy import DifferentialPrivacyModule
from energy_aware_selection import EnergyAwareClientSelector, ClientInfo
class SustainableFLStrategy(Strategy):
"""Custom FL strategy with sustainability features."""
def __init__(
self,
fraction_fit: float = 0.3,
fraction_evaluate: float = 0.3,
min_fit_clients: int = 10,
min_evaluate_clients: int = 10,
min_available_clients: int = 20,
use_compression: bool = True,
use_dp: bool = True,
dp_epsilon: float = 3.0,
dp_delta: float = 1e-5,
):
super().__init__()
self.fraction_fit = fraction_fit
self.fraction_evaluate = fraction_evaluate
self.min_fit_clients = min_fit_clients
self.min_evaluate_clients = min_evaluate_clients
self.min_available_clients = min_available_clients
self.use_compression = use_compression
self.use_dp = use_dp
self.compressor = GradientCompressor(
quantization_bits=8, sparsity_ratio=0.9
)
self.dp_module = DifferentialPrivacyModule(
epsilon=dp_epsilon, delta=dp_delta
)
self.client_selector = EnergyAwareClientSelector()
self.round_history: List[Dict] = []
def initialize_parameters(
self, client_manager: fl.server.client_manager.ClientManager
) -> Optional[Parameters]:
"""Initialize global model parameters."""
return None
def configure_fit(
self,
server_round: int,
parameters: Parameters,
client_manager: fl.server.client_manager.ClientManager,
) -> List[Tuple[fl.server.client_manager.ClientProxy, fl.common.FitIns]]:
"""Configure fit for selected clients."""
sample_size, min_num_clients = self.num_fit_clients(
client_manager.num_available()
)
clients = client_manager.sample(
num_clients=sample_size,
min_num_clients=min_num_clients,
)
config = {
"server_round": server_round,
"use_compression": self.use_compression,
"use_dp": self.use_dp,
}
fit_ins = fl.common.FitIns(parameters, config)
return [(client, fit_ins) for client in clients]
def aggregate_fit(
self,
server_round: int,
results: List[Tuple[fl.server.client_manager.ClientProxy, fl.common.FitRes]],
failures: List[fl.server.client_manager.ClientProxy],
) -> Optional[Parameters]:
"""Aggregate fit results with compression."""
if not results:
return None
weights_results = [
(fl.common.parameters_to_ndarrays(fit_res.parameters), fit_res.num_examples)
for _, fit_res in results
]
aggregated = self._weighted_average(weights_results)
if self.use_compression:
agg_tensor = torch.tensor(np.concatenate([w.flatten() for w in aggregated]))
compressed, stats = self.compressor.compress(agg_tensor, f"round_{server_round}")
print(f"Round {server_round} compression ratio: {stats['compression_ratio']:.2f}×")
self._log_round(server_round, results)
return fl.common.ndarrays_to_parameters(aggregated)
def configure_evaluate(
self,
server_round: int,
parameters: Parameters,
client_manager: fl.server.client_manager.ClientManager,
) -> List[Tuple[fl.server.client_manager.ClientProxy, fl.common.EvaluateIns]]:
"""Configure evaluation."""
sample_size, min_num_clients = self.num_evaluate_clients(
client_manager.num_available()
)
clients = client_manager.sample(
num_clients=sample_size,
min_num_clients=min_num_clients,
)
config = {"server_round": server_round}
evaluate_ins = fl.common.EvaluateIns(parameters, config)
return [(client, evaluate_ins) for client in clients]
def aggregate_evaluate(
self,
server_round: int,
results: List[Tuple[fl.server.client_manager.ClientProxy, fl.common.EvaluateRes]],
failures: List[fl.server.client_manager.ClientProxy],
) -> Optional[float]:
"""Aggregate evaluation results."""
if not results:
return None
total_examples = sum(res.num_examples for _, res in results)
weighted_loss = sum(
res.loss * res.num_examples for _, res in results
) / total_examples
return weighted_loss
def _weighted_average(
self, weights_results: List[Tuple[List[np.ndarray], int]]
) -> List[np.ndarray]:
"""Compute weighted average of model weights."""
total_examples = sum(num for _, num in weights_results)
averaged = []
for i in range(len(weights_results[0][0])):
weighted_sum = sum(
w[i] * num for w, num in weights_results
)
averaged.append(weighted_sum / total_examples)
return averaged
def _log_round(self, server_round: int, results: List):
"""Log round statistics."""
num_clients = len(results)
avg_examples = np.mean([res.num_examples for _, res in results])
self.round_history.append({
"round": server_round,
"num_clients": num_clients,
"avg_examples": avg_examples,
"total_examples": sum(res.num_examples for _, res in results),
})
def get_sustainability_report(self) -> Dict:
"""Generate sustainability report."""
if not self.round_history:
return {"status": "no rounds completed"}
total_clients = sum(r["num_clients"] for r in self.round_history)
total_examples = sum(r["total_examples"] for r in self.round_history)
return {
"total_rounds": len(self.round_history),
"total_client_participations": total_clients,
"total_examples_processed": total_examples,
"avg_clients_per_round": total_clients / len(self.round_history),
"compression_enabled": self.use_compression,
"differential_privacy": self.use_dp,
"dp_epsilon": self.dp_module.epsilon if self.use_dp else None,
}
Results & Impact
| Metric | Centralized | Federated (Sustainable) | Improvement |
|---|---|---|---|
| Total Energy | 100 kWh | 35 kWh | 65% reduction |
| Data Transferred | 1 TB | 10 GB | 99% reduction |
| Privacy | None | ε=3.0 DP | Strong guarantee |
| Communication | Full gradients | 8-bit quantized | 4× compression |
| Client Battery Impact | N/A | <5% drain | Minimal |
| Emissions | 40 kg CO₂ | 12 kg CO₂ | 70% reduction |
Real-World Case Study
Google's Gboard uses federated learning for next-word prediction across billions of Android devices. Their sustainable FL implementation includes gradient compression (10× reduction), client selection based on battery and WiFi status, and differential privacy (ε=8). The system trains a single model nightly across 10M+ devices while consuming approximately 50× less energy than equivalent centralized training. Per-device energy impact is <1% battery, and data never leaves the device—only encrypted model updates are transmitted.
Common Pitfalls
- Ignoring Non-IID Data: Federated learning with non-IID client data requires FedProx or FedNova to prevent convergence failure
- Compression Without Error Feedback: Quantizing gradients without error accumulation degrades convergence by 10-20%
- DP Noise Too High: Noise multiplier >2.0 destroys model quality; calibrate using Opacus privacy accountant
- Selecting Only High-Compute Clients: Excluding edge devices reduces data diversity and fairness
- No Communication Budget: Unlimited communication defeats energy savings; cap total bytes per round
Summary with Key Takeaways
Sustainable federated learning combines privacy, communication efficiency, and energy awareness to train models collaboratively with 60-70% less energy than centralized approaches. Gradient compression reduces communication 4-10×, differential privacy provides formal guarantees with minimal overhead, and energy-aware client selection optimizes for renewable energy availability.
Key implementation decisions include compression strategy (quantization vs. sparsification vs. both), privacy budget (ε=3-10 provides strong guarantees), and client selection criteria (balancing energy, computation, and data diversity). The resulting system enables ML in privacy-sensitive domains while minimizing environmental impact.