RDD [X：矢量[字符串]以RDD [X：矢量[字符串] +迭代器：矢量[字符串]

Question

DocsRDD：

RDD[Vector[String]] 

DocsRDD：

Vector(Doc1.txt, Doc2.txt, Doc5.txt) 
Vector(Doc4.txt, Doc3.txt) 
Vector(Doc6.txt, Doc9.txt) 

我想要什么只是所有的双对文档的，例如用于DocsRDD我想

AllDualDocsRDD：

Vector(Doc1.txt, Doc2.txt) 
Vector(Doc1.txt, Doc5.txt) 
Vector(Doc2.txt, Doc5.txt) 
Vector(Doc4.txt, Doc3.txt) 
Vector(Doc6.txt, Doc9.txt) 

下面是我的代码示例（我是Spark，Scala的新手）。

val AllDualDocsRDD = DocsRDD.map(e => if (e.size > 2) { 
          val V_iter = (1 to e.size).flatMap(e.combinations).filter(_.size == 2).toVector 
          V_iter.foreach(println) 
          //Here I Cannot put V_iter : scala.Vector[Vector[String]] 
         } 
         else e) 

但是好像我撞到了墙上！有谁知道我能做到这一点？

Answer 1

尝试：

sc.parallelize(
    Seq(Vector("Doc1.txt", "Doc2.txt", "Doc5.txt")) 
).flatMap(v => v.combinations(Math.min(v.size, 2))) 

Answer 2

这个怎么样在RDD直接flatmap：

val AllDualDocsRDD = DocsRDD.flatMap( x => { 
            if(x.size > 2){ 
              x.combinations(2).toSeq 
             } 
            else Seq(x) 
            } 
          ) 

这做这项工作。