当前位置：首页 > news >正文

基于LSTM的时间序列服务器负载预测：从数据预处理到模型部署的完整实战

news 2026/8/13 12:27:11

1. 引言与背景

在云计算和微服务架构日益普及的今天，服务器负载预测已成为保障服务质量（QoS）、实现弹性伸缩和节约运维成本的核心技术。传统的阈值告警和基于规则的伸缩策略往往存在滞后性，而时间序列预测方法能够提前感知负载趋势，为主动运维提供决策依据。

近年来，长短期记忆网络（LSTM）因其在处理长序列依赖和克服梯度消失方面的卓越能力，已成为时间序列预测的首选深度学习模型之一。与传统的ARIMA、指数平滑等统计方法相比，LSTM能够自动捕捉非线性特征和多变量之间的复杂交互，尤其适用于互联网流量、CPU使用率、内存占用等具有周期性和突发性的服务器指标。

本文将从数据采集→数据清洗→特征工程→模型构建→训练调优→评估可视化→部署推理的全流程出发，使用真实的服务器监控数据集（模拟），手把手构建一套完整的LSTM负载预测系统。全文代码均基于Python 3.10+、TensorFlow 2.15+和PySpark 3.5+（用于大数据预处理），并涵盖分布式数据处理、超参数搜索、模型量化部署等前沿实践。

1. 引言与背景

2. 环境准备与数据说明

2.1 实验环境

2.2 数据集说明

3. 大数据预处理（PySpark + Pandas 混合架构）

3.1 数据加载与初步探查

3.2 缺失值处理与异常检测

4. 特征工程与时间序列构造

4.1 构造时序特征

4.2 多变量归一化

4.3 构建监督学习样本（滑动窗口）

5. LSTM模型设计与超参数调优

5.1 基准模型架构

5.2 超参数优化（Optuna + Keras Tuner）

6. 模型评估与可视化分析

6.1 测试集评估指标

6.2 预测曲线与残差分析

6.3 周期性趋势捕捉

7. 模型优化：注意力机制与Transformer对比

7.1 带注意力机制的LSTM

7.2 简易Transformer（仅编码器）

7.3 三种模型对比

8. 模型部署与实时推理（TensorFlow Serving + 量化）

8.1 模型保存为SavedModel格式

8.2 动态量化以加速推理

8.3 使用Flask构建实时预测API

9. 分布式训练与大数据扩展（可选）

10. 总结与生产落地建议

2. 环境准备与数据说明

2.1 实验环境

本实验采用以下软硬件配置：

操作系统：Ubuntu 22.04 LTS
CPU：Intel Xeon Gold 6248（分配8核）
内存：32GB DDR4
GPU：NVIDIA A10（24GB显存，用于加速训练）
Python版本：3.10.12
核心库：
- tensorflow==2.15.0
- pyspark==3.5.0
- pandas==2.1.4
- numpy==1.26.2
- matplotlib==3.8.2
- seaborn==0.13.0
- scikit-learn==1.3.2
- optuna==3.5.0（超参数优化）
- tensorflow-model-optimization==0.7.4（模型压缩）

2.2 数据集说明

由于真实服务器监控数据涉及隐私，本文使用公开的ClusterTrace 2019（谷歌集群轨迹数据集）的采样版本，并模拟生成30天的分钟级负载数据（共43,200个时间点）。包含以下字段：

字段名	类型	描述
timestamp	datetime	时间戳（分钟粒度）
cpu_usage	float	CPU使用率（0~100）
mem_usage	float	内存使用率（0~100）
disk_io	float	磁盘读写速率（MB/s）
net_in	float	网络入流量（Mbps）
net_out	float	网络出流量（Mbps）
active_connections	int	活跃连接数
load_avg_5m	float	5分钟平均负载

我们的目标是基于过去24小时（1440分钟）的历史数据，预测未来1小时（60分钟）的CPU使用率序列。

3. 大数据预处理（PySpark + Pandas 混合架构）

面对43,200条记录（若扩展至数月则达百万级），我们利用PySpark进行分布式缺失值填充、异常检测和重采样，再利用Pandas进行特征工程。

3.1 数据加载与初步探查

python

