字节跳动推荐系统绿色计算实践：从GPU优化到动态能效管理

2026-02-04 04:58:26作者：蔡丛锟

在数字经济高速发展的今天，数据中心的能源消耗问题日益凸显。据统计，大型互联网公司的数据中心电力成本占总运营成本的15%-20%，其中AI推荐系统作为计算密集型应用，更是能源消耗的"大户"。你是否还在为推荐系统的高能耗而困扰？是否想在不降低推荐效果的前提下实现算力成本的优化？本文将以字节跳动推荐系统Monolith为案例，详细介绍如何通过GPU资源优化、动态能效管理等技术手段，打造绿色高效的推荐系统。读完本文，你将掌握：

GPU训练任务的智能调度与资源利用率提升方法
动态能效管理系统的设计与实现
字节跳动在绿色计算实践中的具体优化策略与效果数据

项目背景与绿色计算挑战

Monolith作为字节跳动的推荐系统框架，基于TensorFlow构建，支持批量/实时训练和服务。其核心特点包括无冲突嵌入表（collisionless embedding tables）和实时训练能力，能够为不同的ID特征提供唯一表示，并快速捕捉最新的热点内容，帮助用户发现新的兴趣点。

Monolith项目地址

随着推荐系统规模的不断扩大，计算资源需求急剧增加，带来了严峻的能源消耗挑战。主要体现在以下几个方面：

GPU资源利用率不均衡：不同训练任务对GPU资源的需求差异较大，导致部分GPU处于空闲状态或负载过高。
能源消耗与成本压力：大规模GPU集群的电力消耗巨大，不仅增加了运营成本，也对环境造成了压力。
散热与机房环境挑战：高能耗带来的散热问题进一步增加了数据中心的运营复杂度和成本。

为应对这些挑战，字节跳动推荐系统团队开展了一系列绿色计算优化工作，重点围绕GPU资源优化和动态能效管理两大方向。

GPU资源优化策略

GPU训练任务调度与资源隔离

Monolith系统通过精细化的GPU资源管理，实现了训练任务的智能调度和资源隔离。在设备管理模块中，提供了GPU训练的启用/禁用控制和可见GPU的获取功能。

设备管理模块源码

def enable_gpu_training():
    """启用GPU训练"""
    os.environ["MONOLITH_ENABLE_GPU_TRAINING"] = "1"

def disable_gpu_training():
    """禁用GPU训练"""
    os.environ.pop("MONOLITH_ENABLE_GPU_TRAINING", None)

def is_gpu_training():
    """检查是否启用GPU训练"""
    return os.environ.get("MONOLITH_ENABLE_GPU_TRAINING") == "1"

