使用 Java 的 Spark MLlib 分类输入格式答案

【问题标题】：Spark MLlib classification input format using Java使用 Java 的 Spark MLlib 分类输入格式
【发布时间】：2017-11-15 09:49:23
【问题描述】：

如何将 DTO 列表转换为 Spark ML 输入数据集格式？

我有 DTO：

public class MachineLearningDTO implements Serializable {
    private double label;
    private double[] features;

    public MachineLearningDTO() {
    }

    public MachineLearningDTO(double label, double[] features) {
        this.label = label;
        this.features = features;
    }

    public double getLabel() {
        return label;
    }

    public void setLabel(double label) {
        this.label = label;
    }

    public double[] getFeatures() {
        return features;
    }

    public void setFeatures(double[] features) {
        this.features = features;
    }
}

和代码：

Dataset<MachineLearningDTO> mlInputDataSet = spark.createDataset(mlInputData, Encoders.bean(MachineLearningDTO.class));
LogisticRegression logisticRegression = new LogisticRegression();
LogisticRegressionModel model = logisticRegression.fit(MLUtils.convertMatrixColumnsToML(mlInputDataSet));

执行代码后我得到：

java.lang.IllegalArgumentException：要求失败：列特征必须是 org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 类型但实际上是 ArrayType(DoubleType,false)。

如果将其更改为 org.apache.spark.ml.linalg.VectorUDT 代码：

VectorUDT vectorUDT = new VectorUDT();
vectorUDT.serialize(Vectors.dense(......));

然后我得到：

java.lang.UnsupportedOperationException：无法推断类的类型 org.apache.spark.ml.linalg.VectorUDT，因为它不符合 bean

在 org.apache.spark.sql.catalyst.JavaTypeInference$.org$apache$spark$sql$catalyst$JavaTypeInference$$serializerFor(JavaTypeInference.scala:437)

【问题讨论】：

标签： java apache-spark apache-spark-mllib apache-spark-ml

【解决方案1】：

我已经想通了，以防万一有人也坚持使用它，我编写了简单的转换器并且它可以工作：

private Dataset<Row> convertToMlInputFormat(List< MachineLearningDTO> data) {
    List<Row> rowData = data.stream()
            .map(dto ->
                    RowFactory.create(dto.getLabel() ? 1.0d : 0.0d, Vectors.dense(dto.getFeatures())))
            .collect(Collectors.toList());
    StructType schema = new StructType(new StructField[]{
            new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
            new StructField("features", new VectorUDT(), false, Metadata.empty()),
    });

    return spark.createDataFrame(rowData, schema);
}

【讨论】：