Comment obtenir une matrice term-document à partir de plusieurs documents avec Spark?

Question

J'essaie de générer une matrice de document-terme à partir de plusieurs documents. Je pourrais lancer le modèle LDA à partir d'une matrice déjà créée, maintenant j'ai besoin de ce recul. J'ai essayé d'implémenter une simple matrice term-doc, mais maintenant je suis coincé. Ce que je faisais était:

//GETS ALL FILES FROM INPUT PATH 
JavaPairRDD<String, String> doc_words = context.wholeTextFiles(input_path); 

//SPLIT BY " " 
JavaPairRDD<String, String> tokenized = doc_words.flatMapValues(Preprocessing_DocumentTermMatrix.WORDS_EXTRACTOR); 

//SEE METHOD WORDS_MAPPER. 
JavaRDD<Tuple2<Tuple2<String, String>, Integer>> rdd = tokenized.flatMap(WORDS_MAPPER); 


//METHOD WORDS_MAPPER 
public static final FlatMapFunction<Tuple2<String, String>, Tuple2<Tuple2<String, String>, Integer>> WORDS_MAPPER = new FlatMapFunction<Tuple2<String, String>, Tuple2<Tuple2<String, String>, Integer>>() { 

    public Iterable<Tuple2<Tuple2<String, String>, Integer>> call(Tuple2<String, String> stringIntegerTuple2) throws Exception { 
     return Arrays.asList(new Tuple2<Tuple2<String, String>, Integer>(new Tuple2<String,String>(stringIntegerTuple2._1(), stringIntegerTuple2._2()), 1)); 
    } 
}; 

Donc, cette fonction me donne un résultat comme celui-ci:

((DOC_0, TERM0), 1) 
((DOC_0, TERM0), 1) 
((DOC_0, TERM1), 1) 
((DOC_1, TERM0), 1) 
((DOC_1, TERM2), 1) 

Je suppose que cela est Allright, mais maintenant je dois réduire et extraire une sortie comme ceci: Ive a essayé beaucoup de choses et n'a pas pu l'obtenir ... Quelqu'un peut-il m'aider?

Answer 1

Voici la solution:

JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> newrdd = JavaPairRDD.fromJavaRDD(rdd).reduceByKey((a, b) -> a + b) 
       .mapToPair(t -> new Tuple2<>(t._1._1, new Tuple2<>(t._1._2, t._2))).groupByKey();