云服务器
云服务器
云服务器
文档指南
请输入
云服务器ECS
产品动态
产品公告
实例停售相关
OpenAPI相关
价格和配额相关
功能下线相关
产品计费
快速入门
用户指南
实例
实例规格清单
选择实例计费方式
登录实例
镜像
公共镜像
自定义镜像
镜像维护周期与EOL应对方案
弹性块存储
快照
网络
监控
运维
云助手
运维编排
管理常用运维任务
补丁管理
最佳实践
AI应用
ArkClaw
OpenClaw
通过应用模板快速部署OpenClaw
通过市场镜像快速部署OpenClaw
手动部署OpenClaw
自助建站
搭建环境
搭建应用
搭建网站
管理Linux实例软件源
性能测试
实例配置
搭建Windows操作系统AD域服务
Windows多用户登录
实例数据
监控
镜像
Linux内核调优
故障处理
远程登录问题
Linux操作系统问题
Windows操作系统问题
操作系统问题
veLinux操作系统问题
Linux操作系统问题
Windows操作系统问题
制作自定义镜像问题
实例无法启动问题
实例负载高问题
网络连接问题
存储类问题
云盘问题
第三方软件问题
API参考
实例
镜像
高性能计算集群
密钥对
标签
云盘
预留存储容量包
快照
自动快照策略
监控
部署集
启动模版
云助手
补丁管理
专有宿主机DDH
产品动态
图说专有宿主机
用户指南
API参考
专有宿主机
相关视频
- 文档首页
云服务器云服务器ECS最佳实践自助建站AIGC实践基于g4i实例部署优化后WDL推理模型
复制全文
下载 pdf基于g4i实例部署优化后WDL推理模型
文档反馈
问问助手本文主要介绍如何在ecs.g4i规格的实例中部署、优化WDL模型，通过硬件加速推理过程，提升推理效率。
功能特色
本实践使用了火山引擎第4代云服务器通用型g4i，该规格实例采用火山引擎自研最新DPU架构并搭载了第六代英特尔® 至强® 可扩展处理器（Granite Rapids），结合新一代硬件AI加速指令AMX-BF16与AMX-FP16（Intel Advanced Matrix Extensions）优化，实现了系统级优化方案加速WDL模型推理效率。
背景信息
Wide & Deep Learning（WDL）模型，是一种结合了线性模型（Wide）和深度神经网络（Deep）的混合模型架构，该模型主要用于解决点击率预估（CTR Prediction）问题，广泛应用于推荐、排序搜索、在线广告等领域。由于WDL模型中，Wide部分主要处理高度稀疏的特征，计算量小且非计算密集型，而CPU处理稀疏特征和不规则内存访问模式的能力较强。对于中小规模推理场景，在CPU实例中部署优化后的WDL推理模型较在GPU实例中部署，可以提供了更好的性价比。
- Wide模型：擅长记忆高频特征组合，但无法泛化到未出现过的特征交互。
- Deep模型：通过嵌入向量泛化低频特征，但对高频组合的学习效率低。
操作步骤
- 购买ecs.g4i规格实例。本节采用的实例配置如下：
- 实例规格：ecs.g4i.16xlarge
- 镜像：Ubuntu 22.04 64 bit
- 网络：分配弹性公网IP
- 登录目标实例。
- 执行如下命令，安装依赖。
- apt updateapt install docker.io
- 执行如下命令，获取基于ecs.g4i规格实例优化后的WDL模型镜像。
- 华北2（北京）华东2（上海）华南1（广州）亚太东南（柔佛）wget https://iaas-debug-beijing.tos-cn-beijing.ivolces.com/tool/cpu-serving/saved_model.tarwget https://iaas-debug-shanghai.tos-cn-shanghai.ivolces.com/tool/cpu-serving/saved_model.tarwget https://iaas-debug-guangzhou.tos-cn-guangzhou.ivolces.com/tool/cpu-serving/saved_model.tarwget https://iaas-debug-southeast-1.tos-ap-southeast-1.ivolces.com/tool/cpu-serving/saved_model.tar
- （可选）按需调整模型推理过程中的批处理参数，本文保持默认。
- 执行如下命令，打开batching_config.txt文件。
- 按i键进入编辑模式，按需调整批处理参数配置。
- max_batch_size { value: 4096 } batch_timeout_micros { value: 5000 } num_batch_threads { value: 64 } max_enqueued_batches { value: 204800 }
- 参数说明：
- max_batch_size：单次批处理的最大样本数量，默认为4096。
- batch_timeout_micros：批处理超时时间（微秒），默认为5000。
- num_batch_threads：批处理线程数，默认为64。
- max_enqueued_batches：最大排队批次数，默认为204800。
- 按esc键退出编辑，输入:wq按回车键，保存并退出文件。
- 执行如下命令，运行镜像。
- docker run -t --name server \--rm --network host \-v /data00/saved_model:/models/wdl \-v /data:/data \cr-for-diting-nodelete-cn-beijing.cr.volces.com/practice-images/tensorflow-serving:tf2.15-cpu-fp16 \--model_name=wdl \--model_base_path=/models/wdl \--enable_batching=true \--batching_parameters_file=/models/wdl/batching_config.txt \--grpc_channel_arguments=grpc.max_concurrent_streams=5000000 \--enable_profiler=true \--grpc_max_threads=1000 \--rest_api_num_threads=1000 \--mixed_precision=bfloat16
说明
- 请将/data00/saved_model修改为WDL模型实际存储路径。
- --rest_api_num_threads：用于设置REST API服务的最大线程数。本文配置为1000。
- --mixed_precision：配置混合精度计算类型，通过FP16（float16）或BF16（bfloat16）数据精度时，可启用CPU的AMX（高级矩阵扩展）指令集，提升推理性能。本文配置为bfloat16。
- 进行压力测试。
- 执行如下命令，创建压测脚本。
- vim grpc_client.py
- 按i键进入编辑模式，复制如下内容至脚本文件。
- 脚本内容 *说明 *您可以按需调整如下参数： *request-rate：配置所需QPS。 *max-concurrency：配置最大并发数。 *batch-size：配置每个查询（query）的批处理（batch）数量。 *num-requests：配置总请求数。脚本内容说明您可以按需调整如下参数：
- request-rate：配置所需QPS。
- max-concurrency：配置最大并发数。
- batch-size：配置每个查询（query）的批处理（batch）数量。
- num-requests：配置总请求数。
import argparseimport asyncioimport timeimport numpy as npimport tensorflow as tffrom concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutorimport grpcfrom tensorflow_serving.apis import predict_pb2from tensorflow_serving.apis import prediction_service_pb2_grpcfrom statistics import meanimport multiprocessing# 日志函数，根据debug参数控制是否输出def debug_print(message, debug=False):if debug:print(message)# 生成单个请求的函数（用于并行处理）def generate_single_request(batch_size=1):request = predict_pb2.PredictRequest()request.model_spec.name = 'wdl'request.model_spec.signature_name = "serving_default"# 生成数值特征 I1-I13 (batch维度)for i in range(1, 14):request.inputs[f"I{i}"].CopyFrom(tf.make_tensor_proto(np.random.uniform(0, 1, [batch_size, 1]), dtype=tf.float32))# 生成类别特征 C1-C26 (batch维度)for i in range(1, 27):categories = [f"cat_{i}_{j}" for j in range(5)]batch_values = [np.random.choice(categories) for _ in range(batch_size)]request.inputs[f"C{i}"].CopyFrom(tf.make_tensor_proto(batch_values, dtype=tf.string))return request# 预生成所有请求数据，支持batch_sizedef generate_all_requests(num_requests, batch_size=1, num_workers=None, debug=False):# 默认使用CPU核心数作为工作进程数if num_workers is None:num_workers = min(multiprocessing.cpu_count(), num_requests)debug_print(f"使用 {num_workers} 个进程并行生成请求数据...", debug)# 每个进程生成的请求数requests_per_worker = [num_requests // num_workers] * num_workersfor i in range(num_requests % num_workers):requests_per_worker[i] += 1# 使用进程池并行生成请求with ProcessPoolExecutor(max_workers=num_workers) as executor:futures = []for count in requests_per_worker:# 为每个进程设置不同的随机种子，确保数据独立性seed = np.random.randint(0, 1000000)future = executor.submit(_worker_generate_requests, count, batch_size, seed)futures.append(future)# 收集所有结果requests = []for future in futures:requests.extend(future.result())return requests# 工作进程函数def _worker_generate_requests(num_requests, batch_size, seed):# 设置随机种子，确保各进程生成不同的数据np.random.seed(seed)return [generate_single_request(batch_size) for _ in range(num_requests)]# 异步请求生成器（使用预生成请求列表）async def get_request(requests, request_rate, burstiness, max_batches, batch_counter, counter_lock, start_time, duration, debug=False):assert burstiness > 0, "Burstiness factor must be positive."theta = 1.0 / (request_rate * burstiness)request_idx = 0total_requests = len(requests)while True:# 检查时间限制current_time = time.time()elapsed_time = current_time - start_timeif elapsed_time > duration:debug_print(f"[get_request] 时间限制触发: {elapsed_time:.4f}s > {duration}s", debug)break# 使用锁保护计数器操作async with counter_lock:if request_idx >= total_requests or batch_counter[0] >= max_batches:reason = "请求索引耗尽" if request_idx >= total_requests else "批次计数达到上限"debug_print(f"[get_request] 计数器限制触发: {reason}, 批次计数: {batch_counter[0]}/{max_batches}", debug)break# 获取当前请求并增加计数器current_request = requests[request_idx]request_idx += 1batch_counter[0] += 1yield current_requestif request_rate != float("inf"):interval = np.random.gamma(shape=burstiness, scale=theta)await asyncio.sleep(interval)# 调用gRPC接口def call_grpc_stub(stub, request):start_time = time.time()result = stub.Predict(request, 10.0) # 10秒超时end_time = time.time()return end_time - start_time # 返回延迟时间# 并发监控器class ConcurrencyMonitor:def __init__(self, max_concurrency):self.max_concurrency = max_concurrencyself.current = 0self.lock = asyncio.Lock()self.highest = 0async def increment(self):async with self.lock:self.current += 1if self.current > self.highest:self.highest = self.currentasync def decrement(self):async with self.lock:self.current -= 1async def get_current(self):async with self.lock:return self.currentasync def get_highest(self):async with self.lock:return self.highest# 监控并发使用情况async def concurrency_monitor(monitor, interval=3, debug=False):"""定期打印并发使用情况"""if not debug:return # 如果不启用debug，则直接返回try:while True:current = await monitor.get_current()highest = await monitor.get_highest()print(f"[监控] 当前并发数: {current}/{monitor.max_concurrency}, 历史最高: {highest}")await asyncio.sleep(interval)except asyncio.CancelledError:# 正常取消debug_print("[监控] 监控任务已停止", debug)passexcept Exception as e:debug_print(f"[监控] 监控任务异常: {e}", debug)# 协程池管理器class WorkerPool:def __init__(self, max_workers, stub, request_generator, semaphore, latencies, monitor, loop, start_time, duration, debug=False):self.max_workers = max_workersself.stub = stubself.request_generator = request_generatorself.semaphore = semaphoreself.latencies = latenciesself.monitor = monitorself.loop = loopself.start_time = start_timeself.duration = durationself.debug = debugself.workers = []self.queue = asyncio.Queue()self.running = True# 创建初始工作协程for i in range(min(max_workers, 5)): # 初始启动5个或更少self._create_worker()def _create_worker(self):worker_id = len(self.workers) + 1task = asyncio.create_task(self._worker_loop(worker_id))self.workers.append(task)debug_print(f"[WorkerPool] 创建工作协程 {worker_id}, 总数: {len(self.workers)}", self.debug)async def _worker_loop(self, worker_id):debug_print(f"[worker-{worker_id}] 启动", self.debug)while self.running:try:# 等待请求request = await asyncio.wait_for(self.queue.get(), timeout=1.0)if request is None: # 退出信号break# 检查是否达到最大时长current_time = time.time()if current_time - self.start_time > self.duration:debug_print(f"[worker-{worker_id}] 退出: 时间限制", self.debug)break# 增加并发计数await self.monitor.increment()async with self.semaphore:try:# 设置请求超时时间remaining_time = self.duration - (time.time() - self.start_time)if remaining_time <= 0:debug_print(f"[worker-{worker_id}] 请求剩余时间不足: {remaining_time:.4f}s", self.debug)breaklatency = await asyncio.wait_for(self.loop.run_in_executor(ThreadPoolExecutor(), call_grpc_stub, self.stub, request),timeout=remaining_time)self.latencies.append(latency)debug_print(f"[worker-{worker_id}] 完成请求, 延迟: {latency:.4f}s", self.debug)except asyncio.TimeoutError:debug_print(f"[worker-{worker_id}] 请求超时: 剩余时间 {remaining_time:.4f}s", self.debug)pass # 超时请求不记录except Exception as e:debug_print(f"[worker-{worker_id}] 请求异常: {e}", self.debug)passfinally:# 减少并发计数await self.monitor.decrement()self.queue.task_done()except asyncio.TimeoutError:# 检查是否需要退出if not self.running:breakasync def submit(self, request):# 如果队列长度大于当前工作协程数，且未达到最大协程数，则创建新的工作协程if self.queue.qsize() >= len(self.workers) and len(self.workers) < self.max_workers:self._create_worker()# 将请求放入队列await self.queue.put(request)async def close(self):# 停止接收新请求self.running = False# 向队列中放入退出信号for _ in range(len(self.workers)):await self.queue.put(None)# 等待所有工作协程完成await asyncio.gather(*self.workers, return_exceptions=True)debug_print(f"[WorkerPool] 所有工作协程已停止, 总数: {len(self.workers)}", self.debug)# 异步执行压测async def async_call_grpc(stub, requests, request_rate, burstiness, max_concurrency, duration, max_batches, debug=False):start_time = time.time()latencies = []semaphore = asyncio.Semaphore(max_concurrency)loop = asyncio.get_running_loop()batch_counter = [0] # 使用列表作为可变对象来原子计数批次counter_lock = asyncio.Lock() # 保护计数器的锁stop_event = asyncio.Event() # 用于通知所有协程停止# 创建请求生成器request_generator = get_request(requests, request_rate, burstiness, max_batches, batch_counter, counter_lock, start_time, duration, debug)# 创建并发监控器concurrency_monitor_obj = ConcurrencyMonitor(max_concurrency)# 启动并发监控（如果启用了debug）monitor_task = asyncio.create_task(concurrency_monitor(concurrency_monitor_obj, 3, debug))# 创建协程池worker_pool = WorkerPool(max_workers=max_concurrency,stub=stub,request_generator=request_generator,semaphore=semaphore,latencies=latencies,monitor=concurrency_monitor_obj,loop=loop,start_time=start_time,duration=duration,debug=debug)# 从请求生成器获取请求并提交给协程池async def feeder():try:async for request in request_generator:if time.time() - start_time > duration or stop_event.is_set():breakawait worker_pool.submit(request)except Exception as e:debug_print(f"[feeder] 异常: {e}", debug)finally:# 关闭协程池await worker_pool.close()stop_event.set()# 启动请求供给协程feeder_task = asyncio.create_task(feeder())# 等待所有任务完成try:await asyncio.gather(feeder_task, return_exceptions=True)finally:# 停止监控任务monitor_task.cancel()await monitor_task# 获取最高并发数highest_concurrency = await concurrency_monitor_obj.get_highest()debug_print(f"[async_call_grpc] 最高并发数: {highest_concurrency}/{max_concurrency}", debug)# 过滤有效请求（避免超时导致的无效数据）valid_latencies = [t for t in latencies if t <= duration]return batch_counter[0], valid_latencies, highest_concurrency# 主函数async def main():parser = argparse.ArgumentParser(description='TensorFlow Serving 压测工具')parser.add_argument('--request-rate', type=float, default=10.0, help='目标QPS')parser.add_argument('--burstiness', type=float, default=1.0, help='突发因子')parser.add_argument('--duration', type=float, default=60.0, help='压测时长（秒）')parser.add_argument('--server', type=str, default='localhost:8500', help='服务地址')parser.add_argument('--num-requests', type=int, default=1000, help='总请求批次数量')parser.add_argument('--max-concurrency', type=int, default=50, help='最大并发数')parser.add_argument('--batch-size', type=int, default=1, help='每个请求的样本批量大小')parser.add_argument('--processes', type=int, default=None, help='生成请求的进程数')parser.add_argument('--debug', action='store_true', help='启用调试信息和监控')args = parser.parse_args()print(f"压测配置:")print(f" 请求速率: {args.request_rate} QPS")print(f" 压测时长: {args.duration} 秒")print(f" 最大请求数: {args.num_requests} 批次")print(f" 最大并发数: {args.max_concurrency}")print(f" 批量大小: {args.batch_size}")print(f" 目标服务器: {args.server}")print(f" 调试模式: {'启用' if args.debug else '禁用'}")# 确保预生成足够的请求generate_count = max(args.num_requests, args.max_concurrency * 2)print(f"正在生成 {generate_count} 条请求数据，批量大小为 {args.batch_size}...")start_gen = time.time()requests = generate_all_requests(generate_count, args.batch_size, args.processes, args.debug)print(f"请求数据生成完成，耗时 {time.time()-start_gen:.2f} 秒")# 创建gRPC连接channel = grpc.insecure_channel(args.server)stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)# 执行压测print(f"开始压测，目标QPS={args.request_rate}, 并发数={args.max_concurrency}, 批量大小={args.batch_size}")start_time = time.time()completed_batches, latencies, highest_concurrency = await async_call_grpc(stub, requests, args.request_rate, args.burstiness, args.max_concurrency, args.duration, args.num_requests,args.debug)# 计算性能指标total_time = max(time.time() - start_time, 0.001)throughput = completed_batches / total_timep99_latency = np.percentile(latencies, 99) if latencies else 0avg_latency = np.mean(latencies) if latencies else 0total_samples = completed_batches * args.batch_sizesamples_per_second = total_samples / total_timeprint("\n============ 压测结果 ============")print(f"完成请求数（批次）: {completed_batches} / {args.num_requests}")print(f"完成样本数: {total_samples}")print(f"总耗时: {total_time:.2f} 秒")print(f"请求吞吐量 (QPS): {throughput:.2f}")print(f"样本吞吐量 (样本/秒): {samples_per_second:.2f}")print(f"平均延迟: {avg_latency:.4f} 秒")print(f"P99 延迟: {p99_latency:.4f} 秒")print(f"最大并发数: {args.max_concurrency}")print(f"实际最高并发数: {highest_concurrency}")print(f"批量大小: {args.batch_size}")if __name__ == "__main__":# Windows系统需要这个条件if hasattr(multiprocessing, 'freeze_support'):multiprocessing.freeze_support()asyncio.run(main())
- 按esc键退出编辑模式，输入:wq按回车键，保存并退出文件。
- 执行如下命令，进行压力测试。
- python3 grpc_client.py \--request-rate 80 \--duration 30 \--num-requests 6000 \--max-concurrency 90 \--batch-size 2048 \--server 192.168.0.2:8500
测试结论
本节展示了在ecs.g4i.16xlarge规格实例中部署、运行WDL模型，开启AMX加速前后模型的最大吞吐信息。
批处理线程数 | 数据精度 | |
批处理线程数 | 数据精度 | |
| BF16（启用AMX加速） | FP32（未启用AMX加速） |
2048 | 161.23 | 16.18 |
1024 | 390.91 | 86.43 |
512 | 824.03 | 226.61 |
因此，在追求性价比、丰富实例资源场景下，您可以使用搭载第六代英特尔® 至强® 可扩展处理器（Granite Rapids）的实例规格（例如通用型g4i）代替GPU实例，通过系统级加速方案获取最优性价比的WDL模型推理服务。
最近更新时间：2025.12.08 16:00:12
这个页面对您有帮助吗？
有用
有用
无用
无用