云服务器
云服务器
文档指南
请输入
- 文档首页
云服务器云服务器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模型镜像。
- 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：配置总请求数。
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加速前后模型的最大吞吐信息。
因此，在追求性价比、丰富实例资源场景下，您可以使用搭载第六代英特尔® 至强® 可扩展处理器（Granite Rapids）的实例规格（例如通用型g4i）代替GPU实例，通过系统级加速方案获取最优性价比的WDL模型推理服务。
最近更新时间：2025.12.08 16:00:12
这个页面对您有帮助吗？
有用
有用
无用
无用