from pyspark.sql import SparkSession from pyspark.sql.functions import col, when, avg, stddev, round from pyspark.sql.types import StructType, StructField, DoubleType, StringType, TimestampType import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns # 初始化Spark Session (配置内存和并行度) spark = SparkSession.builder \ .appName("ServerLoadPrediction") \ .config("spark.executor.memory", "8g") \ .config("spark.driver.memory", "4g") \ .config("spark.sql.shuffle.partitions", "200") \ .getOrCreate() # 定义Schema (加速读取) schema = StructType([ StructField("timestamp", StringType(), True), StructField("cpu_usage", DoubleType(), True), StructField("mem_usage", DoubleType(), True), StructField("disk_io", DoubleType(), True), StructField("net_in", DoubleType(), True), StructField("net_out", DoubleType(), True), StructField("active_connections", DoubleType(), True), StructField("load_avg_5m", DoubleType(), True) ]) # 读取CSV (模拟从HDFS或S3加载) df_spark = spark.read.schema(schema).option("header", "true").csv("s3://bucket/server_load_30d.csv") # 转换为Spark时间戳类型 from pyspark.sql.functions import to_timestamp df_spark = df_spark.withColumn("timestamp", to_timestamp(col("timestamp"), "yyyy-MM-dd HH:mm:ss")) # 查看数据概况 print(f"总记录数: {df_spark.count()}") df_spark.printSchema() df_spark.describe().show()

3.2 缺失值处理与异常检测

服务器监控常因网络抖动导致数据缺失。我们采用线性插值填充缺失，并用3σ原则检测异常值，将其替换为前后2小时的移动中位数。

python

from pyspark.sql.functions import lag, lead, when, abs from pyspark.sql.window import Window # 1. 缺失值检查 null_counts = df_spark.select([col(c).isNull().cast("int").alias(c) for c in df_spark.columns]).sum() print("各列缺失数:", null_counts.collect()[0].asDict()) # 2. 使用Spark的窗口函数进行线性插值 (简化版: 用前向+后向填充) window_spec = Window.orderBy("timestamp") df_filled = df_spark.withColumn("cpu_usage_fill", when(col("cpu_usage").isNull(), (lag("cpu_usage", 1).over(window_spec) + lead("cpu_usage", 1).over(window_spec)) / 2 ).otherwise(col("cpu_usage"))) # 对其他列类似处理 (此处省略重复代码，实际可封装函数) # 为简洁，使用na.fill的均值填充 (但线性插值更精确，此处仅做演示) df_filled = df_filled.fillna({"cpu_usage": 0, "mem_usage": 0, "disk_io": 0, "net_in": 0, "net_out": 0, "active_connections": 0, "load_avg_5m": 0}) # 3. 异常值检测 (基于3σ，按列计算) stats = df_filled.select([avg(c).alias(f"avg_{c}"), stddev(c).alias(f"std_{c}") for c in df_filled.columns if c != "timestamp"]) stats_collect = stats.collect()[0].asDict() # 替换异常值为中位数 (为减少shuffle，转换为Pandas后处理，但大数据场景建议用UDF) # 此处采用Pandas UDF方式 (Grouped Map) from pyspark.sql.functions import pandas_udf, PandasUDFType from pyspark.sql.types import StructType, StructField, DoubleType @pandas_udf(returnType=df_filled.schema, functionType=PandasUDFType.GROUPED_MAP) def outlier_clean_udf(pdf): # 对每一列计算z-score，超过3则替换为该列中位数 for col_name in pdf.columns: if col_name != "timestamp": median_val = pdf[col_name].median() std_val = pdf[col_name].std() if std_val > 1e-6: pdf[col_name] = pdf[col_name].mask((pdf[col_name] - pdf[col_name].mean()).abs() > 3 * std_val, median_val) else: pdf[col_name] = pdf[col_name].mask((pdf[col_name] - pdf[col_name].mean()).abs() > 3 * std_val, median_val) return pdf # 对全量数据应用 (注意: 此操作会触发shuffle，实际可分批) df_cleaned = df_filled.groupby("timestamp").apply(outlier_clean_udf) # 按时间分组无意义，需传入所有行 # 更合理做法：使用toPandas()后处理，但数据量大会OOM，此处我们直接toPandas()因为仅有4万条 df_pandas = df_filled.toPandas() df_pandas['timestamp'] = pd.to_datetime(df_pandas['timestamp']) # 再次检查异常值 (箱线图) plt.figure(figsize=(12,6)) sns.boxplot(data=df_pandas[['cpu_usage','mem_usage','disk_io','net_in','net_out']]) plt.title("Boxplot after outlier cleaning") plt.show()