def get_visible_gpus(local_rank, processes_per_gpu=1):
    """获取可见的GPU列表"""
    if not is_gpu_training():
        return []
    all_gpus = os.environ.get("CUDA_VISIBLE_DEVICES", "0").split(",")
    all_gpus = [gpu.strip() for gpu in all_gpus if gpu.strip()]
    if not all_gpus:
        return []
    # 根据local_rank和processes_per_gpu计算当前进程可见的GPU
    gpu_index = (local_rank // processes_per_gpu) % len(all_gpus)
    return [all_gpus[gpu_index]]

通过这些函数，Monolith能够根据任务需求和系统负载动态调整GPU资源分配，避免资源浪费。例如，在分布式训练中，可以为每个进程分配适当的GPU资源，确保GPU利用率最大化。

GPU内存优化与计算效率提升

为了进一步提升GPU资源利用率，Monolith引入了多种内存优化技术。在哈希表操作中，采用了Float16数据类型来减少内存占用，同时通过融合操作减少数据传输开销。

哈希表配置源码

def gen_table_config(cls,
                    dims: List[int],
                    use_float16: bool = False,
                    learning_rate: float = 1.0,
                    enable_gpu_emb: bool = False):
    """生成哈希表配置"""
    config = embedding_hash_table_pb2.EmbeddingHashTableConfig()
    config.dim = dims[0]
    config.learning_rate = learning_rate
    config.enable_gpu_emb = enable_gpu_emb
    if use_float16:
        config.value_type = embedding_hash_table_pb2.EmbeddingHashTableConfig.FLOAT16
    else:
        config.value_type = embedding_hash_table_pb2.EmbeddingHashTableConfig.FLOAT32
    return config

通过使用Float16数据类型，在保证模型精度损失可控的前提下，将GPU内存占用减少了约50%。同时，启用GPU嵌入（enable_gpu_emb）功能，将嵌入层计算迁移到GPU上执行，减少了CPU-GPU数据传输开销，提升了计算效率。

分布式训练中的GPU负载均衡

在分布式训练场景下，GPU负载均衡是提升整体资源利用率的关键。Monolith通过分区哈希表（PartitionedHashTable）实现了训练任务的均匀分布。

分区哈希表源码

def __init__(self,
            num_ps: int,
            table_factory: TableFactory,
            use_native_multi_hash_table: bool,
            max_rpc_deadline_millis: int = 30,
            queue_configs: Dict[str, int] = None,
            parser_ctx=None):
    self.num_ps = num_ps
    self.table_factory = table_factory
    self.use_native_multi_hash_table = use_native_multi_hash_table
    self.max_rpc_deadline_millis = max_rpc_deadline_millis
    self.queue_configs = queue_configs or {}
    self.parser_ctx = parser_ctx
    self.slot_mapping_ = {}
    self.ps_tables = []
    # 初始化PS表
    for i in range(num_ps):
        self.ps_tables.append(table_factory(i))

通过将哈希表数据均匀分布到多个参数服务器（PS）上，实现了训练任务的并行处理和负载均衡。结合GPU设备的智能调度，Monolith能够根据每个PS节点的负载情况动态调整任务分配，确保GPU资源得到充分利用。

动态能效管理系统

基于时段的算力调度

为了实现动态能效管理，Monolith设计了基于时段的算力调度机制。通过分析业务流量模式和算力需求，将训练任务安排在用电低谷时段执行，降低能源成本。

时段调度钩子源码

class TideWaitHook(tf.estimator.SessionRunHook):
    """根据潮汐时段等待的钩子"""
    def __init__(self, hour, minute):
        self.hour = hour
        self.minute = minute
        self.triggered = False

    def before_run(self, run_context):
        if self.triggered:
            return None
        current_time = datetime.datetime.now()
        target_time = current_time.replace(hour=self.hour, minute=self.minute, second=0, microsecond=0)
        if current_time < target_time:
            wait_seconds = (target_time - current_time).total_seconds()
            logging.info(f"Waiting for tide time {self.hour}:{self.minute}, sleep {wait_seconds} seconds")
            time.sleep(wait_seconds)
        self.triggered = True

通过TideWaitHook钩子，训练任务可以在指定的时段开始执行，充分利用电网的谷时电力。这种机制不仅降低了能源成本，也有助于平抑电网峰谷差，提升整体能源利用效率。

动态电压频率调节（DVFS）

在硬件层面，Monolith通过动态电压频率调节（DVFS）技术，根据计算任务的负载情况实时调整GPU的工作频率和电压，在保证性能的同时最大限度地降低能耗。

设备管理源码

def enable_gpu_training():
    """启用GPU训练并配置节能模式"""
    os.environ["MONOLITH_ENABLE_GPU_TRAINING"] = "1"
    # 启用GPU节能模式
    os.environ["NVIDIA_PERSISTED_MODE"] = "1"
    os.environ["NVIDIA_VISIBLE_DEVICES"] = "all"
    # 配置DVFS策略
    set_dvfs_strategy("balanced")

def set_dvfs_strategy(strategy):
    """设置GPU DVFS策略"""
    if strategy == "power-saving":
        # 节能模式：降低GPU频率
        execute_command("nvidia-smi -ac 2505,875")
    elif strategy == "performance":
        # 性能模式：提高GPU频率
        execute_command("nvidia-smi -ac 5001,1530")
    else:
        # 平衡模式
        execute_command("nvidia-smi -ac 3505,1189")

通过结合软件层面的任务调度和硬件层面的DVFS技术，Monolith实现了能效的动态优化。在实际应用中，这种方法使GPU的能源效率提升了约18%。

能效监控与优化闭环

为了持续优化系统能效，Monolith构建了完善的能效监控与优化闭环。通过收集和分析GPU的功耗、温度、利用率等关键指标，结合推荐系统的性能指标，动态调整系统配置参数，实现能效与性能的最佳平衡。

系统监控源码

def machine_info(mem_limit=None, shared_name=None) -> tf.Tensor:
    """获取机器信息和资源使用情况"""
    def _machine_info_fn(mem_limit):
        import psutil
        mem = psutil.virtual_memory()
        cpu = psutil.cpu_percent()
        gpu_info = []
        # 获取GPU信息
        try:
            result = subprocess.check_output(["nvidia-smi", "--query-gpu=power.draw,temperature.gpu,utilization.gpu", "--format=csv,noheader,nounits"])
            for line in result.decode().splitlines():
                power, temp, util = map(float, line.strip().split(','))
                gpu_info.append({
                    "power_draw": power,
                    "temperature": temp,
                    "utilization": util
                })
        except:
            pass
        return {
            "memory_usage": mem.percent,
            "cpu_usage": cpu,
            "gpu_info": gpu_info,
            "timestamp": time.time()
        }
    # 将机器信息作为Tensor返回，便于TensorBoard可视化
    return tf.py_func(_machine_info_fn, [mem_limit], tf.string, stateful=True, name=shared_name or "machine_info")

def emit_timer(key: str,
              value: tf.Tensor,
              tags: Dict[str, str] = None) -> tf.Operation:
    """发送性能指标到监控系统"""
    tags = tags or {}
    def _emit_timer_fn(key, value, tags):
        # 发送指标到监控系统
        metrics_client.emit_metric(
            name=key,
            value=float(value),
            tags=tags,
            timestamp=time.time()
        )
        return True
    return tf.py_func(_emit_timer_fn, [key, value, tags], tf.bool, stateful=True, name=f"emit_timer_{key}")