我来给你一个完整的Java大数据预测建模实战案例,我会用时间序列预测和机器学习两个维度来展示。
📊 案例：电商销售预测系统
整体架构设计
graph TD
A[历史数据源] --> B[数据清洗/特征工程]
B --> C[模型训练]
C --> D[模型评估]
D --> E[模型部署]
E --> F[实时预测]
F --> G[预测结果可视化]
核心实现代码
1 数据准备与特征工程
import java.time.LocalDate;
import java.util.*;
import java.util.stream.Collectors;
// 特征工程类
public class FeatureEngineering {
// 数据实体
static class SalesRecord {
LocalDate date;
double sales;
String productId;
String category;
double price;
int quantity;
String promotion; // 是否有促销
int weatherScore; // 天气影响指数
boolean isHoliday; // 是否节假日
// constructor, getters/setters...
}
// 特征提取器
public static List<Map<String, Object>> extractFeatures(List<SalesRecord> records) {
List<Map<String, Object>> features = new ArrayList<>();
for (SalesRecord record : records) {
Map<String, Object> featureMap = new HashMap<>();
// 1. 时间特征
LocalDate date = record.date;
featureMap.put("dayOfWeek", date.getDayOfWeek().getValue());
featureMap.put("month", date.getMonthValue());
featureMap.put("isWeekend", date.getDayOfWeek().getValue() > 5);
featureMap.put("dayOfMonth", date.getDayOfMonth());
// 2. 历史统计特征（滑动窗口）
featureMap.put("ma7", calculateMovingAverage(records, date, 7, record.productId));
featureMap.put("ma30", calculateMovingAverage(records, date, 30, record.productId));
featureMap.put("salesLag1", getPreviousDaysSales(records, date, 1, record.productId));
featureMap.put("salesLag7", getPreviousDaysSales(records, date, 7, record.productId));
// 3. 业务特征
featureMap.put("price", record.price);
featureMap.put("promotion", record.promotion.equals("Y") ? 1 : 0);
featureMap.put("weatherScore", record.weatherScore);
featureMap.put("isHoliday", record.isHoliday ? 1 : 0);
// 4. 分类特征
featureMap.put("category", record.category);
features.add(featureMap);
}
return features;
}
// 计算移动平均
private static double calculateMovingAverage(List<SalesRecord> records,
LocalDate date, int window, String productId) {
List<SalesRecord> pastRecords = records.stream()
.filter(r -> r.productId.equals(productId))
.filter(r -> r.date.isBefore(date))
.limit(window)
.collect(Collectors.toList());
return pastRecords.stream()
.mapToDouble(r -> r.sales)
.average()
.orElse(0);
}
// 获取前一天销量
private static double getPreviousDaysSales(List<SalesRecord> records,
LocalDate date, int lag, String productId) {
return records.stream()
.filter(r -> r.productId.equals(productId))
.filter(r -> r.date.equals(date.minusDays(lag)))
.mapToDouble(r -> r.sales)
.findFirst()
.orElse(0);
}
}
2 模型训练（集成LightGBM）
import com.microsoft.ml.lightgbm.*;
import java.io.*;
// 使用LightGBM进行预测
public class SalesPredictor {
private Booster booster;
// 训练模型
public void trainModel(List<Map<String, Object>> features,
List<Double> labels) throws Exception {
// 转换为LightGBM需要的数据格式
Dataset data = createDataset(features, labels);
// 设置训练参数
Map<String, String> params = new HashMap<>();
params.put("objective", "regression");
params.put("metric", "rmse");
params.put("learning_rate", "0.05");
params.put("num_leaves", "31");
params.put("feature_fraction", "0.8");
params.put("bagging_fraction", "0.8");
params.put("bagging_freq", "5");
params.put("num_iterations", "200");
params.put("early_stopping_round", "50");
// 创建Booster
booster = new Booster(params, data);
// 训练
booster.train();
System.out.println("Model trained successfully!");
}
// 数据转换
private Dataset createDataset(List<Map<String, Object>> features,
List<Double> labels) throws Exception {
// 特征名称映射
String[] featureNames = {
"dayOfWeek", "month", "isWeekend", "dayOfMonth",
"ma7", "ma30", "salesLag1", "salesLag7",
"price", "promotion", "weatherScore", "isHoliday"
};
// 构造浮点特征矩阵
double[][] featureValues = new double[features.size()][featureNames.length];
for (int i = 0; i < features.size(); i++) {
Map<String, Object> feature = features.get(i);
for (int j = 0; j < featureNames.length; j++) {
Object value = feature.get(featureNames[j]);
featureValues[i][j] = value instanceof Number ?
((Number) value).doubleValue() : 0;
}
}
// 创建Dataset
Dataset dataset = new Dataset(featureValues, featureNames.length);
dataset.setLabel(labels.stream().mapToDouble(d -> d).toArray());
return dataset;
}
// 预测单个样本
public double predict(Map<String, Object> features) {
// 转换特征为double数组
int[] colIdx = {0,1,2,3,4,5,6,7,8,9,10,11};
double[] rowValues = new double[colIdx.length];
String[] featureNames = {
"dayOfWeek", "month", "isWeekend", "dayOfMonth",
"ma7", "ma30", "salesLag1", "salesLag7",
"price", "promotion", "weatherScore", "isHoliday"
};
for (int i = 0; i < featureNames.length; i++) {
Object value = features.get(featureNames[i]);
rowValues[i] = value instanceof Number ?
((Number) value).doubleValue() : 0;
}
double[] prediction = booster.predict(rowValues, 1);
return prediction[0];
}
// 批量预测
public List<Double> predictBatch(List<Map<String, Object>> features) {
return features.stream()
.map(this::predict)
.collect(Collectors.toList());
}
// 模型保存
public void saveModel(String filePath) throws IOException {
booster.saveModel(filePath);
}
// 模型加载
public void loadModel(String filePath) throws IOException {
booster = Booster.loadModel(filePath);
}
}
3 预测调度与结果处理
import java.util.concurrent.*;
import java.util.stream.*;
public class PredictionEngine {
private SalesPredictor predictor;
// 每日预测任务
public Map<String, Double> predictDailySales(LocalDate targetDate,
List<SalesRecord> historyData) {
// 生成每日预测特征
List<Map<String, Object>> features = generatePredictionFeatures(targetDate, historyData);
// 执行预测
Map<String, Double> predictions = new HashMap<>();
for (Map.Entry<String, List<Map<String, Object>>> entry :
groupByProduct(features).entrySet()) {
String productId = entry.getKey();
List<Map<String, Object>> productFeatures = entry.getValue();
// 批量预测
List<Double> results = predictor.predictBatch(productFeatures);
predictions.put(productId, results.get(0));
}
return predictions;
}
// 未来7天预测
public Map<LocalDate, Map<String, Double>> predictWeekly(LocalDate startDate,
List<SalesRecord> history) {
Map<LocalDate, Map<String, Double>> weeklyPredictions = new HashMap<>();
for (int i = 0; i < 7; i++) {
LocalDate date = startDate.plusDays(i);
Map<String, Double> dailyPredictions = predictDailySales(date, history);
weeklyPredictions.put(date, dailyPredictions);
// 预测后更新历史数据，用于下一步预测
history = addPredictedRecords(history, date, dailyPredictions);
}
return weeklyPredictions;
}
// 特征生成
private List<Map<String, Object>> generatePredictionFeatures(LocalDate date,
List<SalesRecord> history) {
List<Map<String, Object>> features = new ArrayList<>();
// 获取所有商品
Set<String> products = history.stream()
.map(r -> r.productId)
.collect(Collectors.toSet());
for (String productId : products) {
Map<String, Object> feature = new HashMap<>();
// 基础特征
feature.put("dayOfWeek", date.getDayOfWeek().getValue());
feature.put("month", date.getMonthValue());
feature.put("isWeekend", date.getDayOfWeek().getValue() > 5);
feature.put("dayOfMonth", date.getDayOfMonth());
// 历史统计特征
String finalProductId = productId;
List<SalesRecord> productHistory = history.stream()
.filter(r -> r.productId.equals(finalProductId))
.collect(Collectors.toList());
// 填充历史特征
// ... 类似前面特征工程的方法
features.add(feature);
}
return features;
}
// 按商品分组
private Map<String, List<Map<String, Object>>> groupByProduct(
List<Map<String, Object>> features) {
return features.stream()
.collect(Collectors.groupingBy(f -> f.get("productId").toString()));
}
// 预测结果缓存
private Map<String, Map<LocalDate, Double>> predictionCache = new ConcurrentHashMap<>();
public void cachePrediction(String productId, LocalDate date, double value) {
predictionCache
.computeIfAbsent(productId, k -> new HashMap<>())
.put(date, value);
}
public Double getCachedPrediction(String productId, LocalDate date) {
return predictionCache
.getOrDefault(productId, Collections.emptyMap())
.get(date);
}
}
4 预测效果评估
public class ModelEvaluator {
public static class EvaluationResult {
double rmse;
double mae;
double mape;
double r2;
@Override
public String toString() {
return String.format(
"RMSE: %.2f | MAE: %.2f | MAPE: %.2f%% | R²: %.4f",
rmse, mae, mape, r2);
}
}
// 评估预测效果
public static EvaluationResult evaluate(List<Double> actual, List<Double> predicted) {
EvaluationResult result = new EvaluationResult();
// RMSE
double sumSquaredError = 0;
// MAE
double sumAbsError = 0;
// MAPE
double sumPercentageError = 0;
// R²
double meanActual = actual.stream().mapToDouble(Double::doubleValue).average().orElse(0);
double sumSquaredTotal = 0;
for (int i = 0; i < actual.size(); i++) {
double error = actual.get(i) - predicted.get(i);
sumSquaredError += error * error;
sumAbsError += Math.abs(error);
sumPercentageError += Math.abs(error / actual.get(i)) * 100;
sumSquaredTotal += Math.pow(actual.get(i) - meanActual, 2);
}
int n = actual.size();
result.rmse = Math.sqrt(sumSquaredError / n);
result.mae = sumAbsError / n;
result.mape = sumPercentageError / n;
result.r2 = 1 - (sumSquaredError / sumSquaredTotal);
return result;
}
// K倍交叉验证
public static List<EvaluationResult> crossValidation(List<SalesRecord> data,
int k) {
List<EvaluationResult> results = new ArrayList<>();
Collections.shuffle(data);
int foldSize = data.size() / k;
for (int i = 0; i < k; i++) {
List<SalesRecord> trainData = new ArrayList<>(
data.subList(0, i * foldSize));
trainData.addAll(data.subList((i + 1) * foldSize, data.size()));
List<SalesRecord> testData = new ArrayList<>(
data.subList(i * foldSize, (i + 1) * foldSize));
// 训练模型
// ... 这里省略训练代码
// 预测
List<Double> predictions = new ArrayList<>();
List<Double> actuals = testData.stream()
.map(SalesRecord::getSales)
.collect(Collectors.toList());
// 评估
EvaluationResult evalResult = evaluate(actuals, predictions);
results.add(evalResult);
System.out.println("Fold " + (i + 1) + ": " + evalResult);
}
return results;
}
}
实际应用示例
public class DemandForecastingApp {
public static void main(String[] args) throws Exception {
// 1. 加载历史数据
List<SalesRecord> historicalData = loadHistoricalData("sales_2020_2023.csv");
// 2. 特征工程
List<Map<String, Object>> features = FeatureEngineering.extractFeatures(historicalData);
List<Double> labels = historicalData.stream()
.map(SalesRecord::getSales)
.collect(Collectors.toList());
// 3. 训练模型
SalesPredictor predictor = new SalesPredictor();
predictor.trainModel(features, labels);
// 4. 模型评估
List<EvaluationResult> cvResults = ModelEvaluator.crossValidation(historicalData, 5);
System.out.println("交叉验证结果: " + cvResults);
// 5. 预测未来7天销量
PredictionEngine engine = new PredictionEngine();
LocalDate startDate = LocalDate.now().plusDays(1);
Map<LocalDate, Map<String, Double>> predictions =
engine.predictWeekly(startDate, historicalData);
// 6. 输出预测结果
predictions.forEach((date, productPredictions) -> {
System.out.println("\n📅 日期: " + date);
productPredictions.forEach((product, sales) -> {
System.out.printf(" 商品 %s: %.2f 件%n", product, sales);
});
});
// 7. 保存模型
predictor.saveModel("sales_prediction_model.txt");
}
private static List<SalesRecord> loadHistoricalData(String filePath) {
// 从CSV加载数据
return new ArrayList<>();
}
}
生产环境部署要点
# Docker部署配置
version: '3.8'
services:
prediction-service:
build: .
ports:
- "8081:8080"
environment:
- MODEL_PATH=/models/sales_model.txt
- DATA_RETENTION_DAYS=365
resources:
memory: 4G
cpus: '4'
redis:
image: redis:alpine
ports:
- "6379:6379"
// 实时特征计算微服务
@RestController
@RequestMapping("/api/prediction")
public class PredictionController {
@Autowired
private PredictionEngine engine;
@GetMapping("/daily")
public ApiResponse predictDaily(@RequestParam LocalDate date) {
List<SalesRecord> history = loadRecentHistory();
Map<String, Double> predictions = engine.predictDailySales(date, history);
return ApiResponse.success(predictions);
}
@GetMapping("/weekly")
public ApiResponse predictWeekly(@RequestParam LocalDate startDate) {
List<SalesRecord> history = loadRecentHistory();
Map<LocalDate, Map<String, Double>> predictions =
engine.predictWeekly(startDate, history);
return ApiResponse.success(predictions);
}
}
预测效果优化建议
| 优化方向 | 方法 | 预期效果 |
|---|---|---|
| 特征工程 | 增加节假日、天气、活动等外部特征 | 准确率提升5-10% |
| 模型选择 | 使用AutoML对比算法（XGBoost、Prophet） | 寻找最优模型 |
| 异常处理 | 检测销售异常（促销、断货）并修正 | 减少极端值影响 |
| 增量学习 | 每月增量训练，使用最新数据 | 适应市场变化 |
| 集成学习 | 多个模型融合（LightGBM + Prophet） | 稳定性提升 |
这个案例涵盖了从数据处理、特征工程、模型训练到预测的完整流程，实际应用中可以根据业务需求调整特征和模型参数，关键是特征工程要贴合业务逻辑，这样才能最大化模型预测效果。