4. 特征工程与时间序列构造

4.1 构造时序特征

为了让LSTM学到周期性，我们构造以下特征：

小时编码 (sin/cos)
星期几编码
滞后特征 (过去1h、2h、6h、12h的均值)
滚动统计 (5min、30min、60min的滑动平均和标准差)

python

# 确保按时间排序 df_pandas = df_pandas.sort_values('timestamp').reset_index(drop=True) # 时间特征 df_pandas['hour'] = df_pandas['timestamp'].dt.hour df_pandas['day_of_week'] = df_pandas['timestamp'].dt.dayofweek # 0=Mon df_pandas['hour_sin'] = np.sin(2 * np.pi * df_pandas['hour'] / 24) df_pandas['hour_cos'] = np.cos(2 * np.pi * df_pandas['hour'] / 24) df_pandas['day_sin'] = np.sin(2 * np.pi * df_pandas['day_of_week'] / 7) df_pandas['day_cos'] = np.cos(2 * np.pi * df_pandas['day_of_week'] / 7) # 滞后特征 (使用shift) for lag in [60, 120, 360, 720]: # 1h, 2h, 6h, 12h df_pandas[f'cpu_lag_{lag}'] = df_pandas['cpu_usage'].shift(lag) df_pandas[f'mem_lag_{lag}'] = df_pandas['mem_usage'].shift(lag) df_pandas[f'conn_lag_{lag}'] = df_pandas['active_connections'].shift(lag) # 滚动统计 (窗口需至少60分钟) for window in [5, 30, 60]: df_pandas[f'cpu_roll_mean_{window}'] = df_pandas['cpu_usage'].rolling(window=window, min_periods=1).mean() df_pandas[f'cpu_roll_std_{window}'] = df_pandas['cpu_usage'].rolling(window=window, min_periods=1).std().fillna(0) df_pandas[f'conn_roll_mean_{window}'] = df_pandas['active_connections'].rolling(window=window, min_periods=1).mean() # 删除前720行的NaN (因为滞后720) df_pandas = df_pandas.dropna().reset_index(drop=True) print(f"特征工程后样本数: {len(df_pandas)}")

4.2 多变量归一化

LSTM对输入尺度敏感，我们使用MinMaxScaler将所有特征缩放到[0,1]。

python

from sklearn.preprocessing import MinMaxScaler # 分离特征列和目标列 feature_cols = [c for c in df_pandas.columns if c not in ['timestamp', 'cpu_usage']] # 目标为cpu_usage target_col = 'cpu_usage' scaler_X = MinMaxScaler() scaler_y = MinMaxScaler() X_scaled = scaler_X.fit_transform(df_pandas[feature_cols]) y_scaled = scaler_y.fit_transform(df_pandas[[target_col]]) print(f"特征维度: {X_scaled.shape}, 目标维度: {y_scaled.shape}")

4.3 构建监督学习样本（滑动窗口）

设定lookback=1440（24小时），horizon=60（预测未来60分钟）。我们采用多步预测策略：输出60个时间点的序列。

python

def create_sequences(X, y, lookback=1440, horizon=60): X_seq, y_seq = [], [] for i in range(lookback, len(X) - horizon + 1): X_seq.append(X[i-lookback:i]) y_seq.append(y[i:i+horizon].flatten()) # 展平为(60,) return np.array(X_seq, dtype=np.float32), np.array(y_seq, dtype=np.float32) X_seq, y_seq = create_sequences(X_scaled, y_scaled, lookback=1440, horizon=60) print(f"样本形状: X={X_seq.shape}, y={y_seq.shape}") # (样本数, 1440, 特征数), (样本数, 60) # 划分训练/验证/测试 (6:2:2) total = X_seq.shape[0] train_end = int(0.6 * total) val_end = int(0.8 * total) X_train, X_val, X_test = X_seq[:train_end], X_seq[train_end:val_end], X_seq[val_end:] y_train, y_val, y_test = y_seq[:train_end], y_seq[train_end:val_end], y_seq[val_end:] print(f"训练集: {X_train.shape}, 验证集: {X_val.shape}, 测试集: {X_test.shape}")

5. LSTM模型设计与超参数调优

5.1 基准模型架构

我们采用双层堆叠LSTM + Dropout + Dense结构，并加入残差连接以缓解梯度退化。

python

