public class JavaDecisionTreeClassificationExample {

  public static void main(String[] args) {

    SparkConf sparkConf = new SparkConf()
        .setAppName("JavaDecisionTreeClassificationExample")
        .setMaster("local");
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);

    // 加载并解析数据文件
    String datapath = "data/mllib/sample_libsvm_data.txt";
    JavaRDD<LabeledPoint> data = MLUtils
        .loadLibSVMFile(jsc.sc(), datapath).toJavaRDD();
    // 将数据分割成训练集和测试集（30%作为测试集）
    JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.7, 0.3});
    JavaRDD<LabeledPoint> trainingData = splits[0];     // 训练数据
    JavaRDD<LabeledPoint> testData = splits[1];         // 测试数据

    // 设置参数
    // 空的categoricalFeaturesInfo表示所有特征都是连续的
    int numClasses = 2;
Map<Integer, Integer> categoricalFeaturesInfo = new HashMap<>();
// 用于决策树节点划分的不纯度度量标准
String impurity = "gini";
// 决策树的最大深度
int maxDepth = 5;
// 用于表示特征的桶的数量
    int maxBins = 32;

    // 训练一个用于分类的决策树模型
    DecisionTreeModel model = DecisionTree
        .trainClassifier(trainingData, numClasses,
      categoricalFeaturesInfo, impurity, maxDepth, maxBins);

    // 在测试实例上评估模型并计算测试误差
    JavaPairRDD<Double, Double> predictionAndLabel =
      testData.mapToPair(p -> 
          new Tuple2<>(model.predict(p.features()), p.label()));
    double testErr =
      predictionAndLabel.filter(pl -> !pl._1().equals(pl._2()))
          .count() / (double) testData.count();

    System.out.println("Test Error: " + testErr);
    System.out.println("Learned classification tree model:\n" 
        + model.toDebugString());

    // 保存和加载模型
    model.save(jsc.sc(), "target/tmp/myDecisionTreeClassificationModel");
    DecisionTreeModel sameModel = DecisionTreeModel
      .load(jsc.sc(), "target/tmp/myDecisionTreeClassificationModel");
  }
}

public class JavaRandomForestClassificationExample {
  public static void main(String[] args) {
    // 创建Spark应用的配置对象，设置应用名称和运行模式（本地模式）
    SparkConf sparkConf = new SparkConf()
        .setAppName("JavaRandomForestClassificationExample")
        .setMaster("local");
    // 创建JavaSparkContext对象，用于与Spark集群进行交互
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);
    
    // 加载并解析数据文件
    String datapath = "data/mllib/sample_libsvm_data.txt"; // 数据文件路径
    JavaRDD<LabeledPoint> data = MLUtils
        .loadLibSVMFile(jsc.sc(), datapath).toJavaRDD(); // 读取数据并转换为JavaRDD
    
    // 将数据随机分割为训练集和测试集（70%用于训练，30%用于测试）
    JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.7, 0.3});
    JavaRDD<LabeledPoint> trainingData = splits[0];     // 训练数据
    JavaRDD<LabeledPoint> testData = splits[1];         // 测试数据

    // 训练一个随机森林分类模型
    // 空的categoricalFeaturesInfo表示所有特征都是连续的
    int numClasses = 2; // 类别数，二分类问题
    // 特征信息，这里为空，表示所有特征都是连续的
    Map<Integer, Integer> categoricalFeaturesInfo = new HashMap<>(); 
    int numTrees = 3; // 树的数量，实际应用中应该更多
    String featureSubsetStrategy = "auto"; // 特征子集策略，让算法自动选择
    String impurity = "gini";     // 用于树节点划分的不纯度度量标准
    int maxDepth = 5;            // 树的最大深度
    int maxBins = 32;             // 用于表示特征的桶的数量
    int seed = 12345;             // 随机种子，用于结果的可重复性

    // 使用训练数据训练随机森林模型
    RandomForestModel model = RandomForest
        .trainClassifier(trainingData, numClasses,
              categoricalFeaturesInfo, numTrees, 
              featureSubsetStrategy, 
              impurity, maxDepth, maxBins,
              seed);
    // 在测试数据上评估模型，并计算测试误差
    // 将测试数据的特征和模型预测的标签组合在一起
    JavaPairRDD<Double, Double> predictionAndLabel = 
      testData.mapToPair(p -> 
          new Tuple2<>(model.predict(p.features()), p.label()));
    // 计算测试误差
    double testErr = 
      predictionAndLabel.filter(pl -> !pl._1().equals(pl._2()))
          .count() / (double) testData.count();
    // 打印测试误差和学习到的随机森林模型的详细信息
    System.out.println("Test Error: " + testErr);
    System.out.println("Learned classification forest model:\n" 
        + model.toDebugString());
    // 将模型保存到指定路径
    model.save(jsc.sc(), "target/tmp/myRandomForestClassificationModel"); 
    // 从指定路径加载模型
    RandomForestModel sameModel = RandomForestModel.load(jsc.sc(), 
      "target/tmp/myRandomForestClassificationModel");
    // 停止JavaSparkContext对象
    jsc.stop();
  }
}

