Sparkを使用してIndexedRowMatrixを作成するときに重複する列

Question

複数のドキュメントのペア間の類似性を計算する必要があります。

JavaPairRDD<String,String> files = sc.wholeTextFiles(file_path); 
    System.out.println(files.count()+"**"); 
    JavaRDD<Row> rowRDD = files.map((Tuple2<String, String> t) -> { 
     return RowFactory.create(t._1,t._2.replaceAll("[^\\w\\s]+","").replaceAll("\\d", "")); 
    }); 
    StructType schema = new StructType(new StructField[]{ 
      new StructField("id", DataTypes.StringType, false, Metadata.empty()), 
      new StructField("sentence", DataTypes.StringType, false, Metadata.empty()) 
     }); 
    Dataset<Row> rows = spark.createDataFrame(rowRDD, schema); 


    Tokenizer tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words"); 
    Dataset<Row> tokenized_rows = tokenizer.transform(rows); 


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

    CountVectorizerModel cvModel = new CountVectorizer().setInputCol("filtered_words").setOutputCol("rowF").setVocabSize(100000).fit(filtred_rows); 
    Dataset<Row> verct_rows = cvModel.transform(filtred_rows); 
    IDF idf = new IDF().setInputCol("rowF").setOutputCol("features"); 
    IDFModel idfModel = idf.fit(verct_rows); 
    Dataset<Row> rescaledData = idfModel.transform(verct_rows); 
JavaRDD<IndexedRow> vrdd = rescaledData.toJavaRDD().map((Row r) -> { 
     //DenseVector dense; 
     String s = r.getAs(0); 
     int index = new Integer(s.replace(s.substring(0,24),"").replace(s.substring(s.indexOf(".txt")),"")); 
     SparseVector sparse = (SparseVector) r.getAs(5); 
     //dense = sparse.toDense();parseVector) r.getAs(5); 
     org.apache.spark.mllib.linalg.Vector vec =    org.apache.spark.mllib.linalg.Vectors.dense(sparse.toDense().toArray()); 
     return new IndexedRow(index, vec); 
    }); 


System.out.println(vrdd.count()+"---"); 
    IndexedRowMatrix mat = new IndexedRowMatrix(vrdd.rdd()); 
    System.out.println(mat.numCols()+"---"+mat.numRows()); 

残念ながら、結果はIndexedRowMatrixが4列（最初のものが重複している）私のデータセットが3つの文書が含まれている場合でも使用して作成されていることを示して次のようにそのために、私はprocced。

3** 
3-- 
1106---4 

この重複の原因を検出するのに手伝ってもらえますか？

Answer 1

おそらく重複は全くなく、データは単に連続したゼロベースの整数でなければならない仕様に準拠していない可能性があります。したがって、numRowsはmax(row.index for row in rows) + 1

import org.apache.spark.mllib.linalg.distributed._ 
import org.apache.spark.mllib.linalg.Vectors 

new IndexedRowMatrix(sc.parallelize(Seq(
    IndexedRow(5, Vectors.sparse(5, Array(), Array()))) // Only one non-empty row 
)).numRows 
// res4: Long = 6 

です。