import tensorflow as tf from tensorflow.keras.models import Model from tensorflow.keras.layers import Input, LSTM, Dropout, Dense, LayerNormalization, Add from tensorflow.keras.regularizers import l2 from tensorflow.keras.optimizers import Adam def build_lstm_model(input_shape, output_dim=60, lstm_units=[128, 64], dropout_rate=0.3, l2_reg=1e-4): inputs = Input(shape=input_shape) # 第一层LSTM (返回序列) x = LSTM(lstm_units[0], return_sequences=True, kernel_regularizer=l2(l2_reg))(inputs) x = LayerNormalization()(x) x = Dropout(dropout_rate)(x) # 第二层LSTM x = LSTM(lstm_units[1], return_sequences=False, kernel_regularizer=l2(l2_reg))(x) x = LayerNormalization()(x) x = Dropout(dropout_rate)(x) # 残差连接: 从第二层输出直接到全连接 (可选) # 加入Dense层 x = Dense(128, activation='relu', kernel_regularizer=l2(l2_reg))(x) x = Dropout(dropout_rate)(x) outputs = Dense(output_dim, activation='linear')(x) model = Model(inputs, outputs) return model # 实例化 input_shape = (1440, X_train.shape[2]) model = build_lstm_model(input_shape) model.compile(optimizer=Adam(learning_rate=0.001), loss='mse', metrics=['mae']) model.summary()

5.2 超参数优化（Optuna + Keras Tuner）

为了找到最佳的超参数组合，我们使用Optuna进行20次试验。

python

import optuna from tensorflow.keras.callbacks import EarlyStopping def objective(trial): # 建议超参数 lstm_units1 = trial.suggest_int('lstm_units1', 64, 256, step=32) lstm_units2 = trial.suggest_int('lstm_units2', 32, 128, step=32) dropout = trial.suggest_float('dropout', 0.1, 0.5) l2_reg = trial.suggest_float('l2_reg', 1e-5, 1e-3, log=True) lr = trial.suggest_float('lr', 1e-4, 1e-2, log=True) model = build_lstm_model(input_shape, lstm_units=[lstm_units1, lstm_units2], dropout_rate=dropout, l2_reg=l2_reg) model.compile(optimizer=Adam(learning_rate=lr), loss='mse', metrics=['mae']) early_stop = EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True) history = model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=30, batch_size=64, callbacks=[early_stop], verbose=0) # 返回验证集最小MAE val_mae = min(history.history['val_mae']) return val_mae # 创建study并优化 study = optuna.create_study(direction='minimize', sampler=optuna.samplers.TPESampler(seed=42)) study.optimize(objective, n_trials=20, timeout=3600) # 1小时 print("最佳参数:", study.best_params) print("最佳验证MAE:", study.best_value) # 使用最佳参数重新训练模型 best_params = study.best_params best_model = build_lstm_model(input_shape, lstm_units=[best_params['lstm_units1'], best_params['lstm_units2']], dropout_rate=best_params['dropout'], l2_reg=best_params['l2_reg']) best_model.compile(optimizer=Adam(learning_rate=best_params['lr']), loss='mse', metrics=['mae']) # 增加epochs至50，并加入ModelCheckpoint from tensorflow.keras.callbacks import ModelCheckpoint, ReduceLROnPlateau checkpoint = ModelCheckpoint('best_lstm.h5', monitor='val_loss', save_best_only=True) reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=3, min_lr=1e-6) history = best_model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=50, batch_size=64, callbacks=[checkpoint, reduce_lr, early_stop], verbose=1)

6. 模型评估与可视化分析

6.1 测试集评估指标

我们使用MAE, RMSE, MAPE以及R²对模型进行全面评估。

python

from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score # 加载最佳模型 best_model.load_weights('best_lstm.h5') # 预测测试集 y_pred_scaled = best_model.predict(X_test) # 反归一化 y_pred = scaler_y.inverse_transform(y_pred_scaled) y_true = scaler_y.inverse_transform(y_test) # 计算全局指标 mae = mean_absolute_error(y_true.flatten(), y_pred.flatten()) rmse = np.sqrt(mean_squared_error(y_true.flatten(), y_pred.flatten())) r2 = r2_score(y_true.flatten(), y_pred.flatten()) # MAPE (避免除零) mape = np.mean(np.abs((y_true - y_pred) / (y_true + 1e-6))) * 100 print(f"=== 测试集整体性能 ===") print(f"MAE: {mae:.4f} %") print(f"RMSE: {rmse:.4f} %") print(f"R²: {r2:.4f}") print(f"MAPE: {mape:.2f} %")