public class JavaKMeansExample {
  public static void main(String[] args) {

    // 创建Spark应用的配置对象，设置应用名称和运行模式（本地模式）
    SparkConf conf = new SparkConf().setAppName("JavaKMeansExample")
        .setMaster("local");
    // 创建JavaSparkContext对象，用于与Spark集群进行交互
    JavaSparkContext jsc = new JavaSparkContext(conf);

    // 加载并解析数据
    // 数据文件路径
    String path = "data/mllib/kmeans_data.txt"; 
    // 读取文本文件并创建一个JavaRDD
    JavaRDD<String> data = jsc.textFile(path); 
    JavaRDD<Vector> parsedData = data.map(s -> {
      String[] sarray = s.split(" "); // 按空格分割每行数据
      // 创建一个用于存储数值的数组
      double[] values = new double[sarray.length]; 
      for (int i = 0; i < sarray.length; i++) {
          // 将字符串转换为双精度浮点数
        values[i] = Double.parseDouble(sarray[i]); 
      }
      return Vectors.dense(values);     // 创建稠密向量
    });
    parsedData.cache();                 // 缓存处理后的数据

    // 使用KMeans算法对数据进行聚类
    int numClusters = 2;             // 聚类数量
    int numIterations = 20;             // 迭代次数
    // 训练KMeans模型
    KMeansModel clusters = KMeans
        .train(parsedData.rdd(), numClusters, numIterations); 

    // 打印聚类中心
    System.out.println("Cluster centers:");
    for (Vector center: clusters.clusterCenters()) {
      System.out.println(" " + center);
    }
    // 计算模型的成本
    double cost = clusters.computeCost(parsedData.rdd());
    System.out.println("Cost: " + cost);

    // 计算Within Set Sum of Squared Errors（WSSSE）来评估聚类效果
    double WSSSE = clusters.computeCost(parsedData.rdd());
    System.out.println("Within Set Sum of Squared Errors = " + WSSSE);

    // 将模型保存到指定路径
    clusters.save(jsc.sc(), 
      "target/org/apache/spark/JavaKMeansExample/KMeansModel"); 
    // 从指定路径加载模型
    KMeansModel sameModel = KMeansModel.load(jsc.sc(), 
      "target/org/apache/spark/JavaKMeansExample/KMeansModel");

    // 停止JavaSparkContext对象
    jsc.stop();
  }
}

public class JavaSimpleFPGrowth {

  public static void main(String[] args) {
    // 创建Spark应用的配置对象，设置应用名称和运行模式（本地模式）
    SparkConf conf = new SparkConf()
            .setAppName("FP-growth Example")    // 设置应用名称
            .setMaster("local");                 // 设置运行模式为本地模式
    // 创建JavaSparkContext对象，用于与Spark集群进行交互
    JavaSparkContext sc = new JavaSparkContext(conf);

    // $example on$
    // 读取数据文件，并创建一个JavaRDD对象
    JavaRDD<String> data = sc.textFile("data/mllib/sample_fpgrowth.txt");

    // 将数据文件中的每一行文本映射为一个由商品组成的列表
    JavaRDD<List<String>> transactions = 
        data.map(line -> Arrays.asList(line.split(" ")));

    // 创建一个FP-growth模型对象，并设置最小支持度为0.2，分区数为10
    FPGrowth fpg = new FPGrowth()
      .setMinSupport(0.2)                 // 设置最小支持度阈值
      .setNumPartitions(10);             // 设置分区数

    // 运行FP-growth算法，生成模型
    FPGrowthModel<String> model = fpg.run(transactions);

    int count = 0;
    // 遍历模型中的频繁项集，并打印每个项集及其支持度
    for (FPGrowth.FreqItemset<String> itemset : 
        model.freqItemsets().toJavaRDD().collect()) {
      count++;
      if (count > 5) {
          break;
      }
      System.out.println("[" + itemset.javaItems() + "], " 
          + itemset.freq());
    }

    // 设置最小置信度为0.8，并生成关联规则
    double minConfidence = 0.8;
    for (AssociationRules.Rule<String> rule
      : model.generateAssociationRules(minConfidence)
          .toJavaRDD().collect()) {
      count++;
      if (count > 5) {
          break;
      }
      // 打印每个关联规则及其置信度
      System.out.println(
        rule.javaAntecedent() + " => " 
            + rule.javaConsequent() + ", " 
            + rule.confidence());
    }

    // 停止JavaSparkContext对象
    sc.stop();
  }
}