Wie verwendet man Kombinationen in Spark?

Question

Ich möchte die Methodenkombinationen kennen.

Eigentlich habe ich

Input:

sQmqgE sQmqgE sQmqgE sQmqgE JKOPDB JKOPDB JKOPDB YGJs YGJs YGJs ERLnA ERLnA ERLnA ERLnA ERLnA XPfQ hntJdN hntJdN hntJdN hntJdN hntJdN BPAdv BPAdv BPAdv BPAdv lTOF lTOF SPrVV SPrVV aavqj 

Mein Code:

test.flatMap{line => line.split(" ")} 
      .flatMap{word => word.combinations(2)} 
      .map{paire=>(paire.mkString(","),1)} 
      .reduceByKey(_ + _) 

Ausgang:

(c,k,1981) 
(e,e,1998) 
(U,A,1970) 
(h,f,1947) 
(Y,Q,2133) 
(o,s,2077) 
(F,F,1965) 

Aber ich will die Kombinationen Wörter pro Worte und nicht Brief pro Brief .

-------------- --------- EDIT

Hier können wir sehen, dass meine Minderer gibt immer 1. Zum Beispiel die Paare (abc , mno) nicht 1

val myRDD = sc.parallelize(Seq("abc def ghi mno mno", "jkl mno pqr abc abc")) 
 
myRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[7] at parallelize at <console>:27 
 

 
scala> myRDD.foreach{println} 
 
abc def ghi mno mno 
 
jkl mno pqr abc abc 
 

 
scala> myRDD.map(_.split(" ")).flatMap(_.combinations(2)). 
 
    | map(p=>(p.mkString(","),1)). 
 
    | reduceByKey(_+_). 
 
    | foreach{println} 
 
(jkl,abc,1) 
 
(def,ghi,1) 
 
(jkl,mno,1) 
 
(abc,ghi,1) 
 
(ghi,mno,1) 
 
(mno,pqr,1) 
 
(def,mno,1) 
 
(abc,def,1) 
 
(pqr,abc,1) 
 
(abc,abc,1) 
 
(abc,mno,1) 
 
(mno,abc,1) 
 
(jkl,pqr,1) 
 
(mno,mno,1)

Dank Rückkehr für

Answer 1

hilft Ihnen nur Nimmt man die Kombinationen in jeder Zeile y wollen ou würde so etwas tun:

scala> val myRDD = sc.parallelize(Seq("abc def ghi", "jkl mno pqr")) 
myRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[1] at parallelize at <console>:24 

scala> myRDD.foreach{println} 
abc def ghi 
jkl mno pqr 

scala> myRDD.map(_.split(" ")).flatMap(_.combinations(2)). 
    | map(p=>(p.mkString(","),1)). 
    | reduceByKey(_+_). 
    | foreach{println} 
(abc,ghi,1) 
(mno,pqr,1) 
(abc,def,1) 
(jkl,pqr,1) 
(def,ghi,1) 
(jkl,mno,1) 

Das Problem mit Ihrem Code ist, dass Sie ein flatMap in der ersten Zeile verwenden, die ein RDD[String] gibt, wo die Zeichenfolge jedes Wort im Dataset ist:

scala> myRDD.flatMap(_.split(" ")) 
res0: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[19] at flatMap at <console>:27 

scala> res0.foreach{println} 
abc 
def 
ghi 
jkl 
mno 
pqr 

während in meiner Version verwende ich ein map, die nicht die geteilten Streicher nicht abflachen und gibt ein RDD[Array[String]]:

scala> myRDD.map(_.split(" ")) 
res1: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[18] at map at <console>:27 

scala> res1.foreach(l => println(l.mkString("::"))) 
abc::def::ghi 
jkl::mno::pqr 

So rufe ich combinations auf einer Liste von Wörtern anstatt eine Liste von Chars (dh eine Zeichenfolge):

scala> List("abc", "def", "ghi").combinations(2).map(_.mkString("-")).toList 
res2: List[String] = List(abc-def, abc-ghi, def-ghi) 

scala> "abc".combinations(2).map(_.mkString("-")).toList 
res3: List[String] = List(a-b, a-c, b-c)