预期输出（基于良好训练）：

text

MAE: 2.3541 % RMSE: 3.8872 % R²: 0.9375 MAPE: 4.21 %

6.2 预测曲线与残差分析

我们随机抽取测试集中一个样本，绘制未来60分钟的预测对比图。

python

import matplotlib.pyplot as plt import numpy as np # 随机选择一个测试样本 idx = np.random.randint(0, X_test.shape[0]) sample_pred = y_pred[idx] sample_true = y_true[idx] plt.figure(figsize=(14,6)) plt.plot(sample_true, label='True CPU Usage', linewidth=2, color='blue') plt.plot(sample_pred, label='LSTM Predicted', linewidth=2, color='red', linestyle='--') plt.fill_between(range(60), sample_true-2*mae, sample_true+2*mae, alpha=0.2, color='gray') plt.xlabel('Future Minutes (Horizon=60)') plt.ylabel('CPU Usage (%)') plt.title(f'LSTM Prediction vs Ground Truth (Test Sample {idx})') plt.legend() plt.grid(alpha=0.3) plt.show() # 残差分布 residuals = sample_true - sample_pred plt.figure(figsize=(12,4)) plt.subplot(1,2,1) plt.hist(residuals, bins=20, edgecolor='k', alpha=0.7) plt.title('Residual Distribution') plt.xlabel('Residual (%)') plt.ylabel('Frequency') plt.subplot(1,2,2) plt.scatter(sample_true, sample_pred, alpha=0.6) plt.plot([0,100], [0,100], 'r--') plt.xlabel('True CPU (%)') plt.ylabel('Predicted CPU (%)') plt.title('Prediction vs True') plt.show()

6.3 周期性趋势捕捉

为了验证模型是否学到日周期，我们连续预测3天的CPU曲线。

python

# 取测试集连续3天（4320分钟）的数据，使用滑动预测（但这里直接取已有测试集） test_days = 3 day_samples = 1440 # 每分钟一个点，一天1440个点 test_start = 0 # 从测试集第一个样本开始 pred_series = [] true_series = [] for i in range(test_days): # 取第i天的样本（假设测试集连续） day_pred = y_pred[test_start + i*day_samples : test_start + (i+1)*day_samples] day_true = y_true[test_start + i*day_samples : test_start + (i+1)*day_samples] pred_series.append(day_pred.flatten()) true_series.append(day_true.flatten()) # 绘制三天对比 plt.figure(figsize=(16,8)) for day in range(test_days): plt.subplot(test_days, 1, day+1) plt.plot(true_series[day], label='True', color='blue', alpha=0.7) plt.plot(pred_series[day], label='Predicted', color='red', linestyle='--', alpha=0.7) plt.title(f'Day {day+1} Prediction') plt.ylabel('CPU %') plt.legend() plt.xlabel('Minutes') plt.tight_layout() plt.show()

7. 模型优化：注意力机制与Transformer对比

虽然LSTM已取得不错效果，但为了探索前沿，我们实验性地加入Bahdanau注意力，并与简单的Transformer进行对比。

7.1 带注意力机制的LSTM

python

from tensorflow.keras.layers import Attention, Concatenate def build_attention_lstm(input_shape, output_dim=60, lstm_units=[128, 64], dropout=0.3): inputs = Input(shape=input_shape) # 编码器 lstm_out = LSTM(lstm_units[0], return_sequences=True)(inputs) lstm_out = Dropout(dropout)(lstm_out) # 注意力层 (使用内置Attention) attention = Attention()([lstm_out, lstm_out]) # 自注意力 # 聚合 pooled = tf.reduce_mean(attention, axis=1) # 全局平均池化 dense = Dense(128, activation='relu')(pooled) dense = Dropout(dropout)(dense) outputs = Dense(output_dim, activation='linear')(dense) model = Model(inputs, outputs) return model attn_model = build_attention_lstm(input_shape) attn_model.compile(optimizer=Adam(1e-3), loss='mse', metrics=['mae']) attn_history = attn_model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=30, batch_size=64, callbacks=[early_stop], verbose=1)

7.2 简易Transformer（仅编码器）

python

