Apache Spark MLlib - 从数据中获取 LabeledPoint (Java)答案

【问题标题】：Apache Spark MLlib - getting LabeledPoint from data (Java)Apache Spark MLlib - 从数据中获取 LabeledPoint (Java)
【发布时间】：2017-01-24 15:22:40
【问题描述】：

我是 Apache Spark 的新手，正在尝试将 .csv 文件中的数据转换为 LabeledPoint 以使用 Apache Spark 的 MLlib 包。我尝试使用以下代码获取 LabeledPoint Data RDD，但事实证明它是 ML Package 的 LabeledPoint Data。现在我想创建正确的 MLlib 包的 LabeledPoint 数据。谁能帮忙。

private static String appName = "learning_RDD";
private static String master = "spark://23.195.26.187:7077" ;
static SparkConf sparkConf = new SparkConf().setMaster("local[1]").setAppName("MLPipelineSample").set("spark.driver.memory", "512m").set("spark.sql.warehouse.dir","D:\\input.txt");
static SparkContext sc = new SparkContext(sparkConf);
static SparkSession spark = SparkSession
          .builder().sparkContext(sc)
          .getOrCreate();

public static void main(String args[]) throws IOException {

    Dataset<Row> trainingData = spark.read().format("com.databricks.spark.csv").option("header", "true").option("inferSchema", "true").load("D:\\abc\\Spark\\WebcontentClassification_UsingSparkML\\WebcontentClassification_UsingSparkML\\NaiveBayes_ML_20ErrorRate\\nutchcsvalldata.csv");

    Tokenizer tokenizer = new Tokenizer().setInputCol("content").setOutputCol("words");
    Dataset<Row> words = tokenizer.transform(trainingData);

    StopWordsRemover remover = new StopWordsRemover().setInputCol("words").setOutputCol("filteredwords");
    Dataset<Row> filteredwords = remover.transform(words);

    HashingTF hashingTF = new HashingTF().setNumFeatures(1000).setInputCol("filteredwords").setOutputCol("rawfeatures");
    Dataset<Row> hashedtf_Vector = hashingTF.transform(filteredwords);

    IDF idf = new IDF().setInputCol("rawfeatures").setOutputCol("features");
    IDFModel idfModel = idf.fit(hashedtf_Vector);
    Dataset<Row> Vectors = idfModel.transform(hashedtf_Vector);             

    Iterator<Row> iterator = Vectors.toLocalIterator();
    List<LabeledPointLabeledPoint> labeledpoints = new ArrayList<LabeledPoint>();
    while(iterator.hasNext())
    {
        Row r = iterator.next();
        int label = r.getAs(2);
        Vector v = r.getAs(16);
        LabeledPoint labeledpoint = new LabeledPoint(label, v);
        labeledpoints.add(labeledpoints);
    }

        // Here I am suppose convert the List into RDD<LabeledPoint> and use                           SVM Algorithm
}

【问题讨论】：

你应该使用RDD 的LabeledPoint，而不是List
您似乎也遵循了一些示例，该示例试图教您有关管道的知识（tokenizer、stopWordRemover、HashingTF、IDF）。我的假设是否正确？
是的，你是对的，我打算从 List 中创建一个 RDD（如果你阅读了最后一行代码中的注释）

标签： java apache-spark

【解决方案1】：

我已经找到了一个解决方案（我没有发布确切的解决方案），首先将单词转换为向量，可以使用以下代码，稍后可以将其制作为 LabeledPoint

    JavaRDD<String> lines = spark.read().textFile(Input_file_path).toJavaRDD();
    JavaRDD<Iterable<String>> words_iterable = lines.map(new Function<String, Iterable<String>>() {
        public Iterable<String> call(String s) throws Exception {
            String[] words = s.split(" ");
            Iterable<String> output = Arrays.asList(words);
            return output;
        }
    });
    Word2Vec word2vec = new Word2Vec(); 
    Word2VecModel word2vecmodel =  word2vec.fit(words_iterable);

【讨论】：

【解决方案2】：

我想创建正确的 MLlib 包的 LabeledPoint 数据

import org.apache.spark.mllib.regression.LabeledPoint;

我想将 List 转换为 RDD 并使用

我认为您需要 map 超过 Vectors 并将 RDD 转换为您需要的格式。

我一直在使用 Scala，但它可能大致像这样翻译

 RDD<LabeledPoint> training = Vectors.map(r -> {
    double label = (double) r.getAs(2); // labels should be doubles
    Vector v = r.getAs(16); // maybe convert this to a dense / sparse array
    return new LabeledPoint(label, v);
  });

【讨论】：

【解决方案3】：

假设您的 csv 文件的每一行中有四个字段，其中第一个字段是您的标签，其余三个字段是您的特征（假设所有字段都是双精度值）。您可以按如下方式创建 LabeledPoint RDD：

JavaSparkContext sc = new JavaSparkContext(sparkConf);
        String path = "com.databricks.spark.csv";
        JavaRDD<String> data = sc.textFile(path);
        JavaRDD<LabeledPoint> parsedData = data
                .map(new Function<String, LabeledPoint>() {
                    public LabeledPoint call(String line) throws Exception {
                        String[] parts = line.split(",");
                        return new LabeledPoint(Double.parseDouble(parts[0]),
                                Vectors.dense(Double.parseDouble(parts[1]),
                                        Double.parseDouble(parts[2]),
                                        Double.parseDouble(parts[3])));
                    }
                });

【讨论】：