java - Sparkを使用して複数のドキュメントから用語とドキュメントのマトリックスを取得するには?

Question

複数のドキュメントから用語とドキュメントのマトリックスを生成しようとしています。作成済みのマトリックスから LDA モデルを実行できましたが、ここでこのステップを戻す必要があります。単純な term-doc マトリックスを実装しようとしましたが、行き詰まりました。私がしたことは：

//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)); 
    } 
};

したがって、この関数は次のような結果をもたらします。

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

これで問題ないと思いますが、今度はそれを減らして、次のような出力を抽出する必要があります。

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

私は多くのことを試しましたが、それを得ることができませんでした...誰かが私を助けることができますか?