from tensorflow.keras.layers import MultiHeadAttention, GlobalAveragePooling1D def build_transformer(input_shape, output_dim=60, head_size=8, num_heads=4, ff_dim=128, dropout=0.2): inputs = Input(shape=input_shape) # 位置编码可省略，使用可学习位置嵌入 (为了简洁，此处不添加) # 多头自注意力 attn = MultiHeadAttention(num_heads=num_heads, key_dim=head_size)(inputs, inputs) attn = Dropout(dropout)(attn) attn = LayerNormalization()(attn + inputs) # 残差 # 前馈网络 ffn = Dense(ff_dim, activation='relu')(attn) ffn = Dense(input_shape[-1])(ffn) ffn = Dropout(dropout)(ffn) ffn = LayerNormalization()(ffn + attn) # 池化 pooled = GlobalAveragePooling1D()(ffn) outputs = Dense(output_dim, activation='linear')(pooled) model = Model(inputs, outputs) return model trans_model = build_transformer(input_shape) trans_model.compile(optimizer=Adam(1e-3), loss='mse', metrics=['mae']) trans_history = trans_model.fit(X_train, y_train, validation_data=(X_val, y_val), epochs=30, batch_size=64, callbacks=[early_stop], verbose=1)

7.3 三种模型对比

python

# 绘制验证损失对比 plt.figure(figsize=(12,5)) plt.plot(history.history['val_loss'], label='LSTM', linewidth=2) plt.plot(attn_history.history['val_loss'], label='LSTM+Attention', linewidth=2) plt.plot(trans_history.history['val_loss'], label='Transformer (Enc)', linewidth=2) plt.xlabel('Epochs') plt.ylabel('Validation MSE') plt.title('Model Comparison on Validation Set') plt.legend() plt.grid(alpha=0.3) plt.show()

通常，带注意力的LSTM会略优于纯LSTM，而Transformer在小数据集上可能过拟合，但若数据量充足，Transformer潜力更大。

8. 模型部署与实时推理（TensorFlow Serving + 量化）

8.1 模型保存为SavedModel格式

python

# 保存最佳LSTM模型 best_model.save('serving_model/lstm_cpu_pred/1', save_format='tf')

8.2 动态量化以加速推理

python

import tensorflow_model_optimization as tfmot # 应用量化感知训练 (QAT) 或 后训练动态量化 converter = tf.lite.TFLiteConverter.from_saved_model('serving_model/lstm_cpu_pred/1') converter.optimizations = [tf.lite.Optimize.DEFAULT] quantized_model = converter.convert() with open('lstm_cpu_quantized.tflite', 'wb') as f: f.write(quantized_model) # 加载量化模型进行推理 (使用TensorFlow Lite解释器) import numpy as np interpreter = tf.lite.Interpreter(model_path='lstm_cpu_quantized.tflite') interpreter.allocate_tensors() input_details = interpreter.get_input_details() output_details = interpreter.get_output_details() # 测试单样本推理 sample_input = X_test[0:1].astype(np.float32) interpreter.set_tensor(input_details[0]['index'], sample_input) interpreter.invoke() output = interpreter.get_tensor(output_details[0]['index']) print("量化模型推理结果:", output.shape)

8.3 使用Flask构建实时预测API

python

from flask import Flask, request, jsonify import numpy as np import tensorflow as tf app = Flask(__name__) # 加载量化模型 (或SavedModel) model = tf.keras.models.load_model('serving_model/lstm_cpu_pred/1') @app.route('/predict', methods=['POST']) def predict(): data = request.json # 期望输入: {"features": [[...]]} 形状为 (1,1440,features) features = np.array(data['features'], dtype=np.float32) pred = model.predict(features) # 反归一化 (需传递scaler_y) pred_cpu = scaler_y.inverse_transform(pred) return jsonify({'predicted_cpu': pred_cpu.tolist()}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

9. 分布式训练与大数据扩展（可选）

当数据量达到千万级时，单机训练受限。我们可以使用TensorFlow分布式策略或Horovod进行多GPU训练。以下为MirroredStrategy示例：

python

strategy = tf.distribute.MirroredStrategy() with strategy.scope(): model = build_lstm_model(input_shape) model.compile(optimizer=Adam(1e-3), loss='mse') # 使用tf.data.Dataset from generator 加载数据 train_dataset = tf.data.Dataset.from_tensor_slices((X_train, y_train)).batch(256).prefetch(tf.data.AUTOTUNE) model.fit(train_dataset, epochs=30, validation_data=(X_val, y_val))

对于PB级数据，可先用Spark提取特征并存储为TFRecord，再使用tf.data并行读取。