java案例如何利用历史大数据建模预测？

wen java案例 2026-08-06 133

我来给你一个完整的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）	稳定性提升